scanner.py

from __future__ import annotations

import ast
import concurrent.futures
import hashlib
import json
import re
import threading
import time
import tomllib
import uuid
import warnings
from collections import Counter
from dataclasses import asdict, dataclass, field
from pathlib import PurePosixPath
from typing import Any, Iterable
from urllib.parse import quote, urlparse

import requests


HF_API = "https://huggingface.co/api"
HF_WEB = "https://huggingface.co"
OSV_API = "https://api.osv.dev/v1"
MAX_FILE_BYTES = 512_000
MAX_FILES = 80
MAX_OSV_PACKAGES = 50
MAX_OSV_VULNERABILITIES = 40
REQUEST_TIMEOUT = (5, 20)
PRODUCT_NAME = "ModelSentry"
OSV_CACHE_TTL_SECONDS = 900

_OSV_BATCH_CACHE: dict[tuple[tuple[str, str], ...], tuple[float, Any]] = {}
_OSV_VULN_CACHE: dict[str, tuple[float, Any]] = {}
_OSV_CACHE_LOCK = threading.Lock()

ALLOWED_EXACT = {
    "README.md",
    "LICENSE",
    "LICENSE.md",
    "requirements.txt",
    "requirements-dev.txt",
    "pyproject.toml",
    "setup.py",
    "setup.cfg",
    "packages.txt",
    "Dockerfile",
    "docker-compose.yml",
    "docker-compose.yaml",
    "environment.yml",
    "environment.yaml",
    ".gitattributes",
    ".modelsentry.json",
}
ALLOWED_SUFFIXES = {".py", ".json", ".yaml", ".yml", ".toml", ".ini", ".cfg", ".md", ".txt"}
SKIP_PARTS = {".git", ".venv", "venv", "node_modules", "dist", "build", "__pycache__"}
SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4}
WEIGHT_SUFFIXES = {".safetensors", ".bin", ".pt", ".pth", ".ckpt", ".gguf", ".onnx"}
EXECUTABLE_SUFFIXES = {".py", ".js", ".ts", ".sh", ".ps1"}
MANIFEST_NAMES = {"requirements.txt", "requirements-dev.txt", "pyproject.toml", "setup.py", "setup.cfg", "packages.txt"}

SECRET_ASSIGNMENT = re.compile(
    r"(?i)\b(api[_-]?key|access[_-]?token|auth[_-]?token|secret|password|passwd|private[_-]?key)\b"
    r"\s*[:=]\s*([\"'])([^\"'\n]{8,})\2"
)
SECRET_VALUE = re.compile(r"(?i)(api[_-]?key|token|secret|password)(\s*[:=]\s*)([^\s,;]+)")


@dataclass(frozen=True)
class Finding:
    rule_id: str
    severity: str
    title: str
    detail: str
    remediation: str
    confidence: str = "high"
    path: str | None = None
    line: int | None = None
    evidence: str | None = None
    status: str = "unresolved"


@dataclass(frozen=True)
class CheckResult:
    category: str
    status: str
    detail: str


@dataclass
class ScanResult:
    target: str
    repo_type: str
    revision: str
    findings: list[Finding] = field(default_factory=list)
    dependencies: list[str] = field(default_factory=list)
    dependency_details: list[dict[str, Any]] = field(default_factory=list)
    artifacts: dict[str, Any] = field(default_factory=dict)
    metadata_summary: dict[str, Any] = field(default_factory=dict)
    checks: list[CheckResult] = field(default_factory=list)
    inspected_files: list[str] = field(default_factory=list)
    skipped_files: int = 0
    notes: list[str] = field(default_factory=list)
    package_inventory: list[dict[str, Any]] = field(default_factory=list)
    vulnerabilities: list[dict[str, Any]] = field(default_factory=list)
    osv_summary: dict[str, Any] = field(default_factory=lambda: {
        "status": "not_checked", "queried_packages": 0, "exact_packages": 0,
        "vulnerability_count": 0, "truncated": False,
    })
    sbom: dict[str, Any] = field(default_factory=dict)

    def finding_groups(self) -> list[dict[str, Any]]:
        grouped: dict[tuple[str, str, str], dict[str, Any]] = {}
        for finding in self.sorted_findings():
            key = (finding.rule_id, finding.severity, finding.status)
            item = grouped.setdefault(key, {
                "rule_id": finding.rule_id,
                "severity": finding.severity,
                "status": finding.status,
                "title": finding.title,
                "detail": finding.detail,
                "remediation": finding.remediation,
                "confidence": finding.confidence,
                "occurrences": [],
            })
            item["occurrences"].append({
                "path": finding.path,
                "line": finding.line,
                "evidence": finding.evidence,
            })
        return sorted(
            grouped.values(),
            key=lambda item: (SEVERITY_ORDER.get(item["severity"], 99), item["rule_id"], item["title"]),
        )

    def sorted_findings(self) -> list[Finding]:
        return sorted(
            self.findings,
            key=lambda item: (SEVERITY_ORDER.get(item.severity, 99), item.path or "", item.line or 0),
        )

    def counts(self) -> dict[str, int]:
        counts = {name: 0 for name in SEVERITY_ORDER}
        for group in self.finding_groups():
            counts[group["severity"]] = counts.get(group["severity"], 0) + 1
        return counts

    def to_dict(self) -> dict[str, Any]:
        return {
            "schema": "modelsentry.scan.v4",
            "target": self.target,
            "repo_type": self.repo_type,
            "revision": self.revision,
            "summary": self.counts(),
            "dependencies": self.dependencies,
            "dependency_details": self.dependency_details,
            "package_inventory": self.package_inventory,
            "vulnerabilities": self.vulnerabilities,
            "osv": self.osv_summary,
            "cyclonedx_sbom": self.sbom,
            "metadata": self.metadata_summary,
            "artifacts": self.artifacts,
            "coverage": [asdict(item) for item in self.checks],
            "inspected_files": self.inspected_files,
            "skipped_files": self.skipped_files,
            "notes": self.notes,
            "finding_occurrence_count": len(self.findings),
            "findings": self.finding_groups(),
        }


class ScanError(RuntimeError):
    pass


def parse_target(value: str) -> tuple[str, str]:
    raw = (value or "").strip().rstrip("/")
    if not raw:
        raise ScanError("Enter a Hugging Face Model or Space URL.")

    repo_type = "model"
    if ":" in raw and not raw.startswith(("http://", "https://")):
        prefix, raw = raw.split(":", 1)
        mapping = {"model": "model", "space": "space", "dataset": "dataset"}
        if prefix.lower() in mapping:
            repo_type = mapping[prefix.lower()]

    if raw.startswith(("http://", "https://")):
        parsed = urlparse(raw)
        if parsed.scheme != "https" or parsed.hostname not in {"huggingface.co", "www.huggingface.co"}:
            raise ScanError("Only https://huggingface.co URLs are accepted.")
        parts = [part for part in parsed.path.split("/") if part]
        if parts and parts[0] in {"models", "spaces", "datasets"}:
            repo_type = {"models": "model", "spaces": "space", "datasets": "dataset"}[parts.pop(0)]
        elif parts and parts[0] == "spaces":
            repo_type = "space"
            parts.pop(0)
        elif parts and parts[0] == "datasets":
            repo_type = "dataset"
            parts.pop(0)
        if len(parts) < 2:
            raise ScanError("The URL must contain an owner and repository name.")
        raw = "/".join(parts[:2])

    parts = raw.split("/")
    if len(parts) != 2 or not all(re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", part) for part in parts):
        raise ScanError("Use owner/repository, space:owner/repository, or a Hugging Face URL.")
    if repo_type == "dataset":
        raise ScanError("Dataset scanning is planned for version two; this MVP supports Models and Spaces.")
    return raw, repo_type


def _api_kind(repo_type: str) -> str:
    return {"model": "models", "space": "spaces", "dataset": "datasets"}[repo_type]


def _get_json(url: str) -> Any:
    try:
        response = requests.get(url, timeout=REQUEST_TIMEOUT, headers={"User-Agent": "ModelSentry/1.0"})
    except requests.Timeout as exc:
        raise ScanError("The Hugging Face Hub request timed out. Please try again.") from exc
    except requests.RequestException as exc:
        raise ScanError("The Hugging Face Hub could not be reached. Please try again.") from exc

    if response.status_code in {401, 403}:
        raise ScanError("The repository is private or access is denied. ModelSentry scans public repositories only.")
    if response.status_code == 404:
        raise ScanError("The repository was not found. Check the owner/repository name and public visibility.")
    if response.status_code == 429:
        raise ScanError("The Hugging Face Hub rate limit was reached. Please wait and try again.")
    if response.status_code >= 500:
        raise ScanError("The Hugging Face Hub is temporarily unavailable. Please try again.")
    if not response.ok:
        raise ScanError(f"The Hugging Face Hub rejected the request (HTTP {response.status_code}).")
    try:
        return response.json()
    except requests.exceptions.JSONDecodeError as exc:
        raise ScanError("The Hugging Face Hub returned an invalid metadata response. Please try again.") from exc


def _safe_path(path: str) -> bool:
    candidate = PurePosixPath(path)
    if candidate.is_absolute() or ".." in candidate.parts or any(part in SKIP_PARTS for part in candidate.parts):
        return False
    return candidate.name in ALLOWED_EXACT or candidate.suffix.lower() in ALLOWED_SUFFIXES


def _candidate_priority(item: tuple[str, int | None]) -> tuple[int, int, str]:
    path = item[0]
    candidate = PurePosixPath(path)
    name = candidate.name.lower()
    if candidate.name in ALLOWED_EXACT or name in MANIFEST_NAMES:
        rank = 0
    elif name in {"app.py", "main.py", "server.py", "api.py", "config.py", "settings.py"}:
        rank = 1
    elif any(term in path.lower() for term in ("auth", "security", "secret", "upload", "network", "route", "worker")):
        rank = 2
    elif candidate.suffix.lower() in EXECUTABLE_SUFFIXES:
        rank = 3
    else:
        rank = 4
    return rank, path.count("/"), path


def _download_text(repo_id: str, repo_type: str, revision: str, path: str, declared_size: int | None) -> str | None:
    if declared_size is not None and declared_size > MAX_FILE_BYTES:
        return None
    prefix = "" if repo_type == "model" else f"{_api_kind(repo_type)}/"
    url = f"{HF_WEB}/{prefix}{quote(repo_id, safe='/')}/resolve/{quote(revision, safe='')}/{quote(path, safe='/')}"
    response = requests.get(
        url,
        timeout=REQUEST_TIMEOUT,
        headers={"User-Agent": "ModelSentry/0.4", "Range": f"bytes=0-{MAX_FILE_BYTES}"},
        stream=True,
    )
    if response.status_code not in {200, 206}:
        return None
    data = bytearray()
    for chunk in response.iter_content(32_768):
        data.extend(chunk)
        if len(data) > MAX_FILE_BYTES:
            return None
    if b"\x00" in data[:4096]:
        return None
    return bytes(data).decode("utf-8", errors="replace")


def _line_evidence(line: str, limit: int = 180) -> str:
    collapsed = " ".join(line.strip().split())
    collapsed = SECRET_ASSIGNMENT.sub(lambda match: f"{match.group(1)} = <redacted>", collapsed)
    collapsed = SECRET_VALUE.sub(lambda match: f"{match.group(1)}{match.group(2)}<redacted>", collapsed)
    collapsed = re.sub(r"(https?://)[^/\s:@]+:[^/\s@]+@", r"\1<redacted>@", collapsed, flags=re.I)
    return collapsed[:limit]


def _finding(
    rule_id: str,
    severity: str,
    title: str,
    detail: str,
    remediation: str,
    path: str | None = None,
    line: int | None = None,
    evidence: str | None = None,
    confidence: str = "high",
    status: str = "unresolved",
) -> Finding:
    return Finding(rule_id, severity, title, detail, remediation, confidence, path, line, evidence, status)


def _line_findings(path: str, text: str) -> Iterable[Finding]:
    """Rules that are safe to apply to arbitrary text, not executable behavior."""
    patterns = [
        (
            "SECRET-001", "high", SECRET_ASSIGNMENT,
            "Possible hardcoded secret",
            "A credential-like value appears to be embedded directly in a text file. Its value is intentionally redacted.",
            "Revoke the exposed value, remove it from history, and use a managed Space secret.",
        ),
        (
            "DATA-001", "high", re.compile(r"\b(?:pickle\.load|pickle\.loads|joblib\.load)\s*\("),
            "Unsafe deserialization path detected",
            "Pickle-compatible formats can execute code when loaded from an untrusted source.",
            "Use a non-executable format such as safetensors or validate provenance before loading.",
        ),
        (
            "DATA-002", "medium", re.compile(r"\byaml\.load\s*\((?![^)]*(?:SafeLoader|safe_load))"),
            "Potentially unsafe YAML loading",
            "Generic YAML loading may construct arbitrary Python objects.",
            "Use yaml.safe_load or explicitly select SafeLoader.",
        ),
    ]
    for number, line in enumerate(text.splitlines(), 1):
        for rule_id, severity, regex, title, detail, remediation in patterns:
            if regex.search(line):
                yield _finding(
                    rule_id, severity, title, detail, remediation, path, number, _line_evidence(line)
                )


def _call_name(node: ast.AST) -> str:
    if isinstance(node, ast.Name):
        return node.id
    if isinstance(node, ast.Attribute):
        prefix = _call_name(node.value)
        return f"{prefix}.{node.attr}" if prefix else node.attr
    return ""


def _constant_assignments(tree: ast.AST) -> dict[str, Any]:
    values: dict[str, Any] = {}
    for node in getattr(tree, "body", []):
        if isinstance(node, (ast.Assign, ast.AnnAssign)):
            target = node.targets[0] if isinstance(node, ast.Assign) and len(node.targets) == 1 else getattr(node, "target", None)
            value = node.value
            if isinstance(target, ast.Name) and isinstance(value, ast.Constant):
                values[target.id] = value.value
    return values


def _resolved_value(node: ast.AST | None, constants: dict[str, Any]) -> Any:
    if isinstance(node, ast.Constant):
        return node.value
    if isinstance(node, ast.Name):
        return constants.get(node.id)
    return None


def _keyword(call: ast.Call, name: str) -> ast.AST | None:
    for keyword in call.keywords:
        if keyword.arg == name:
            return keyword.value
    return None


def _domains_from_expr(node: ast.AST | None, constants: dict[str, Any]) -> set[str]:
    if node is None:
        return set()
    value = _resolved_value(node, constants)
    if isinstance(value, str):
        return {match.group(1).lower() for match in re.finditer(r"https?://([A-Za-z0-9.-]+)", value)}
    if isinstance(node, ast.JoinedStr):
        pieces = []
        for part in node.values:
            if isinstance(part, ast.Constant) and isinstance(part.value, str):
                pieces.append(part.value)
            elif isinstance(part, ast.FormattedValue):
                resolved = _resolved_value(part.value, constants)
                pieces.append(str(resolved) if resolved is not None else "x")
        return {match.group(1).lower() for match in re.finditer(r"https?://([A-Za-z0-9.-]+)", "".join(pieces))}
    return set()


def _python_ast_findings(path: str, text: str, controls: dict[str, Any]) -> Iterable[Finding]:
    try:
        with warnings.catch_warnings():
            warnings.simplefilter("ignore")
            tree = ast.parse(text, filename="<untrusted-source>")
    except SyntaxError:
        return

    constants = _constant_assignments(tree)
    telemetry_modules = {"sentry_sdk", "wandb", "mlflow", "posthog", "mixpanel"}
    network_calls = {
        "requests.get", "requests.post", "requests.put", "requests.patch", "requests.delete", "requests.request",
        "httpx.get", "httpx.post", "httpx.put", "httpx.patch", "httpx.delete", "urllib.request.urlopen",
        "aiohttp.ClientSession", "socket.socket",
    }
    subprocess_calls = {"subprocess.run", "subprocess.Popen", "subprocess.call", "subprocess.check_call", "subprocess.check_output", "os.system"}
    secret_terms = ("token", "secret", "password", "passwd", "api_key", "apikey", "private_key")
    expected_domains = {str(item).lower() for item in controls.get("expected_outbound_domains", []) if item}
    assigned_domains: dict[str, set[str]] = {}
    for statement in ast.walk(tree):
        if isinstance(statement, ast.Assign):
            domains = _domains_from_expr(statement.value, constants)
            for target in statement.targets:
                if isinstance(target, ast.Name) and domains:
                    assigned_domains[target.id] = domains

    for node in ast.walk(tree):
        if isinstance(node, (ast.Import, ast.ImportFrom)):
            names = [alias.name.split(".", 1)[0] for alias in node.names] if isinstance(node, ast.Import) else [(node.module or "").split(".", 1)[0]]
            for name in names:
                if name in telemetry_modules:
                    yield _finding(
                        "TEL-001", "medium", "Telemetry or analytics SDK imported",
                        f"The source imports the {name} telemetry-capable SDK.",
                        "Document collected fields and destinations, and disable telemetry by default when practical.",
                        path, getattr(node, "lineno", None), _line_evidence(text.splitlines()[node.lineno - 1]),
                    )

        if isinstance(node, ast.Assign):
            for target in node.targets:
                if isinstance(target, ast.Subscript) and _call_name(target.value) == "os.environ" and isinstance(target.ctx, ast.Store):
                    key = _resolved_value(target.slice, constants)
                    yield _finding(
                        "ENV-001", "info", "Environment variable is configured",
                        "The application writes a process environment setting; this is configuration behavior, not proof of secret access.",
                        "Document operational environment settings that materially affect behavior.",
                        path, node.lineno, _line_evidence(text.splitlines()[node.lineno - 1]), status="observed",
                    )

        if isinstance(node, ast.Subscript) and _call_name(node.value) == "os.environ" and isinstance(node.ctx, ast.Load):
            key = _resolved_value(node.slice, constants)
            if isinstance(key, str):
                secret_like = any(term in key.lower() for term in secret_terms)
                yield _finding(
                    "ENV-004" if secret_like else "ENV-003",
                    "medium" if secret_like else "info",
                    "Secret-like environment variable is read" if secret_like else "Configuration environment variable is read",
                    f"The application reads {key}; reading does not by itself prove disclosure or transmission.",
                    "Ensure secret-like values are never logged, rendered, or sent to an undocumented destination.",
                    path, getattr(node, "lineno", None), _line_evidence(text.splitlines()[node.lineno - 1]), status="observed",
                )

        if not isinstance(node, ast.Call):
            continue
        name = _call_name(node.func)
        line = text.splitlines()[node.lineno - 1] if node.lineno and node.lineno <= len(text.splitlines()) else name

        remote_node = _keyword(node, "trust_remote_code")
        if _resolved_value(remote_node, constants) is True:
            revision = _resolved_value(_keyword(node, "revision"), constants)
            pinned = isinstance(revision, str) and bool(re.fullmatch(r"[0-9a-fA-F]{40,64}", revision))
            yield _finding(
                "CODE-004" if pinned else "CODE-001",
                "medium" if pinned else "high",
                "Remote repository code is enabled at an immutable revision" if pinned else "Remote repository code is enabled without an immutable revision",
                "Loading may execute Python supplied by the model repository. " + ("The exact revision is pinned." if pinned else "No immutable revision could be established."),
                "Review remote code and retain an immutable revision pin." if pinned else "Pin and review the exact model revision before enabling remote code.",
                path, node.lineno, _line_evidence(line), status="controlled" if pinned else "unresolved",
            )

        if name in subprocess_calls:
            shell_value = _resolved_value(_keyword(node, "shell"), constants)
            has_timeout = _keyword(node, "timeout") is not None
            if name == "os.system" or shell_value is True:
                severity, rule, title = "high", "CODE-002", "Shell execution is enabled"
                detail = "Shell interpretation can turn constructed or user-controlled text into commands."
            else:
                severity, rule, title = "low", "CODE-005", "Subprocess capability uses no explicit shell"
                detail = "The application launches a child process without an explicit shell." + (" A timeout is present." if has_timeout else "")
            yield _finding(
                rule, severity, title, detail,
                "Avoid shell execution; use fixed argument lists, allowlisted binaries, and bounded timeouts.",
                path, node.lineno, _line_evidence(line), status="controlled" if severity == "low" and has_timeout else "unresolved",
            )

        if isinstance(node.func, ast.Name) and node.func.id in {"eval", "exec"}:
            yield _finding(
                "CODE-003", "high", "Dynamic code execution detected",
                "Dynamic evaluation can turn untrusted input into executable code.",
                "Replace dynamic evaluation with a strict parser or explicit dispatch table.",
                path, node.lineno, _line_evidence(line),
            )

        if name in network_calls:
            network_domains = _domains_from_expr(node.args[0], constants) if node.args else set()
            if not network_domains and node.args and isinstance(node.args[0], ast.Name):
                network_domains = assigned_domains.get(node.args[0].id, set())
            network_documented = bool(expected_domains) and bool(network_domains) and network_domains.issubset(expected_domains)
            yield _finding(
                "NET-001", "info" if network_documented else "medium",
                "Documented outbound network capability" if network_documented else "Outbound network capability detected",
                "The application contains executable network-client behavior." + (" All literal destinations are covered by the repository declaration." if network_documented else ""),
                "Document every destination and purpose, minimize transmitted data, and disclose retention.",
                path, node.lineno, _line_evidence(line), status="documented" if network_documented else "unresolved",
            )

        env_key = None
        if name in {"os.getenv", "os.environ.get"} and node.args:
            env_key = _resolved_value(node.args[0], constants)
        elif name == "load_dotenv":
            yield _finding(
                "ENV-002", "info", "Dotenv configuration loading detected",
                "The application can load local environment configuration; this is not proof that a secret is exposed.",
                "Keep dotenv files out of source control and never render loaded credentials.",
                path, node.lineno, _line_evidence(line), status="observed",
            )
        if isinstance(env_key, str):
            secret_like = any(term in env_key.lower() for term in secret_terms)
            yield _finding(
                "ENV-004" if secret_like else "ENV-003",
                "medium" if secret_like else "info",
                "Secret-like environment variable is read" if secret_like else "Configuration environment variable is read",
                f"The application reads {env_key}; reading does not by itself prove disclosure or transmission.",
                "Ensure secret-like values are never logged, rendered, or sent to an undocumented destination.",
                path, node.lineno, _line_evidence(line), status="observed",
            )


def _requirements_findings(path: str, text: str) -> Iterable[Finding]:
    telemetry_packages = {"sentry-sdk", "wandb", "mlflow", "posthog", "mixpanel", "analytics-python"}
    for number, raw in enumerate(text.splitlines(), 1):
        line = raw.strip()
        if not line or line.startswith(("#", "--index-url", "--extra-index-url", "-f ", "--find-links")):
            continue
        normalized = line.lower().split(";", 1)[0].strip()
        package_match = re.match(r"([a-z0-9_.-]+)", normalized)
        package_name = package_match.group(1).replace("_", "-") if package_match else ""
        if package_name in telemetry_packages:
            yield _finding(
                "TEL-001", "medium", "Telemetry or analytics dependency declared",
                f"The dependency manifest includes {package_name}.",
                "Document whether telemetry is enabled, what is collected, and every destination.",
                path, number, _line_evidence(line),
            )
        if package_name == "python-dotenv":
            yield _finding(
                "ENV-002", "info", "Dotenv configuration dependency declared",
                "The application can load local environment configuration; this is not proof of secret exposure.",
                "Keep dotenv files out of source control and never render loaded credentials.",
                path, number, _line_evidence(line), status="observed",
            )

        is_vcs = "git+" in normalized
        is_direct_url = normalized.startswith(("http://", "https://", "-e ")) or " @ http" in normalized
        if is_vcs:
            pinned_commit = bool(re.search(r"@[0-9a-fA-F]{12,40}(?:#|$)", line))
            if not pinned_commit:
                yield _finding(
                    "DEP-002", "high", "Git dependency is not commit-pinned",
                    "The build can retrieve changing source code without an immutable commit pin.",
                    "Pin the dependency to a full commit SHA and document its license.",
                    path, number, _line_evidence(line),
                )
            continue
        if is_direct_url:
            hashed = bool(re.search(r"--hash\s*=\s*sha256:[0-9a-fA-F]{64}", line))
            versioned_release = bool(re.search(r"/releases/download/[^/\s]+/", line, re.I))
            if not hashed:
                yield _finding(
                    "DEP-003" if versioned_release else "DEP-004",
                    "medium" if versioned_release else "high",
                    "Versioned release asset lacks an integrity hash" if versioned_release else "Direct dependency URL lacks an integrity hash",
                    "The remote artifact can be replaced without changing this repository." if versioned_release else "The build retrieves an external artifact without cryptographic integrity verification.",
                    "Record and enforce a SHA-256 hash for the reviewed artifact.",
                    path, number, _line_evidence(line),
                )
            continue
        package = line.split(";", 1)[0].strip()
        if package.startswith("-"):
            continue
        if not re.search(r"(?:===|==)\s*[^*\s]+", package):
            yield _finding(
                "DEP-001", "medium", "Dependency is not exactly pinned",
                "A floating dependency can change the build without a repository change.",
                "Pin an exact tested version and use an automated review process for updates.",
                path, number, _line_evidence(line),
            )


def _canonical_package_name(name: str) -> str:
    return re.sub(r"[-_.]+", "-", name).lower()


def _parse_requirement(raw: str, path: str, line: int | None) -> dict[str, Any] | None:
    requirement = raw.strip()
    if not requirement or requirement.startswith(("#", "--", "-r ", "-c ", "-f ")):
        return None
    base = requirement.split(";", 1)[0].strip()
    lowered = base.lower()
    if "git+" in lowered:
        direct_name = re.match(r"([A-Za-z0-9][A-Za-z0-9._-]*)(?:\[[^\]]+\])?\s*@\s*git\+", base, re.I)
        egg = re.search(r"[#&]egg=([A-Za-z0-9._-]+)", base, re.I)
        name = _canonical_package_name((direct_name or egg).group(1)) if direct_name or egg else None
        source_type = "vcs"
        version = None
    elif lowered.startswith(("http://", "https://", "-e ")) or " @ http" in lowered:
        direct_name = re.match(r"([A-Za-z0-9][A-Za-z0-9._-]*)(?:\[[^\]]+\])?\s*@\s*https?://", base, re.I)
        name = _canonical_package_name(direct_name.group(1)) if direct_name else None
        source_type = "direct"
        version = None
    else:
        name_match = re.match(r"([A-Za-z0-9][A-Za-z0-9._-]*)(?:\[[^\]]+\])?", base)
        name = _canonical_package_name(name_match.group(1)) if name_match else None
        exact = re.match(
            r"^[A-Za-z0-9][A-Za-z0-9._-]*(?:\[[^\]]+\])?\s*(?:===|==)\s*([^*\s,]+)\s*$",
            base,
        )
        version = exact.group(1) if exact else None
        source_type = "exact" if version else "unpinned"
    if not name:
        return None
    purl = f"pkg:pypi/{quote(name, safe='')}" + (f"@{quote(version, safe='')}" if version else "")
    return {
        "ecosystem": "PyPI",
        "name": name,
        "version": version,
        "constraint": base if source_type in {"exact", "unpinned"} else source_type,
        "source_type": source_type,
        "purl": purl,
        "occurrences": [{"path": path, "line": line, "evidence": _line_evidence(requirement)}],
    }


def _package_inventory(files: dict[str, str]) -> list[dict[str, Any]]:
    discovered: list[dict[str, Any]] = []
    for path, text in sorted(files.items()):
        name = PurePosixPath(path).name.lower()
        if name.startswith("requirements") and name.endswith(".txt"):
            for number, raw in enumerate(text.splitlines(), 1):
                item = _parse_requirement(raw, path, number)
                if item:
                    discovered.append(item)
        elif name == "pyproject.toml":
            try:
                payload = tomllib.loads(text)
            except (tomllib.TOMLDecodeError, ValueError):
                continue
            project = payload.get("project") if isinstance(payload, dict) else None
            if not isinstance(project, dict):
                continue
            groups: list[tuple[str, list[Any]]] = []
            dependencies = project.get("dependencies")
            if isinstance(dependencies, list):
                groups.append(("project.dependencies", dependencies))
            optional = project.get("optional-dependencies")
            if isinstance(optional, dict):
                for group, values in sorted(optional.items()):
                    if isinstance(values, list):
                        groups.append((f"project.optional-dependencies.{group}", values))
            for group, values in groups:
                for raw in values:
                    if isinstance(raw, str):
                        item = _parse_requirement(raw, path, None)
                        if item:
                            item["occurrences"][0]["section"] = group
                            discovered.append(item)

    grouped: dict[tuple[str, str | None, str, str], dict[str, Any]] = {}
    for item in discovered:
        key = (item["name"], item["version"], item["source_type"], item["constraint"])
        existing = grouped.setdefault(key, {key: value for key, value in item.items() if key != "occurrences"} | {"occurrences": []})
        existing["occurrences"].extend(item["occurrences"])
    return sorted(grouped.values(), key=lambda item: (item["name"], item["version"] or "", item["source_type"]))


def _cache_get(cache: dict[Any, tuple[float, Any]], key: Any) -> Any | None:
    with _OSV_CACHE_LOCK:
        cached = cache.get(key)
        if cached and time.monotonic() - cached[0] <= OSV_CACHE_TTL_SECONDS:
            return cached[1]
        if cached:
            cache.pop(key, None)
    return None


def _cache_put(cache: dict[Any, tuple[float, Any]], key: Any, value: Any) -> None:
    with _OSV_CACHE_LOCK:
        cache[key] = (time.monotonic(), value)


def _osv_batch_query(packages: list[tuple[str, str]]) -> list[dict[str, Any]]:
    key = tuple(packages)
    cached = _cache_get(_OSV_BATCH_CACHE, key)
    if cached is not None:
        return cached
    payload = {
        "queries": [
            {"package": {"ecosystem": "PyPI", "name": name}, "version": version}
            for name, version in packages
        ]
    }
    response = requests.post(
        f"{OSV_API}/querybatch", json=payload, timeout=REQUEST_TIMEOUT,
        headers={"User-Agent": "ModelSentry/0.4"},
    )
    response.raise_for_status()
    results = response.json().get("results")
    if not isinstance(results, list) or len(results) != len(packages):
        raise ValueError("OSV returned an unexpected batch response")
    _cache_put(_OSV_BATCH_CACHE, key, results)
    return results


def _osv_vulnerability(vulnerability_id: str) -> dict[str, Any]:
    cached = _cache_get(_OSV_VULN_CACHE, vulnerability_id)
    if cached is not None:
        return cached
    response = requests.get(
        f"{OSV_API}/vulns/{quote(vulnerability_id, safe='-')}",
        timeout=REQUEST_TIMEOUT,
        headers={"User-Agent": "ModelSentry/0.4"},
    )
    response.raise_for_status()
    payload = response.json()
    if not isinstance(payload, dict) or not payload.get("id"):
        raise ValueError("OSV returned an unexpected vulnerability record")
    _cache_put(_OSV_VULN_CACHE, vulnerability_id, payload)
    return payload


def _osv_record_ids(record: dict[str, Any]) -> set[str]:
    return {
        value for value in [record.get("id"), *(record.get("aliases") or [])]
        if isinstance(value, str) and value
    }


def _dedupe_osv_records(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
    groups: list[dict[str, Any]] = []
    for record in records:
        identities = _osv_record_ids(record)
        overlaps = [group for group in groups if identities & group["identities"]]
        if not overlaps:
            groups.append({"identities": set(identities), "records": [record]})
            continue
        primary = overlaps[0]
        primary["identities"].update(identities)
        primary["records"].append(record)
        for extra in overlaps[1:]:
            primary["identities"].update(extra["identities"])
            primary["records"].extend(extra["records"])
            groups.remove(extra)

    deduped = []
    for group in groups:
        preferred = max(
            group["records"],
            key=lambda item: (
                bool(item.get("summary")),
                bool((item.get("database_specific") or {}).get("severity")),
                str(item.get("id", "")).startswith("GHSA-"),
            ),
        )
        merged = dict(preferred)
        merged["aliases"] = sorted(group["identities"] - {str(merged.get("id"))})
        deduped.append(merged)
    return sorted(deduped, key=lambda item: str(item.get("id")))


def _osv_severity(record: dict[str, Any]) -> str:
    candidates = []
    database = record.get("database_specific") or {}
    if isinstance(database, dict):
        candidates.append(database.get("severity"))
    for affected in record.get("affected") or []:
        for key in ("database_specific", "ecosystem_specific"):
            details = affected.get(key) or {}
            if isinstance(details, dict):
                candidates.append(details.get("severity"))
    normalized = {str(value).lower() for value in candidates if value}
    for severity in ("critical", "high", "medium", "moderate", "low"):
        if severity in normalized:
            return "medium" if severity == "moderate" else severity
    return "medium"


def _osv_fixed_versions(record: dict[str, Any], package_name: str) -> list[str]:
    fixed: set[str] = set()
    for affected in record.get("affected") or []:
        package = affected.get("package") or {}
        if _canonical_package_name(str(package.get("name") or "")) != package_name:
            continue
        for item_range in affected.get("ranges") or []:
            for event in item_range.get("events") or []:
                if isinstance(event, dict) and isinstance(event.get("fixed"), str):
                    fixed.add(event["fixed"])
    return sorted(fixed)


def _correlate_osv(inventory: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[Finding], dict[str, Any]]:
    exact = sorted({(item["name"], item["version"]) for item in inventory if item.get("source_type") == "exact" and item.get("version")})
    summary = {
        "status": "not_applicable" if not exact else "complete",
        "queried_packages": 0,
        "exact_packages": len(exact),
        "vulnerability_count": 0,
        "truncated": len(exact) > MAX_OSV_PACKAGES,
    }
    if not exact:
        return [], [], summary
    queried = exact[:MAX_OSV_PACKAGES]
    summary["queried_packages"] = len(queried)
    try:
        batch = _osv_batch_query(queried)
    except Exception as exc:
        summary.update({"status": "unavailable", "error": type(exc).__name__})
        return [], [], summary

    package_ids: list[tuple[str, str, list[str]]] = []
    all_ids: list[str] = []
    for (name, version), result in zip(queried, batch):
        ids = [item.get("id") for item in (result.get("vulns") or []) if isinstance(item, dict) and item.get("id")]
        if result.get("next_page_token"):
            summary["truncated"] = True
        package_ids.append((name, version, ids))
        all_ids.extend(ids)
    unique_ids = list(dict.fromkeys(all_ids))
    if len(unique_ids) > MAX_OSV_VULNERABILITIES:
        summary["truncated"] = True
    records: dict[str, dict[str, Any]] = {}
    selected_ids = unique_ids[:MAX_OSV_VULNERABILITIES]
    failures = 0
    with concurrent.futures.ThreadPoolExecutor(max_workers=min(8, len(selected_ids) or 1)) as pool:
        future_ids = {pool.submit(_osv_vulnerability, vulnerability_id): vulnerability_id for vulnerability_id in selected_ids}
        for future, vulnerability_id in future_ids.items():
            try:
                records[vulnerability_id] = future.result()
            except Exception:
                failures += 1
    if failures or summary["truncated"]:
        summary["status"] = "partial"

    vulnerabilities: list[dict[str, Any]] = []
    findings: list[Finding] = []
    for name, version, ids in package_ids:
        package_records = _dedupe_osv_records([records[item] for item in ids if item in records and not records[item].get("withdrawn")])
        for record in package_records:
            severity = _osv_severity(record)
            fixed = _osv_fixed_versions(record, name)
            vulnerability_id = str(record.get("id"))
            aliases = [item for item in record.get("aliases") or [] if isinstance(item, str)]
            item = {
                "id": vulnerability_id,
                "aliases": aliases,
                "package": name,
                "version": version,
                "severity": severity,
                "summary": record.get("summary") or "Known vulnerability reported by OSV",
                "fixed_versions": fixed,
                "published": record.get("published"),
                "modified": record.get("modified"),
                "url": f"https://osv.dev/vulnerability/{quote(vulnerability_id, safe='-')}",
            }
            vulnerabilities.append(item)
            fix_text = f" Upgrade to {', '.join(fixed)} or later after compatibility testing." if fixed else " Review the advisory for a patched or unaffected release."
            findings.append(_finding(
                "VULN-001", severity, "Known vulnerability affects an exactly pinned dependency",
                "OSV reports that an exactly pinned package version is affected by a known vulnerability.",
                "Review exploitability in this application and update the dependency." + fix_text,
                evidence=f"{name}=={version} · {vulnerability_id}" + (f" · fixed {', '.join(fixed)}" if fixed else ""),
            ))
    vulnerabilities.sort(key=lambda item: (SEVERITY_ORDER.get(item["severity"], 99), item["package"], item["id"]))
    summary["vulnerability_count"] = len(vulnerabilities)
    return vulnerabilities, findings, summary


def _build_cyclonedx(result: ScanResult) -> dict[str, Any]:
    root_ref = f"urn:uuid:{uuid.uuid5(uuid.NAMESPACE_URL, f'https://huggingface.co/{result.repo_type}/{result.target}@{result.revision}')}"
    components = []
    refs: dict[tuple[str, str | None, str], str] = {}
    for item in result.package_inventory:
        key = (item["name"], item.get("version"), item["source_type"])
        if item["source_type"] in {"exact", "unpinned"}:
            bom_ref = item["purl"]
        else:
            digest = hashlib.sha256(f"{item['name']}|{item['source_type']}|{item['constraint']}".encode()).hexdigest()[:24]
            bom_ref = f"urn:modelsentry:dependency:{digest}"
        refs[key] = bom_ref
        component: dict[str, Any] = {
            "type": "library",
            "bom-ref": bom_ref,
            "name": item["name"],
            "purl": item["purl"],
            "properties": [
                {"name": "modelsentry:source_type", "value": item["source_type"]},
                {"name": "modelsentry:constraint", "value": item["constraint"]},
                {"name": "modelsentry:source_paths", "value": ",".join(sorted({entry["path"] for entry in item["occurrences"]}))},
            ],
        }
        if item.get("version"):
            component["version"] = item["version"]
        components.append(component)

    vulnerabilities = []
    for item in result.vulnerabilities:
        ref = refs.get((item["package"], item["version"], "exact"))
        vulnerability: dict[str, Any] = {
            "id": item["id"],
            "source": {"name": "OSV", "url": item["url"]},
            "ratings": [{"severity": item["severity"], "method": "other"}],
            "description": item["summary"],
            "recommendation": "Upgrade to a reviewed fixed version." if item["fixed_versions"] else "Review the advisory and select a patched or unaffected version.",
        }
        if ref:
            vulnerability["affects"] = [{"ref": ref}]
        if item.get("published"):
            vulnerability["published"] = item["published"]
        if item.get("modified"):
            vulnerability["updated"] = item["modified"]
        vulnerabilities.append(vulnerability)

    bom: dict[str, Any] = {
        "$schema": "https://cyclonedx.org/schema/bom-1.6.schema.json",
        "bomFormat": "CycloneDX",
        "specVersion": "1.6",
        "serialNumber": root_ref,
        "version": 1,
        "metadata": {
            "component": {
                "type": "application",
                "bom-ref": root_ref,
                "name": result.target,
                "version": result.revision,
                "properties": [
                    {"name": "modelsentry:repository_type", "value": result.repo_type},
                    {"name": "modelsentry:static_coverage", "value": str(result.artifacts.get("static_coverage") or "unknown")},
                ],
            },
        },
        "components": components,
        "dependencies": [{"ref": root_ref, "dependsOn": [component["bom-ref"] for component in components]}],
    }
    if vulnerabilities:
        bom["vulnerabilities"] = vulnerabilities
    return bom


def _docker_findings(path: str, text: str) -> Iterable[Finding]:
    lines = text.splitlines()
    if not any(re.match(r"\s*USER\s+\S+", line, re.I) for line in lines):
        yield _finding(
            "CTR-001", "medium", "Container has no non-root USER declaration",
            "The final container stage may run with root privileges.",
            "Create an unprivileged user and set it in the final image stage.", path,
        )
    for number, line in enumerate(lines, 1):
        match = re.match(r"\s*FROM\s+([^\s]+)", line, re.I)
        if match and "@sha256:" not in match.group(1):
            yield _finding(
                "CTR-002", "low", "Container base image is not digest-pinned",
                "A mutable image tag can resolve to different content later.",
                "Pin the reviewed base image by digest and update it deliberately.",
                path, number, _line_evidence(line),
            )


def _python_syntax_finding(path: str, text: str) -> Finding | None:
    if not path.endswith(".py"):
        return None
    try:
        # Parsing is static, but Python can still emit SyntaxWarning text derived
        # from untrusted source. Keep repository contents out of service logs.
        with warnings.catch_warnings():
            warnings.simplefilter("ignore")
            ast.parse(text, filename="<untrusted-source>")
    except SyntaxError as exc:
        return _finding(
            "QUAL-001", "low", "Python source could not be parsed",
            "Static analysis may be incomplete because the file is not valid for this Python parser.",
            "Verify the file encoding and intended Python version.", path, exc.lineno, confidence="medium",
        )
    return None


def _load_controls(files: dict[str, str]) -> dict[str, Any]:
    text = files.get(".modelsentry.json")
    if not text:
        return {}
    try:
        payload = json.loads(text)
    except json.JSONDecodeError:
        return {}
    if not isinstance(payload, dict):
        return {}
    domains = payload.get("expected_outbound_domains", [])
    if not isinstance(domains, list):
        domains = []
    payload["expected_outbound_domains"] = [
        value.lower().strip()
        for value in domains
        if isinstance(value, str) and re.fullmatch(r"[A-Za-z0-9.-]+", value.strip())
    ]
    return payload


def human_bytes(value: int | None) -> str:
    if not value:
        return "0 B"
    amount = float(value)
    for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
        if amount < 1024 or unit == "TiB":
            return f"{amount:.2f} {unit}" if unit != "B" else f"{int(amount)} B"
        amount /= 1024
    return f"{amount:.2f} TiB"


def _dependency_from_tag(tag: str) -> tuple[str, str] | None:
    if tag.startswith("base_model:"):
        remainder = tag[len("base_model:"):]
        relation = "base_model"
        if ":" in remainder:
            possible_relation, possible_repo = remainder.split(":", 1)
            if "/" in possible_repo:
                relation, remainder = possible_relation, possible_repo
        if re.fullmatch(r"[A-Za-z0-9._-]+/[A-Za-z0-9._-]+", remainder):
            return remainder, relation
    if tag.startswith("dataset:"):
        remainder = tag[len("dataset:"):]
        if re.fullmatch(r"[A-Za-z0-9._-]+/[A-Za-z0-9._-]+", remainder):
            return remainder, "dataset"
    return None


def _extract_dependencies(metadata: dict[str, Any], readme: str) -> tuple[list[str], dict[str, set[str]], set[str]]:
    card_data = metadata.get("cardData") or metadata.get("card_data") or {}
    relations: dict[str, set[str]] = {}

    def add(repo: str, relation: str) -> None:
        if re.fullmatch(r"[A-Za-z0-9._-]+/[A-Za-z0-9._-]+", repo):
            relations.setdefault(repo, set()).add(relation)

    for key, relation in (("base_model", "base_model"), ("models", "model"), ("datasets", "dataset")):
        values = card_data.get(key, [])
        if isinstance(values, str):
            values = [values]
        if isinstance(values, list):
            for value in values:
                if value:
                    add(str(value), relation)
    for tag in metadata.get("tags") or []:
        if isinstance(tag, str):
            parsed = _dependency_from_tag(tag)
            if parsed:
                add(*parsed)

    documented_sources: set[str] = set()
    source_pattern = re.compile(
        r"(?i)(?:extracted|derived|converted|fine[- ]?tuned)\s+from\s+"
        r"(?:\[[^\]]+\]\()?https://huggingface\.co/([A-Za-z0-9._-]+/[A-Za-z0-9._-]+)"
    )
    for match in source_pattern.finditer(readme):
        repo = match.group(1)
        documented_sources.add(repo)
        relations.setdefault(repo, set()).add("documented_source")
    return sorted(relations), relations, documented_sources


def _artifact_inventory(metadata: dict[str, Any], files: dict[str, str]) -> tuple[dict[str, Any], list[Finding]]:
    siblings = metadata.get("siblings") or []
    all_paths: set[str] = set()
    sizes: dict[str, int] = {}
    format_counts: Counter[str] = Counter()
    weight_paths: list[str] = []
    total_bytes = 0
    weight_bytes = 0
    findings: list[Finding] = []

    for sibling in siblings:
        path = sibling.get("rfilename") or sibling.get("path")
        if not path:
            continue
        all_paths.add(path)
        raw_size = sibling.get("size")
        if not isinstance(raw_size, int):
            raw_size = (sibling.get("lfs") or {}).get("size")
        size = int(raw_size) if isinstance(raw_size, int) else 0
        sizes[path] = size
        total_bytes += size
        suffix = PurePosixPath(path).suffix.lower()
        if suffix:
            format_counts[suffix.lstrip(".")] += 1
        if suffix in WEIGHT_SUFFIXES:
            weight_paths.append(path)
            weight_bytes += size

    declared_tensor_bytes = 0
    referenced_shards: set[str] = set()
    index_files = [path for path in files if path.endswith(".safetensors.index.json")]
    for index_path in index_files:
        try:
            payload = json.loads(files[index_path])
            declared = (payload.get("metadata") or {}).get("total_size")
            if isinstance(declared, int):
                declared_tensor_bytes += declared
            weight_map = payload.get("weight_map") or {}
            parent = PurePosixPath(index_path).parent
            for shard in weight_map.values():
                if not isinstance(shard, str):
                    continue
                resolved = str(parent / shard) if str(parent) != "." and "/" not in shard else shard
                referenced_shards.add(resolved)
        except (json.JSONDecodeError, TypeError, AttributeError):
            findings.append(_finding(
                "ART-001", "high", "Safetensors index is not valid JSON",
                "The weight index could not be parsed, so shard completeness cannot be established.",
                "Regenerate the index and verify every referenced shard before publishing.", index_path,
            ))

    missing_shards = sorted(referenced_shards - all_paths)
    if missing_shards:
        findings.append(_finding(
            "ART-002", "high", "Safetensors index references missing shards",
            f"The index references {len(missing_shards)} shard files that are absent from the repository.",
            "Upload every referenced shard or regenerate the index from the published artifact set.",
            index_files[0] if index_files else None,
            evidence=", ".join(missing_shards[:3]),
        ))

    architecture = None
    for path, text in files.items():
        if PurePosixPath(path).name != "config.json":
            continue
        try:
            config = json.loads(text)
        except json.JSONDecodeError:
            continue
        architectures = config.get("architectures")
        architecture = config.get("_class_name") or config.get("model_type")
        if not architecture and isinstance(architectures, list) and architectures:
            architecture = architectures[0]
        if architecture:
            break

    inventory = {
        "repository_file_count": len(all_paths),
        "repository_bytes": total_bytes,
        "repository_size": human_bytes(total_bytes),
        "weight_file_count": len(weight_paths),
        "weight_bytes": weight_bytes,
        "weight_size": human_bytes(weight_bytes),
        "weight_formats": dict(sorted((key, value) for key, value in format_counts.items() if f".{key}" in WEIGHT_SUFFIXES)),
        "safetensors_index_count": len(index_files),
        "index_referenced_shards": len(referenced_shards),
        "index_declared_tensor_bytes": declared_tensor_bytes,
        "index_declared_tensor_size": human_bytes(declared_tensor_bytes),
        "missing_shards": missing_shards,
        "architecture": architecture,
    }
    return inventory, findings


def _model_documentation_findings(
    repo_id: str,
    metadata: dict[str, Any],
    readme: str,
    declared_dependencies: set[str],
    documented_sources: set[str],
    artifacts: dict[str, Any],
) -> Iterable[Finding]:
    if not readme:
        yield _finding(
            "DOC-002", "high", "Model card is missing",
            "The repository has no README model card describing purpose, operation, limitations, or provenance.",
            "Add a model card with intended use, limitations, provenance, licensing, safety, and tested usage.",
        )
        return

    card_data = metadata.get("cardData") or {}
    pipeline = metadata.get("pipeline_tag") or card_data.get("pipeline_tag")
    if not pipeline:
        yield _finding(
            "DOC-003", "medium", "Model task or pipeline is not declared",
            "Hub clients cannot reliably determine the model's intended inference task.",
            "Add an appropriate pipeline_tag to the model-card metadata.", "README.md",
        )

    lower = readme.lower()
    headings = [
        re.sub(r"[*_`]", "", match.group(1)).strip().lower()
        for match in re.finditer(r"(?m)^#{1,6}\s+(.+?)\s*$", readme)
    ]
    intended_terms = (
        "intended use", "intended uses", "intended usage", "evaluated use",
        "use case", "use cases", "uses", "usage", "applications", "model use",
    )
    limitation_terms = ("limitation", "risk", "bias", "known issue", "out-of-scope", "out of scope", "safety", "misuse")
    has_intended_use = any(any(term in heading for term in intended_terms) for heading in headings)
    has_limitations = any(any(term in heading for term in limitation_terms) for heading in headings)
    if not has_intended_use or not has_limitations:
        missing = " and ".join(
            label for present, label in ((has_intended_use, "intended use"), (has_limitations, "limitations")) if not present
        )
        yield _finding(
            "DOC-004", "medium", "Model card lacks expected usage boundaries",
            f"No clear {missing} section was found.",
            "Document supported uses, unsupported uses, known failure modes, and evaluation limits.", "README.md",
        )

    audience_tag = any(str(tag).lower() == "not-for-all-audiences" for tag in metadata.get("tags") or [])
    safety_terms = ("content warning", "responsible use", "safety", "adult-only", "18+", "age restriction")
    if audience_tag and not any(term in lower for term in safety_terms):
        yield _finding(
            "DOC-005", "medium", "Audience-restricted model lacks a clear safety notice",
            "The repository is tagged not-for-all-audiences but the model card does not clearly state content risks or responsible-use boundaries.",
            "Add a prominent content warning, intended audience, prohibited uses, and relevant safety limitations.", "README.md",
        )

    call_pattern = re.compile(
        r"(?:from_pretrained|snapshot_download|hf_hub_download)\s*\(\s*[\"']"
        r"([A-Za-z0-9._-]+/[A-Za-z0-9._-]+)[\"']",
        re.MULTILINE,
    )
    for match in call_pattern.finditer(readme):
        referenced = match.group(1)
        if referenced != repo_id and referenced.split("/", 1)[1] == repo_id.split("/", 1)[1]:
            line = readme[:match.start(1)].count("\n") + 1
            yield _finding(
                "DOC-006", "medium", "Usage example references a different repository owner",
                f"The example loads {referenced}, although this repository is {repo_id}.",
                f"Change the example to {repo_id} or explain why the alternate repository is required.",
                "README.md", line, referenced,
            )

    undeclared_sources = documented_sources - declared_dependencies
    for source in sorted(undeclared_sources):
        yield _finding(
            "PROV-001", "medium", "Documented source is missing from declared provenance",
            f"The model card says weights were derived or extracted from {source}, but Hub base-model metadata does not declare it.",
            "Declare the immediate source model in base_model metadata so the provenance chain is machine-readable.",
            "README.md", evidence=source,
        )

    if artifacts.get("weight_bytes", 0) >= 2 * 1024**3 and not any(
        term in lower for term in ("vram", "gpu memory", "memory requirement", "hardware requirement", "minimum gpu")
    ):
        yield _finding(
            "DOC-007", "low", "Large model has no quantitative hardware guidance",
            f"Published weights occupy approximately {artifacts.get('weight_size')}, but no VRAM or memory requirement was found.",
            "Document tested precision, minimum/recommended VRAM, system RAM, and expected inference hardware.", "README.md",
        )


def _metadata_summary(metadata: dict[str, Any], artifacts: dict[str, Any]) -> dict[str, Any]:
    card_data = metadata.get("cardData") or {}
    return {
        "license": card_data.get("license") or metadata.get("license"),
        "library": metadata.get("library_name") or card_data.get("library_name"),
        "pipeline_tag": metadata.get("pipeline_tag") or card_data.get("pipeline_tag"),
        "gated": metadata.get("gated", False),
        "private": metadata.get("private", False),
        "architecture": artifacts.get("architecture"),
    }


def _build_checks(result: ScanResult, files: dict[str, str]) -> list[CheckResult]:
    rules = {
        finding.rule_id for finding in result.findings
        if finding.status == "unresolved" and finding.severity != "info"
    }

    def status(category: str, prefixes: tuple[str, ...], detail: str, applies: bool = True) -> CheckResult:
        if not applies:
            return CheckResult(category, "not_applicable", detail)
        attention = any(rule.startswith(prefix) for rule in rules for prefix in prefixes)
        if result.skipped_files and category == "Static application code" and applies:
            return CheckResult(category, "partial", detail + f" {result.skipped_files} eligible files were not inspected.")
        return CheckResult(category, "attention" if attention else "passed", detail)

    has_code = any(PurePosixPath(path).suffix.lower() in EXECUTABLE_SUFFIXES or PurePosixPath(path).name == "Dockerfile" for path in files)
    has_manifest = any(PurePosixPath(path).name.lower() in MANIFEST_NAMES for path in files)
    has_weights = bool(result.artifacts.get("weight_file_count"))
    checks = [
        status("Repository metadata", ("LIC-", "DOC-003"), "License and task declarations were checked."),
        status("Model card / documentation", ("DOC-",), "Model-card completeness and usage examples were checked.", result.repo_type == "model"),
        status("Artifact inventory", ("ART-",), "Published file formats, sizes, and safetensors index references were checked."),
        status("Static application code", ("CODE-", "NET-", "SECRET-", "DATA-", "ENV-", "TEL-", "CTR-"), "No executable application files were present." if not has_code else "Allowlisted source files were statically inspected.", has_code),
        status("Dependency manifests", ("DEP-",), "No supported dependency manifest was present." if not has_manifest else "Supported dependency manifests were inspected.", has_manifest),
        status("Provenance and upstream licenses", ("PROV-", "LIC-UPSTREAM-"), "Declared and documented upstream repositories were inspected."),
    ]
    if has_weights:
        checks.append(CheckResult("Binary weight contents", "not_checked", "Weight tensors were inventoried but never downloaded or parsed."))
    checks.append(CheckResult("Runtime behavior", "not_checked", "Static analysis does not execute or observe the repository at runtime."))
    osv = result.osv_summary
    if osv.get("status") == "not_checked":
        checks.append(CheckResult("Known vulnerabilities", "not_checked", "OSV correlation has not been run for this result."))
    elif osv.get("status") == "not_applicable":
        checks.append(CheckResult("Known vulnerabilities", "not_applicable", "No exactly pinned PyPI versions were available for a defensible OSV query."))
    elif osv.get("status") == "unavailable":
        checks.append(CheckResult("Known vulnerabilities", "not_checked", "OSV was unavailable; no vulnerability conclusion was made."))
    elif result.vulnerabilities:
        qualifier = " Results may be incomplete because configured query limits were reached." if osv.get("status") == "partial" else ""
        checks.append(CheckResult("Known vulnerabilities", "attention", f"OSV reported {len(result.vulnerabilities)} known vulnerabilities affecting exact pins.{qualifier}"))
    elif osv.get("status") == "partial":
        checks.append(CheckResult("Known vulnerabilities", "partial", "OSV completed only a bounded partial query; absence of a match is not a clean verdict."))
    else:
        checks.append(CheckResult("Known vulnerabilities", "passed", f"OSV returned no known matches for {osv.get('queried_packages', 0)} exactly pinned packages at scan time."))
    checks.append(CheckResult(
        "CycloneDX SBOM",
        "passed" if result.package_inventory else "not_applicable",
        f"A CycloneDX 1.6 inventory was generated for {len(result.package_inventory)} declared components." if result.package_inventory else "No supported package declarations were available for an SBOM component inventory.",
    ))
    return checks


def _fetch_dependency_details(
    repo_id: str,
    metadata: dict[str, Any],
    readme: str,
    current_license: str | None,
) -> tuple[list[dict[str, Any]], list[Finding]]:
    _, initial_relations, _ = _extract_dependencies(metadata, readme)
    queue: list[tuple[str, str, set[str], int]] = []
    for dependency, relations in initial_relations.items():
        dep_type = "dataset" if relations == {"dataset"} else "model"
        queue.append((dependency, dep_type, set(relations), 0))

    details: list[dict[str, Any]] = []
    findings: list[Finding] = []
    seen: set[tuple[str, str]] = set()
    permissive = {"apache-2.0", "mit", "bsd-2-clause", "bsd-3-clause", "cc-by-4.0"}
    noncommercial = {"cc-by-nc-4.0", "cc-by-nc-sa-4.0", "non-commercial", "noncommercial"}

    while queue and len(seen) < 12:
        dependency, dep_type, relations, depth = queue.pop(0)
        key = (dependency, dep_type)
        if key in seen or dependency == repo_id:
            continue
        seen.add(key)
        try:
            dep_meta = _get_json(f"{HF_API}/{_api_kind(dep_type)}/{quote(dependency, safe='/')}")
        except Exception:
            details.append({
                "repository": dependency,
                "type": dep_type,
                "relation": sorted(relations),
                "depth": depth + 1,
                "status": "unavailable",
                "license": None,
                "revision": None,
            })
            findings.append(_finding(
                "PROV-002", "low", "Upstream repository metadata is unavailable",
                f"{PRODUCT_NAME} could not retrieve public metadata for {dependency}.",
                "Verify that the dependency exists, is public, and is named correctly.",
                evidence=dependency, confidence="medium",
            ))
            continue

        dep_card = dep_meta.get("cardData") or {}
        dep_license = dep_card.get("license") or dep_meta.get("license")
        details.append({
            "repository": dependency,
            "type": dep_type,
            "relation": sorted(relations),
            "depth": depth + 1,
            "status": "available",
            "license": dep_license,
            "revision": dep_meta.get("sha"),
            "gated": dep_meta.get("gated", False),
        })
        if not dep_license:
            findings.append(_finding(
                "LIC-UPSTREAM-001", "medium", "Upstream dependency has no declared license",
                f"The license for {dependency} could not be established from Hub metadata.",
                "Review the upstream repository and document compatible redistribution terms.",
                evidence=dependency,
            ))
        elif str(dep_license).lower() in noncommercial and str(current_license).lower() in permissive:
            findings.append(_finding(
                "LIC-UPSTREAM-002", "high", "Upstream license may conflict with the declared license",
                f"{dependency} declares {dep_license}, while this repository declares {current_license}.",
                "Do not represent the combined work as permissively licensed until the upstream restriction is resolved.",
                evidence=dependency,
            ))

        if depth < 1 and dep_type == "model":
            transitive, transitive_relations, _ = _extract_dependencies(dep_meta, "")
            for child in transitive:
                child_relations = set(transitive_relations[child])
                child_type = "dataset" if child_relations == {"dataset"} else "model"
                child_relations.add("transitive")
                queue.append((child, child_type, child_relations, depth + 1))

    return details, findings


def analyze_files(
    repo_id: str,
    repo_type: str,
    revision: str,
    files: dict[str, str],
    metadata: dict[str, Any] | None = None,
    skipped_files: int = 0,
) -> ScanResult:
    metadata = metadata or {}
    result = ScanResult(repo_id, repo_type, revision, inspected_files=sorted(files), skipped_files=skipped_files)

    card_data = metadata.get("cardData") or metadata.get("card_data") or {}
    controls = _load_controls(files)
    readme = files.get("README.md", "")
    artifacts, artifact_findings = _artifact_inventory(metadata, files)
    result.artifacts = artifacts
    result.metadata_summary = _metadata_summary(metadata, artifacts)
    result.findings.extend(artifact_findings)

    dependencies, relations, documented_sources = _extract_dependencies(metadata, readme)
    result.dependencies = dependencies
    declared_dependencies = {
        repo for repo, kinds in relations.items() if any(kind != "documented_source" for kind in kinds)
    }
    if repo_type == "model":
        result.findings.extend(_model_documentation_findings(
            repo_id, metadata, readme, declared_dependencies, documented_sources, artifacts
        ))

    license_name = card_data.get("license") or metadata.get("license")
    has_license_file = any(PurePosixPath(path).name.lower().startswith("license") for path in files)
    if not license_name and not has_license_file:
        result.findings.append(_finding(
            "LIC-001", "high", "No license declaration found",
            "Users cannot reliably determine reuse, redistribution, or commercial-use rights.",
            "Add an SPDX-recognized license to the repository metadata and include the full license text.",
        ))
    elif str(license_name).lower() in {"other", "unknown"}:
        result.findings.append(_finding(
            "LIC-002", "medium", "Custom or ambiguous license declaration",
            "Automated compatibility checks cannot determine the granted rights.",
            "Name the governing license clearly and summarize important restrictions in the repository card.",
        ))
    if str(license_name).lower() in {"cc-by-nc-4.0", "cc-by-nc-sa-4.0", "non-commercial", "noncommercial"}:
        result.findings.append(_finding(
            "LIC-003", "high", "Noncommercial license detected",
            "The declared license restricts commercial use and may affect downstream applications.",
            "Surface the restriction prominently and verify compatibility with every downstream use.",
        ))

    readme_lower = readme.lower()
    if repo_type == "space" and not any(term in readme_lower for term in ("privacy", "retention", "stored", "logging")):
        result.findings.append(_finding(
            "DOC-001", "medium", "No obvious privacy or retention disclosure",
            "Users may not know whether submitted text, files, or media leave the Space or are retained.",
            "Document external destinations, logging, retention, deletion, and sensitive-data limitations.",
            "README.md" if readme else None,
        ))

    for path, text in files.items():
        result.findings.extend(_line_findings(path, text))
        name = PurePosixPath(path).name.lower()
        if path.endswith(".py"):
            result.findings.extend(_python_ast_findings(path, text, controls))
        if name.startswith("requirements") and name.endswith(".txt"):
            result.findings.extend(_requirements_findings(path, text))
        if name == "dockerfile":
            result.findings.extend(_docker_findings(path, text))
        syntax = _python_syntax_finding(path, text)
        if syntax:
            result.findings.append(syntax)

    result.package_inventory = _package_inventory(files)

    deduped: dict[tuple[Any, ...], Finding] = {}
    for finding in result.findings:
        key = (finding.rule_id, finding.path, finding.line, finding.title)
        deduped[key] = finding
    result.findings = list(deduped.values())
    if skipped_files:
        result.notes.append(f"{skipped_files} candidate files were skipped because of scan limits or size caps.")
    result.notes.append("Static inspection cannot prove that a repository is safe or describe runtime behavior completely.")
    result.sbom = _build_cyclonedx(result)
    result.checks = _build_checks(result, files)
    return result


def scan_repository(target: str) -> ScanResult:
    repo_id, repo_type = parse_target(target)
    metadata = _get_json(f"{HF_API}/{_api_kind(repo_type)}/{quote(repo_id, safe='/')}?blobs=true")
    revision = metadata.get("sha")
    if not revision or not re.fullmatch(r"[0-9a-f]{40,64}", revision):
        raise ScanError("The Hub did not return an immutable repository revision.")

    candidates: list[tuple[str, int | None]] = []
    for sibling in metadata.get("siblings") or []:
        path = sibling.get("rfilename") or sibling.get("path")
        if path and _safe_path(path):
            size = sibling.get("size")
            candidates.append((path, int(size) if isinstance(size, int) else None))
    candidates.sort(key=_candidate_priority)

    selected = candidates[:MAX_FILES]
    files: dict[str, str] = {}
    skipped = max(0, len(candidates) - len(selected))
    for path, size in selected:
        text = _download_text(repo_id, repo_type, revision, path, size)
        if text is None:
            skipped += 1
        else:
            files[path] = text
    result = analyze_files(repo_id, repo_type, revision, files, metadata, skipped)
    result.artifacts.update({
        "eligible_text_files": len(candidates),
        "inspected_text_files": len(files),
        "skipped_eligible_files": skipped,
        "static_coverage": "partial" if skipped else "complete",
    })
    result.checks = _build_checks(result, files)
    if repo_type == "model":
        card_data = metadata.get("cardData") or {}
        details, upstream_findings = _fetch_dependency_details(
            repo_id,
            metadata,
            files.get("README.md", ""),
            card_data.get("license") or metadata.get("license"),
        )
        result.dependency_details = details
        result.findings.extend(upstream_findings)
        deduped: dict[tuple[Any, ...], Finding] = {}
        for finding in result.findings:
            deduped[(finding.rule_id, finding.path, finding.line, finding.title, finding.evidence)] = finding
        result.findings = list(deduped.values())
    result.vulnerabilities, vulnerability_findings, result.osv_summary = _correlate_osv(result.package_inventory)
    result.findings.extend(vulnerability_findings)
    deduped: dict[tuple[Any, ...], Finding] = {}
    for finding in result.findings:
        deduped[(finding.rule_id, finding.path, finding.line, finding.title, finding.evidence)] = finding
    result.findings = list(deduped.values())
    if result.osv_summary.get("status") == "unavailable":
        result.notes.append("OSV was unavailable, so exact package versions were not checked for known vulnerabilities.")
    elif result.osv_summary.get("truncated"):
        result.notes.append("OSV correlation reached a configured query limit; vulnerability results are partial.")
    result.sbom = _build_cyclonedx(result)
    result.checks = _build_checks(result, files)
    return result


def result_json(result: ScanResult) -> str:
    return json.dumps(result.to_dict(), indent=2, ensure_ascii=False)