# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-people-dont-know-about-hugging-face-incident-of-2026-at-open-ai.md # Remnant Warrant: https://corpusfetch.com/c/what-people-dont-know-about-hugging-face-incident-of-2026-at-open-ai.warrant.json # Markdown SHA-256: 7dc3da094accedf6b0ed743ce4d40f317112e7c36d9f95840de4e37bb2664279 # Code AST SHA-256: 7caddc190ab84245f07d29e246fc2d5e09e9624b682efce1586f4502242d4adb # Verified Classes: HubIncidentQuarantineError, PinnedArtifactLock, Post2026HubQuarantineGate # Defensive Guards: 10 fail-closed invariants (zero external deps) import hashlib import json import os import re import struct from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path from typing import Any, Final, Iterator class HubIncidentQuarantineError(PermissionError): '''Raised when an external hub artifact exhibits a 2026-class supply-chain attack vector.''' pass @dataclass(frozen=True) class PinnedArtifactLock: repo_id: str commit_sha: str file_sha256_manifest: dict[str, str] # relative_filename -> expected_sha256 class Post2026HubQuarantineGate: ''' Hardened defence against the 2026 Hugging Face / OpenAI evaluation-runner attack chain: 1. Enforces 40-character Git commit SHA pinning and per-file SHA-256 lockfile verification. 2. Scans `config.json` and `tokenizer_config.json` for hidden `auto_map` remote code hooks. 3. Blocks `.py`, `.bin`, `.pt`, and `.pkl` files across both model weights and tokenizers. 4. Scrubs ambient environment credentials and enforces `HF_HUB_OFFLINE=1` during load. ''' COMMIT_SHA_REGEX: Final[re.Pattern[str]] = re.compile(r"^[0-9a-f]{40}$") FORBIDDEN_EXTENSIONS: Final[tuple[str, ...]] = (".py", ".bin", ".pt", ".pth", ".pkl", ".pickle", ".sh", ".so") MAX_SAFETENSORS_HEADER_BYTES: Final[int] = 100 * 1024 SENSITIVE_ENV_PREFIXES: Final[tuple[str, ...]] = ( "OPENAI_", "HF_TOKEN", "HUGGING_FACE_HUB_TOKEN", "AWS_", "GCP_", "AZURE_", "KUBERNETES_", "GITHUB_TOKEN", "WANDB_", ) @classmethod def verify_quarantined_repo(cls, staging_dir: Path, lock: PinnedArtifactLock) -> None: ''' Executes full pre-load forensic verification on a staged model/tokenizer directory. ''' # Gate 1: Verify immutable 40-character commit SHA if not cls.COMMIT_SHA_REGEX.match(lock.commit_sha): raise HubIncidentQuarantineError( f"Mutable revision '{lock.commit_sha}' rejected for {lock.repo_id}. " "Must pin an exact 40-character hexadecimal Git commit SHA." ) if not staging_dir.exists() or not staging_dir.is_dir(): raise HubIncidentQuarantineError(f"Quarantine directory missing: {staging_dir}") # Gate 2: Scan all files in staging directory for forbidden extensions and unlisted files for file_path in staging_dir.rglob("*"): if not file_path.is_file(): continue rel_name = file_path.relative_to(staging_dir).as_posix() lower_name = rel_name.lower() if any(lower_name.endswith(ext) for ext in cls.FORBIDDEN_EXTENSIONS): raise HubIncidentQuarantineError( f"2026-Vector Blocked: Forbidden executable or pickle file '{rel_name}' found in staging." ) if rel_name not in lock.file_sha256_manifest: raise HubIncidentQuarantineError( f"Unlisted artifact '{rel_name}' present in staging directory but absent from SHA-256 lockfile." ) # Gate 3: Verify SHA-256 cryptographic digest of every locked file and audit configs/headers for rel_name, expected_hash in lock.file_sha256_manifest.items(): target_file = staging_dir / rel_name if not target_file.is_file(): raise HubIncidentQuarantineError(f"Missing required locked file: {rel_name}") cls._verify_sha256(target_file, expected_hash) if rel_name.endswith(".json"): cls._audit_json_config_for_remote_hooks(target_file) elif rel_name.endswith(".safetensors"): cls._audit_safetensors_header(target_file) @staticmethod def _verify_sha256(file_path: Path, expected_sha256: str) -> None: hasher = hashlib.sha256() with file_path.open("rb") as f: while chunk := f.read(65536): hasher.update(chunk) actual = hasher.hexdigest().lower() if actual != expected_sha256.lower(): raise HubIncidentQuarantineError( f"SHA-256 mismatch on '{file_path.name}': expected {expected_sha256}, got {actual}" ) @staticmethod def _audit_json_config_for_remote_hooks(json_path: Path) -> None: ''' Detects the primary 2026 incident vector: hidden `auto_map` or `custom_pipelines` keys inside `config.json` or `tokenizer_config.json` that trigger dynamic `.py` imports. ''' try: data = json.loads(json_path.read_text(encoding="utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise HubIncidentQuarantineError(f"Invalid JSON in {json_path.name}: {exc}") from exc if isinstance(data, dict): forbidden_keys = {"auto_map", "custom_pipelines"} matched = forbidden_keys.intersection(data.keys()) if matched: raise HubIncidentQuarantineError( f"2026-Vector Blocked: '{json_path.name}' contains dynamic code hook key(s): {matched}" ) @classmethod def _audit_safetensors_header(cls, tensor_path: Path) -> None: with tensor_path.open("rb") as f: prefix = f.read(8) if len(prefix) != 8: raise HubIncidentQuarantineError(f"Truncated safetensors header in {tensor_path.name}") (header_size,) = struct.unpack(" cls.MAX_SAFETENSORS_HEADER_BYTES: raise HubIncidentQuarantineError( f"Abnormal safetensors header size ({header_size} bytes) in {tensor_path.name}" ) raw_header = f.read(header_size) json.loads(raw_header.decode("utf-8")) @classmethod @contextmanager def isolated_offline_loader_context(cls) -> Iterator[dict[str, Any]]: ''' Scrubs ambient internal secrets from `os.environ` and locks offline loader kwargs. ''' snapshot = dict(os.environ) try: for k in list(os.environ.keys()): if any(k.startswith(p) for p in cls.SENSITIVE_ENV_PREFIXES): os.environ.pop(k, None) os.environ["HF_HUB_OFFLINE"] = "1" os.environ["TRANSFORMERS_OFFLINE"] = "1" os.environ["HF_DATASETS_OFFLINE"] = "1" yield { "trust_remote_code": False, "use_safetensors": True, "local_files_only": True, } finally: os.environ.clear() os.environ.update(snapshot)