"""Explicit business export schema v1; no operational application imports.""" import math import re from datetime import datetime from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit VERSION = 1 SECRET_FIELDS = frozenset({ "access_token", "access_token_encrypted", "private_token", "refresh_token", "token", "oauth_states", "orcid_states", "orcid_verifications", "oauth_info", "auth", "session", "session_secret", "session_key", "cookies", "cookie", "authorization", "password", "client_secret", "api_key", "fernet_key", "github_token", "email_api_key", "SUBMISSION_TOKEN_ENCRYPTION_KEY", "ORCID_CLIENT_SECRET", "HF_TOKEN", "oauth_state", "verification_code", "code_verifier", "csrf_token", }) TRANSIENT_FIELDS = frozenset({"lease_expires_at", "evaluator_dispatch_ready"}) _CREDENTIAL_KEY = re.compile(r"(?i)(?:token|secret|password|credential|signature|api[-_]?key|access[-_]?key|authorization|cookie|session|(?:^|[-_])(?:sig|state|code|auth|key)(?:$|[-_]))") _TOKEN = re.compile(r"\b(?:hf_|gh[pousr]_|github_pat_|sk-|re_)[A-Za-z0-9_-]{20,}|\beyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+") _ASSIGNMENT = re.compile(r'''(?ix)(["']?(?:access[_-]?token|refresh[_-]?token|client[_-]?secret|api[_-]?key|password|authorization|cookie|session[_-]?(?:key|secret)|token|secret|credential)["']?\s*[:=]\s*)(?:\[redacted\]|"[^"\r\n]*"|'[^'\r\n]*'|[^\s,;&}\]]+)''') _URL = re.compile(r"https?://[^\s<>\"']+") class ExportError(ValueError): """Messages contain fixed diagnostic codes only, never source values.""" def fail(code): raise ExportError(code) def sanitize_url(value): try: parsed = urlsplit(value) if parsed.scheme not in {"http", "https"} or not parsed.hostname: fail("invalid_url") hostname = parsed.hostname if ":" in hostname: hostname = f"[{hostname}]" netloc = hostname + (f":{parsed.port}" if parsed.port else "") query = [(key, sanitize_text(item)) for key, item in parse_qsl(parsed.query, keep_blank_values=True) if not _CREDENTIAL_KEY.search(key)] fragment = sanitize_text(parsed.fragment) if "=" in parsed.fragment: fragment = urlencode([(key, sanitize_text(item)) for key, item in parse_qsl(parsed.fragment, keep_blank_values=True) if not _CREDENTIAL_KEY.search(key)]) return urlunsplit((parsed.scheme, netloc, _TOKEN.sub("[redacted]", parsed.path), urlencode(query), fragment)) except (ValueError, TypeError): fail("invalid_url") def sanitize_text(value): value = re.sub(r"(?im)\b(?:authorization|cookie|set-cookie)\s*:\s*[^\r\n]+", lambda match: match.group().split(":", 1)[0] + ": [redacted]", value) value = _URL.sub(lambda match: sanitize_url(match.group()), value) value = _TOKEN.sub("[redacted]", value) value = re.sub(r"(?i)\b(bearer|basic)\s+[^\s,;\"']+", lambda match: match.group(1) + " [redacted]", value) return _ASSIGNMENT.sub(lambda match: match.group(1) + "[redacted]", value) def fields(names, rule="text"): return {name: rule for name in names.split()} PROFILE_FIELDS = fields("name role affiliation email supervisor_name saved_at") EVENT_FIELDS = {**fields("event_id event action actor actor_worker_id actor_hf_username submission_id case case_id", "id"), **fields("timestamp previous_state current_state result final_outcome reason requested_action"), "attempt": "int"} EVENT = ("submission_event", EVENT_FIELDS) PROFILE_REVISION = {**fields("timestamp actor_hf_username researcher_identity_id"), "changes": ("profile_changes",)} RULE_ACCEPTANCE = fields("researcher_identity_id rules_version accepted_at actor_hf_username") PROFILE = {**PROFILE_FIELDS, **fields("hf_username hf_subject hf_sub orcid_id submitter_id researcher_identity_id", "id"), **fields("identity_created_at orcid_connected_at rules_version accepted_at profile_updated_at"), "original_profile": PROFILE_FIELDS, "profile_revisions": ("list", PROFILE_REVISION), "rules_acceptances": ("list", RULE_ACCEPTANCE)} ACCOUNT = fields("username hf_username hf_subject hf_sub id created_at updated_at last_seen_at") ACCOUNT_RULE = ("legacy_account", ACCOUNT) TEAM = {**fields("team_id owner_submitter_id owner_hf_username", "id"), **fields("team_name affiliation created_at"), "members": ("map", "text"), "invitations": ("list", "id")} RESERVATION = fields("orcid_id reserved_at") REGISTRY = {"submitters": ("map", PROFILE), "reservations": ("map", RESERVATION), "hf_accounts": ("map", ACCOUNT_RULE), "teams": ("map", TEAM), "publication_sequences": ("map", ("map", "int"))} CONSENT = fields("policy_version accepted_at") SNAPSHOT = {**fields("hf_username_snapshot orcid display_name task_name track_id user_dataset repo_subdir submitted_at primary_metric rules_version"), "researcher_identity": PROFILE_FIELDS, "paper_url": "url", "code_url": "url", "private_token_provided": "bool", "private_token_consent": CONSENT} METRIC_NAMES = set("pmf psnr ssim lpips rendering_pmf velocity_mae overall_error".split()) METRIC_NAMES.update(f"{metric}_{view}" for metric in ("pmf", "psnr", "ssim", "lpips") for view in ("train", "novel")) METRIC_NAMES.update(f"{view}_{metric}" for view in ("novel", "reinit") for metric in ("pmf", "psnr", "ssim", "lpips")) PROPERTY_METRICS = set("rendering_pmf velocity_mae overall_error materials_count".split()) PROPERTY_METRICS.update(f"{view}_{metric}" for view in ("novel", "reinit") for metric in ("pmf", "psnr", "ssim", "lpips")) for material, parameters in { "newtonian": ("log10_mu", "log10_kappa"), "non_newtonian": ("log10_mu", "log10_kappa", "yield", "log10_eta"), "elastic": ("log10_E", "nu"), "plasticine": ("log10_E", "log10_yield", "nu"), "sand": ("friction",), }.items(): PROPERTY_METRICS.update(f"{material}_{parameter}" for parameter in (*parameters, "error", "v")) PROPERTY_METRICS.add("non_newtonian_log10_yield") METRIC_NAMES.update(PROPERTY_METRICS) METRIC_NAMES.update(f"{metric}_{suffix}" for metric in PROPERTY_METRICS for suffix in ("count", "std")) METRICS = {name: "number" for name in METRIC_NAMES} CONFIG = {**fields("IMAGE_METRIC_FRAME_BATCH_SIZE PROPERTY_ESTIMATION_FRAME_BATCH_SIZE FUTURE_PREDICTION_FRAME_BATCH_SIZE VIDEO_GENERATION_FRAME_BATCH_SIZE IMAGE_METRIC_IMAGE_LOAD_WORKERS", "int"), **fields("IMAGE_METRIC_LPIPS_NET IMAGE_METRIC_IMAGE_LOADER requested_device effective_device pmf_commit"), "packages": {name: "text" for name in ("pmf", "torch", "torchvision", "torchmetrics", "numpy", "opencv-python-headless", "Pillow")}} PROVENANCE = {"schema_version": "int", "input": fields("repository_id repository_type subdirectory revision resolved_at"), "evaluator": {**fields("git_commit git_commit_source source_digest protocol_version manifest_digest metric_configuration_digest"), "manifest_version": "version", "metric_configuration_version": "int", "metric_configuration": CONFIG}} OPERATOR_RESUME_EVENT = {**EVENT_FIELDS, "backup_directory": "text", "previous_evaluator": PROVENANCE["evaluator"], "retained_results_digest": "sha256", "retained_scenes": ("list", "id"), "reset_incomplete_case_starts": ("map", "timestamp"), "verification": {"pmf_tree_at_both_commits": "sha1", "worker_source_change": "text", "full_view_regression": "text"}} LEGACY_OPERATOR_RESUME_EVENT = {**EVENT_FIELDS, "backup_directory": "text", "retained_results_digest": "sha256", "retained_scenes": "int"} OPERATOR_PAUSE_EVENT = {**EVENT_FIELDS, "backup_directory": "text"} PROVENANCE["compatibility_recoveries"] = ("list", ("operator_resume_record",)) SUBMISSION = {**fields("submission_id researcher_identity_id hf_username_snapshot team_id owner owner_username task_name track_id worker_id last_renewed_by source_submission_id submitter_orcid orcid_id", "id"), **fields("display_name team_name user_dataset repo_subdir primary_metric status submitted_at started_at reclaimed_at completed_at updated_at cancel_requested_at error_message quota_consumed_at quota_refunded_at quota_refund_reason evaluation_revision"), **fields("scenes_total scenes_done scenes_failed scenes_count worker_attempt", "int"), "deleted": "bool", "evaluator_config_version": "version", "score": "number", "metrics": METRICS, "creation_snapshot": SNAPSHOT, "submission_events": ("list", EVENT), "evaluation_provenance": PROVENANCE, "scene_failures": ("map", "text")} SUBMISSION["history_written_at"] = "timestamp" SUBMISSION["requeued_at"] = "timestamp" RETRY = {**fields("attempts", "int"), **fields("status error"), "infrastructure": "bool"} PROGRESS = {**fields("submission_id task_name last_worker current_scene", "id"), **fields("current_stage last_updated_at next_retry_at last_retry_at"), **fields("scenes_total current_attempt current_case_index", "int"), "scene_results": ("map", {"metrics": METRICS, "ts": "text"}), "scene_failures": ("map", "text"), "scene_retries": ("map", RETRY), "infrastructure_failures": ("map", "bool"), "case_started_at": ("map", "text"), "submission_events": ("list", EVENT)} REVISION = {**fields("timestamp field previous_value new_value"), **fields("event_id actor_hf_username submission_id", "id")} METADATA = {**fields("submission_id task_name researcher_identity_id", "id"), **fields("display_name metadata_updated_at"), "paper_url": "url", "code_url": "url", "revisions": ("list", REVISION)} ENTRY = fields("researcher_identity_id track_id task_name submission_id published_by published_at") PUBLICATION_EVENT = {**fields("event_id request_id actor_hf_username timestamp action requested_action researcher_identity_id track_id reason"), "previous_publication": ENTRY, "new_publication": ENTRY} WATERMARK = {**fields("request_id action submission_id replace_submission_id"), "sequence": "int"} PUBLICATIONS = {"entries": ("list", ENTRY), "versions": ("map", ("map", WATERMARK)), "history": ("list", PUBLICATION_EVENT)} OPERATOR_PUBLICATION_EVENT = {**fields("event operator request_id"), "timestamp": "timestamp", "target": fields("researcher_identity_id track_id", "id"), "old_value": ENTRY, "new_value": ENTRY} PUBLICATIONS["events"] = ("list", OPERATOR_PUBLICATION_EVENT) REQUEST = {**fields("request_id action task_name track_id submission_id requested_by hf_subject researcher_identity_id replace_submission_id", "id"), **fields("requested_at completed_at status message"), "publication_sequence": "int"} TOMBSTONE = fields("submission_id task_name researcher_identity_id deleted_at request_id event_id event actor_hf_username reason") TOMBSTONE["team_id"] = "id" WORKER_HEARTBEAT = {"worker_id": "id", "worker_run_id": "id", "ready": "bool", "heartbeat_at": "timestamp"} SCHEMAS = {"registry": REGISTRY, "migration": REGISTRY, "submission": SUBMISSION, "history": SUBMISSION, "legacy_best": SUBMISSION, "progress": PROGRESS, "metadata": METADATA, "publications": PUBLICATIONS, "publication_request": REQUEST, "deletion_request": REQUEST, "tombstone": TOMBSTONE, "audit": EVENT} def clean(value, rule, depth=0): if depth > 40: fail("nesting_limit") if value is None: return None if isinstance(rule, dict): if not isinstance(value, dict): fail("expected_object") result = {} for key, item in value.items(): if key in SECRET_FIELDS or key in TRANSIENT_FIELDS: continue if key not in rule: fail("unclassified_field") result[key] = clean(item, rule[key], depth + 1) return result if isinstance(rule, tuple): if rule[0] in {"submission_event", "operator_resume_record"}: if isinstance(value, dict) and value.get("event") == "operator_compatible_resume": required = {"event", "event_id", "timestamp", "reason", "backup_directory", "retained_results_digest", "retained_scenes"} if not required <= value.keys(): fail("incomplete_operator_resume_event") result = clean(value, LEGACY_OPERATOR_RESUME_EVENT, depth + 1) if any(result.get(key) is None for key in required) or result["retained_scenes"] < 1: fail("invalid_operator_resume_event") clean(result["timestamp"], "timestamp", depth + 1) return result if rule[0] == "submission_event" and isinstance(value, dict) and value.get("event") == "operator_priority_pause": return clean(value, OPERATOR_PAUSE_EVENT, depth + 1) if rule[0] == "operator_resume_record" and (not isinstance(value, dict) or value.get("event") != "operator_compatibility_resume"): fail("invalid_operator_resume_event") if isinstance(value, dict) and value.get("event") == "operator_compatibility_resume": required = {"event_id", "timestamp", "submission_id", "reason", "backup_directory", "previous_evaluator", "retained_results_digest", "retained_scenes", "verification"} if not required <= value.keys(): fail("incomplete_operator_resume_event") result = clean(value, OPERATOR_RESUME_EVENT, depth + 1) if any(result.get(key) is None for key in required) or not isinstance(result["verification"], dict): fail("incomplete_operator_resume_event") if set(result["verification"]) != {"pmf_tree_at_both_commits", "worker_source_change", "full_view_regression"} or any(item is None for item in result["verification"].values()): fail("incomplete_operator_resume_event") retained = result["retained_scenes"] if not retained or len(set(retained)) != len(retained): fail("invalid_retained_scene_identity") clean(result["timestamp"], "timestamp", depth + 1) return result return clean(value, rule[1], depth + 1) if rule[0] == "legacy_account": return clean(value, "id" if isinstance(value, str) else rule[1], depth + 1) if rule[0] == "list": if not isinstance(value, list): fail("expected_array") return [clean(item, rule[1], depth + 1) for item in value] if rule[0] == "map": if not isinstance(value, dict): fail("expected_map") return {clean(key, "id", depth + 1): clean(item, rule[1], depth + 1) for key, item in value.items()} if rule[0] == "profile_changes": return clean(value, {key: {"old_value": "text", "new_value": "text"} for key in PROFILE_FIELDS}, depth + 1) fail("unknown_schema_rule") if rule == "bool": if type(value) is not bool: fail("expected_boolean") return value if rule in {"number", "int"}: if type(value) not in {int, float} or not math.isfinite(value) or rule == "int" and (type(value) is not int or value < 0): fail("invalid_number") return value if rule == "version" and type(value) is int: return clean(value, "int", depth + 1) if not isinstance(value, str): fail("expected_string") if rule in {"sha1", "sha256"}: if not re.fullmatch(r"[0-9a-f]{40}" if rule == "sha1" else r"[0-9a-f]{64}", value): fail("invalid_evidence_digest") return value if rule == "timestamp": try: if datetime.fromisoformat(value.replace("Z", "+00:00")).tzinfo is None: fail("invalid_timestamp") except ValueError: fail("invalid_timestamp") if rule == "url": return sanitize_url(value) if value else value sanitized = sanitize_text(value) if rule == "id" and (not value or len(value) > 512 or not re.fullmatch(r"[A-Za-z0-9_.:@/-]+", value) or sanitized != value or ".." in value.split("/")): fail("invalid_identifier") return sanitized