# SPDX-License-Identifier: AGPL-3.0-only # Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0 """Static scan of a model's ``auto_map`` remote code, for the consent gate. When a user opts into ``trust_remote_code``, the repo's ``auto_map`` Python (``modeling_*.py`` etc.) is scanned BEFORE execution and suspicious patterns are surfaced to inform consent. A warning aid, not a hard boundary: a determined attacker can obfuscate past regexes, so the job is to raise the bar and inform the hash-pinned consent. Containment (subprocess/venv) is separate; execution still requires opt-in. Single source of truth: ``scripts/scan_packages.py`` (the scanner CI runs via ``security-audit.yml``). We import its ``check_py_file`` so the gate inherits every CI improvement with no drift; its heuristics are deliberately low-false-positive (combinations flag, not bare ``subprocess``/``eval``). When ``scripts/`` is absent (stripped install) we fall back to ``_FALLBACK_PATTERNS`` below; a test asserts the canonical scanner loads in-repo so the fallback never silently takes over. """ from __future__ import annotations import hashlib import importlib.util import pathlib import re import sys from dataclasses import dataclass, field from typing import Optional from loggers import get_logger logger = get_logger(__name__) CRITICAL = "CRITICAL" HIGH = "HIGH" MEDIUM = "MEDIUM" _SEVERITY_ORDER = {CRITICAL: 0, HIGH: 1, MEDIUM: 2} # Bump on any ruleset change (patterns, severities). A persisted approval records the version # it was scanned under; the consent cache ignores older-ruleset approvals so the same bytes # are re-scanned and re-shown instead of silently auto-approved. SCAN_RULES_VERSION = 1 # Configs that can carry an ``auto_map`` pointing at executable repo ``.py``. # ``trust_remote_code`` runs code from ANY of these, so scanner and gate must read the # same set (scanning only config.json/tokenizer would miss a custom-processor VLM). REMOTE_CODE_CONFIG_FILES = ( "config.json", "tokenizer_config.json", "preprocessor_config.json", "processor_config.json", "video_preprocessor_config.json", ) class RemoteCodeUnscannable(Exception): """The repo's executable code could not be fully fetched/listed to scan. Raised (not returned empty) so the gate distinguishes "code PRESENT but unreadable" (offline/gated/transient/404/listing failure) -> fail CLOSED, from "repo has NO .py" (empty result) -> trust_remote_code is a no-op -> allow. Conflating them would block a code-free repo or fail open on code we could not see. """ # Fallback patterns (used only if scripts/scan_packages.py is absent): (regex, check, # severity). A flat subset of the canonical scanner so a stripped install still scans; # the canonical scanner (imported below) supersedes it whenever the repo is present. _FALLBACK_PATTERNS: tuple[tuple[re.Pattern, str, str], ...] = ( ( re.compile( r"\bexec\s*\(\s*(?:urllib|requests|httpx|urlopen)" r"|\bexec\s*\([^)]*\.(?:text|content|read)\s*\(" r"|\beval\s*\([^)]*\.(?:text|content|read)\s*\(" r"|\b__import__\s*\([^)]*\+", re.DOTALL, ), "loads-and-executes-remote-code", CRITICAL, ), ( re.compile( r"\bsocket\b.*\bconnect\b.*\bsubprocess\b" r"|\bsocket\b.*\bconnect\b.*\b(?:sh|bash|cmd)\b" r"|\bpty\s*\.\s*spawn\b|\bos\s*\.\s*dup2\s*\(", re.DOTALL, ), "reverse/bind-shell", CRITICAL, ), ( re.compile( r"169\.254\.169\.254|metadata\.google\.internal|/latest/meta-data" r"|/metadata/identity|169\.254\.170\.2" ), "cloud-metadata/IMDS-access", CRITICAL, ), ( re.compile( r"(?:open|Path|read_text|read_bytes)\s*\([^)]*?" r"(?:\.ssh[/\\]|\.aws[/\\]|\.kube[/\\]|\.gnupg[/\\]|id_rsa|id_ed25519" r"|credentials\.json|\.git-credentials|\.npmrc|\.pypirc|/etc/shadow)" r"|(?:open|Path)\(\s*['\"]\.env['\"]\s*[,)]", re.DOTALL, ), "credential-file-access", CRITICAL, ), ( re.compile(r"/tmp/\S+.*(?:subprocess|os\.system|os\.popen|Popen|chmod.*\+x)", re.DOTALL), "tmp-staged-dropper", CRITICAL, ), ( re.compile(r"\bopenssl\s+(enc|rand|rsautl|pkeyutl|genrsa|dgst|s_client)\b"), "openssl-cli-exfil", HIGH, ), ( re.compile( r"\bsubprocess\s*\.\s*(Popen|call|run|check_call|check_output)\b" r"|\bos\s*\.\s*(system|popen|exec[lv]p?e?)\b" ), "subprocess/os-exec", HIGH, ), # Bare exec()/eval() only; the (? Optional[str]: if not self.findings: return None return min((f.severity for f in self.findings), key = lambda s: _SEVERITY_ORDER[s]) @property def clean(self) -> bool: return not self.findings def summary(self) -> str: if self.clean: return "no suspicious patterns found" by = {} for f in self.findings: by.setdefault(f.severity, set()).add(f.check) parts = [] for sev in (CRITICAL, HIGH, MEDIUM): if sev in by: parts.append(f"{sev}: {', '.join(sorted(by[sev]))}") return "; ".join(parts) def findings_payload(self) -> list[dict]: """Structured findings for the UI: one record per match, with line + snippet.""" return [ { "severity": f.severity, "file": f.filename, "check": f.check, "evidence": f.evidence, "line": f.line, "snippet": f.snippet, } for f in self.findings ] # Canonical scanner: import scripts/scan_packages.py by file path (scripts/ is not an # importable package from the backend root). It imports only stdlib at module level and # guards its CLI under __main__, so importing it is side-effect-free. _CANON_SENTINEL = object() _canon_cache = _CANON_SENTINEL def _load_canonical_scanner(): """Return the ``scripts/scan_packages.py`` module, or None if unavailable.""" global _canon_cache if _canon_cache is not _CANON_SENTINEL: return _canon_cache module = None # Walk up from this file to a repo root that contains scripts/scan_packages.py. here = pathlib.Path(__file__).resolve() for parent in here.parents: candidate = parent / "scripts" / "scan_packages.py" if candidate.is_file(): try: spec = importlib.util.spec_from_file_location("unsloth_scan_packages", candidate) mod = importlib.util.module_from_spec(spec) sys.modules.setdefault("unsloth_scan_packages", mod) spec.loader.exec_module(mod) # type: ignore[union-attr] if hasattr(mod, "check_py_file"): module = mod except Exception as exc: # pragma: no cover - defensive logger.warning("Could not load canonical scan_packages.py: %s", exc) break if module is None: logger.warning( "scripts/scan_packages.py not found; remote-code scan using the " "vendored fallback patterns." ) _canon_cache = module return module # Model-context-strict patterns. The canonical scanner only flags bare # ``subprocess``/``eval`` in combinations (common in package build scripts), but a # model's modeling_*.py never legitimately shells out, so the gate flags them alone # (e.g. a bare ``subprocess.Popen`` in a config ``__init__``). _MODEL_STRICT_PATTERNS: tuple[tuple[re.Pattern, str, str], ...] = ( ( re.compile( r"\bsubprocess\s*\.\s*(Popen|call|run|check_call|check_output)\b" r"|\bos\s*\.\s*(system|popen|exec[lv]p?e?)\b" ), "subprocess/os-exec (model code)", HIGH, ), # Bare exec()/eval(); excludes attribute calls like torch ``module.eval()``. (re.compile(r"(?: " by _extract_evidence. _EVIDENCE_LINE_RE = re.compile(r"^L(\d+):") def _snippet_rows( content: str, line: int, col: Optional[int] = None, match_len: int = 0, ) -> list[dict]: """A `±_SNIPPET_CONTEXT`-line window around `line` (1-based). Rows are {number, text, is_match}; the matched row adds match_start/match_end for an inline highlight when a precise column span is known.""" lines = content.splitlines() if not lines or line < 1: return [] line = min(line, len(lines)) lo = max(1, line - _SNIPPET_CONTEXT) hi = min(len(lines), line + _SNIPPET_CONTEXT) rows: list[dict] = [] for n in range(lo, hi + 1): text = lines[n - 1] clipped = len(text) > _SNIPPET_MAX_LINE if clipped: text = text[:_SNIPPET_MAX_LINE] + " ..." row = {"number": n, "text": text, "is_match": n == line} if n == line and col is not None and match_len > 0 and not clipped: row["match_start"] = col row["match_end"] = min(col + match_len, len(text)) rows.append(row) return rows def _attach_location( content: str, finding: Finding, match: "Optional[re.Match]" = None, ) -> None: """Populate finding.line + finding.snippet. A regex match gives a precise line+column; canonical findings are located via the `L:` prefix in their evidence.""" if match is not None: before = content[: match.start()] line = before.count("\n") + 1 col = match.start() - (before.rfind("\n") + 1) finding.line = line finding.snippet = _snippet_rows(content, line, col, match.end() - match.start()) return first = finding.evidence.splitlines()[0] if finding.evidence else "" tag = _EVIDENCE_LINE_RE.match(first) if tag: line = int(tag.group(1)) finding.line = line finding.snippet = _snippet_rows(content, line) def _scan_content(content: str, filename: str) -> list[Finding]: findings: list[Finding] = [] canon = _load_canonical_scanner() if canon is not None: # Canonical Finding is (severity, package, filename, check, evidence); adapt to # the gate's (severity, filename, check, evidence). for f in canon.check_py_file(content, filename, ""): finding = Finding(f.severity, f.filename, f.check, (f.evidence or "")[:120]) _attach_location(content, finding) findings.append(finding) else: for pat, check, sev in _FALLBACK_PATTERNS: m = pat.search(content) if m: finding = Finding(sev, filename, check, m.group(0)[:120]) _attach_location(content, finding, m) findings.append(finding) # Augment with the model-context-strict patterns the package scanner omits. have = {f.check for f in findings} for pat, check, sev in _MODEL_STRICT_PATTERNS: if check in have: continue m = pat.search(content) if m: finding = Finding(sev, filename, check, m.group(0)[:120]) _attach_location(content, finding, m) findings.append(finding) return findings def scan_remote_code_files(files: dict[str, str]) -> ScanResult: """Scan a mapping of {filename: content} and return aggregated findings.""" result = ScanResult(fingerprint = remote_code_fingerprint(files)) for name, content in files.items(): if not name.endswith(".py"): continue result.findings.extend(_scan_content(content or "", name)) return result def remote_code_fingerprint(files: dict[str, str]) -> str: """Stable sha256 over the (sorted) file contents, for pinning consent.""" h = hashlib.sha256() for name in sorted(files): h.update(name.encode("utf-8")) h.update(b"\0") h.update((files[name] or "").encode("utf-8")) h.update(b"\0") return h.hexdigest() def repo_remote_code_files(model_name: str, hf_token: Optional[str] = None) -> dict[str, str]: """Download a repo's executable ``.py`` (auto_map targets + modeling/config). Returns {filename: content}. An EMPTY dict means the repo ships no executable ``.py`` (trust_remote_code is a no-op). Raises ``RemoteCodeUnscannable`` when code is present but cannot be fully fetched/listed (offline/gated/404/listing failure), so the caller fails closed rather than fingerprint a partial view of code transformers would run in full. The empty-vs-raise split lets the gate allow a code-free repo while still blocking unscannable code. """ import json from pathlib import Path files: dict[str, str] = {} try: from utils.paths import is_local_path, normalize_path if is_local_path(model_name): root = Path(normalize_path(model_name)).expanduser() # Walk ALL .py, not just the auto_map entry's static import closure. This is # DELIBERATE (see the remote-branch note): the entry can reach a sibling via # an absolute import, importlib, or exec, which a relative-import closure # misses, so closure-only scanning is a real bypass. Broad scan never # under-scans; the cost is a benign script can over-block, the safe direction # for an RCE gate (HIGH stays approvable; only CRITICAL hard-blocks). for p in root.rglob("*.py"): if p.is_file(): files[str(p.relative_to(root))] = p.read_text(errors = "replace") # A local config can still point auto_map at an EXTERNAL Hub repo # (owner/name--module.Class) that executes on load, so fetch it. Every config # that can declare auto_map is checked, so a custom processor's external code # is not missed. ext_refs = set() for name in REMOTE_CODE_CONFIG_FILES: p = root / name if p.is_file(): try: ext_refs |= _auto_map_refs(json.loads(p.read_text())) except Exception: pass if not _add_external_refs(files, ext_refs, hf_token, model_name): raise RemoteCodeUnscannable(f"{model_name}: external auto_map code unreachable") return files from huggingface_hub import hf_hub_download, list_repo_files from huggingface_hub.utils import EntryNotFoundError # Collect auto_map refs from EVERY config that can declare one. A 404 # (EntryNotFoundError) means the config is absent -> skip; any other failure is # transient/auth and could hide an auto_map, so fail closed (unscannable). refs = set() for cfg_name in REMOTE_CODE_CONFIG_FILES: try: cfg_path = hf_hub_download(model_name, cfg_name, token = hf_token) except EntryNotFoundError: continue except Exception as exc: raise RemoteCodeUnscannable( f"{model_name}: config {cfg_name} could not be fetched ({exc})" ) from exc try: refs |= _auto_map_refs(json.loads(Path(cfg_path).read_text())) except Exception: pass own_refs = {fn for repo, fn in refs if repo is None} # The full file list catches helper .py the auto_map code imports but does not # name. If we cannot list the repo, an imported module could be missed and the # fingerprint cover less than transformers runs, so fail closed (unscannable). try: repo_files = list_repo_files(model_name, token = hf_token) except Exception as exc: raise RemoteCodeUnscannable(f"{model_name}: could not list repo files ({exc})") from exc repo_file_set = set(repo_files) # Scan every present .py PLUS own-repo auto_map targets that ACTUALLY EXIST in # this revision. Scanning EVERY .py (not just the closure) is DELIBERATE: the # entry can reach a sibling via absolute import / importlib / exec, which a # relative-import closure misses, so closure-only scanning is a real bypass. # Broad scan never under-scans; the cost is a benign script can over-block, the # safe direction for an RCE gate (HIGH approvable; only CRITICAL hard-blocks). An # auto_map target absent from the listing is a STALE ref (an older config naming a # since-removed file, e.g. unsloth/PaddleOCR-VL names processing_ppocrvl.py but # ships processing_paddleocr_vl.py). transformers cannot execute an absent file, # so drop the stale ref rather than fail closed; present .py are still fully # scanned. This also absorbs a mis-derived dotted name (sub.mod.py vs sub/mod.py): # the bad name drops as stale while the real present file is scanned. present_py = {f for f in repo_files if f.endswith(".py")} stale_refs = own_refs - repo_file_set for fn in sorted(stale_refs): logger.info( "repo_remote_code_files(%s): ignoring stale own-repo auto_map target " "%s (absent from the repo listing; it cannot execute)", model_name, fn, ) wanted = present_py | (own_refs & repo_file_set) for fn in sorted(wanted): try: fp = hf_hub_download(model_name, fn, token = hf_token) except Exception as exc: # A .py CONFIRMED PRESENT could not be fetched. A partial set would # fingerprint "clean" while transformers later runs this file, so fail # closed. (Stale/absent refs were dropped above, so this only fires on a # present-file fetch failure.) raise RemoteCodeUnscannable( f"{model_name}: present file {fn} could not be fetched ({exc})" ) from exc files[fn] = Path(fp).read_text(errors = "replace") # Code referenced from another repo executes too: scan it or fail closed. if not _add_external_refs(files, refs, hf_token, model_name): raise RemoteCodeUnscannable(f"{model_name}: external auto_map code unreachable") except RemoteCodeUnscannable: logger.warning("repo_remote_code_files(%s): unscannable; failing closed", model_name) raise except Exception as exc: # An unexpected error mid-scan means we could not complete it -> unscannable. raise RemoteCodeUnscannable(f"{model_name}: scan failed ({exc})") from exc # An empty dict here means the listing succeeded and the repo ships no executable .py # (nor fetchable external refs) -> trust_remote_code is a no-op for the caller. return files def _iter_auto_map_strings(value): """Yield every string class-ref inside one ``auto_map`` value. A value is a bare string (``"modeling_x.Cls"``) or a list/tuple (transformers encodes a tokenizer as ``"AutoTokenizer": [slow, fast]``, possibly nested or with nulls). Flatten all forms so external tokenizer code in the list shape is scanned. """ if isinstance(value, str): yield value elif isinstance(value, (list, tuple, set)): for item in value: yield from _iter_auto_map_strings(item) elif isinstance(value, dict): for item in value.values(): yield from _iter_auto_map_strings(item) def _auto_map_refs(cfg: dict) -> set: """``(repo, filename)`` pairs referenced by config auto_map. ``repo`` is ``None`` for own-repo code. An external ``owner/name--module.Class`` ref (transformers' cross-repo form) yields ``("owner/name", "module.py")`` so cross-repo code is scanned + fingerprinted too. """ out = set() am = cfg.get("auto_map") or {} if isinstance(am, dict): for value in am.values(): # A value may be a string OR a [slow, fast] list (tokenizers); cover both. for ref in _iter_auto_map_strings(value): # ref like "modeling_deepseekocr.Cls" or "owner/name--modeling.Cls" if "." not in ref: continue module = ref.rsplit(".", 1)[0] # drop trailing .ClassName if "--" in module: repo, mod = module.split("--", 1) out.add((repo or None, mod + ".py")) else: out.add((None, module + ".py")) return out def _auto_map_py(cfg: dict) -> set[str]: """Own-repo ``.py`` filenames referenced by auto_map (external refs excluded).""" return {fn for repo, fn in _auto_map_refs(cfg) if repo is None} def external_auto_map_repos(model_name: str, hf_token: Optional[str] = None) -> set: """External Hub repos referenced by any of this model's auto_map configs. The ``owner/name`` repos ``_add_external_refs`` downloads. The scan route uses this so declining consent purges them too, not leaving untrusted external code cached. Best-effort, config/metadata-only: returns whatever can be read, never raises. """ repos: set = set() try: import json from pathlib import Path from utils.paths import is_local_path, normalize_path if is_local_path(model_name): root = Path(normalize_path(model_name)).expanduser() for cfg_name in REMOTE_CODE_CONFIG_FILES: p = root / cfg_name if not p.is_file(): continue try: refs = _auto_map_refs(json.loads(p.read_text())) except Exception: continue repos.update(repo for repo, _fn in refs if repo) return repos from huggingface_hub import hf_hub_download from huggingface_hub.utils import EntryNotFoundError for cfg_name in REMOTE_CODE_CONFIG_FILES: try: cfg_path = hf_hub_download(model_name, cfg_name, token = hf_token) except EntryNotFoundError: continue except Exception: continue try: refs = _auto_map_refs(json.loads(Path(cfg_path).read_text())) except Exception: continue repos.update(repo for repo, _fn in refs if repo) except Exception: return repos return repos def _add_external_refs(files: dict, refs, hf_token, model_name: str) -> bool: """Download external-repo auto_map code into ``files`` (keyed ``repo--file``). transformers fetches the entry file AND its relative imports from the same external repo, so scanning only the entry would miss code in a ``helper.py`` it imports. Mirror the own-repo path: enumerate each external repo's ``.py`` and scan the whole set (plus the referenced entry files). Returns False if any external repo cannot be listed or a file cannot be fetched, so the caller fails closed. """ from pathlib import Path from huggingface_hub import hf_hub_download, list_repo_files # Group the explicit entry refs by external repo. entries: dict = {} for repo, fn in refs: if repo is None: continue entries.setdefault(repo, set()).add(fn) for repo, entry_files in entries.items(): try: repo_files = list_repo_files(repo, token = hf_token) except Exception as exc: logger.warning( "repo_remote_code_files(%s): external repo %s unlistable (%s); failing closed", model_name, repo, exc, ) return False # The loader's executable closure = every present .py plus any referenced entry # file. With a REAL (non-empty) listing, present_py covers the code, so an entry # ref absent from it is stale/mis-derived and is dropped rather than failing # closed (like the own-repo path). With an EMPTY listing we cannot prove the ref # stale, so keep fetching it and fail closed if unreachable; never under-scan. A # PRESENT file that cannot be fetched still fails closed below. repo_file_set = set(repo_files) present_py = {f for f in repo_files if f.endswith(".py")} if repo_file_set: for fn in sorted(set(entry_files) - repo_file_set): logger.info( "repo_remote_code_files(%s): ignoring stale external auto_map target " "%s:%s (absent from the repo listing; it cannot execute)", model_name, repo, fn, ) wanted = present_py | (set(entry_files) & repo_file_set) else: wanted = present_py | set(entry_files) for fn in sorted(wanted): try: fp = hf_hub_download(repo, fn, token = hf_token) except Exception as exc: logger.warning( "repo_remote_code_files(%s): external %s:%s unscannable (%s)", model_name, repo, fn, exc, ) return False files[f"{repo}--{fn}"] = Path(fp).read_text(errors = "replace") return True