AID-H-003.009: Model Artifact Static Malware & Unsafe-Serialization Scanning

Version 1.20261015 · Actionable sub-technique

Statically scan model artifacts, checkpoints, adapters, and packaged model bundles before they are imported, mirrored, loaded, promoted, or deployed. The control inspects serialized model bytes and archive members for unsafe deserialization primitives, dangerous pickle globals, embedded code, custom loader requirements, unexpected scripts, and model-format policy violations without executing the artifact.

Pillars: model, infra

Lifecycle: building, validation, operation

Scope boundary

Owns format-aware static malware, unsafe-serialization, archive, and loader-hazard scanning of model artifacts before any model loader executes them. Its clean verdict is admission evidence, not proof of safe execution.

AID-H-003.001: AID-H-003.009 scans model artifact formats and serialized objects; AID-H-003.001 evaluates third-party software dependencies and packages. Model-file loader hazards require dedicated format semantics beyond package SCA.

AID-H-003.006: AID-H-003.009 determines whether the artifact contains known unsafe serialization or malware indicators; AID-H-003.006 verifies the artifact's declared composition and provenance. A trusted attestation does not make unsafe bytes executable, and a clean scan does not establish provenance.

AID-I-001.005: AID-H-003.009 performs static analysis before any loader runs; AID-I-001.005 detonates and observes first load inside an isolated sandbox. Dynamic loading can reveal behavior that static scanners miss, while sandbox execution must not replace pre-load rejection.

AID-H-025.001: AID-H-003.009 analyzes serialized model and adapter artifacts; AID-H-025.001 analyzes generated code and execution requests for dangerous constructs. Both are pre-execution gates, but they operate on different artifact languages and loaders.

AID-H-002.009: AID-H-003.009 statically scans models, checkpoints, and adapters for unsafe serialization and embedded code; AID-H-002.009 admits untrusted dataset and structured-data artifacts against parser-specific features. Route each object by its actual format and loader; a model-artifact scan does not make a dataset parser safe, and parser admission does not approve model loading.

Warning

Static Scanning Is Necessary but Not Sufficient

Pickle and model-artifact scanners are a required admission layer, but they are not a proof of safe execution. 2025 scanner-bypass disclosures against pickle/model scanners, including CVE-2025-1716, CVE-2025-1944, CVE-2025-1945, CVE-2025-10155 through CVE-2025-10157, and nullifAI-style bypass techniques, show that malformed pickle opcodes, archive tricks, and unsupported loader paths can evade static tools. Treat a clean scan as one input to promotion only: require tensor-only formats where possible, signed provenance, offline loading, dynamic first-load sandboxing for pickle-like artifacts, and a rapid scanner-update process.

Implementation guidance

AID-H-003.009-G001

Run static model-file scanners before any model import, mirror promotion, registry acceptance, or deployment, and produce a signed admission verdict that release gates can consume.
Delivery level: production artifact-admission job

Scan one deployment-controlled byte snapshot before any deserializer or framework loader sees it. The scanner, artifact digest, report, and admitted copy must all refer to those same bytes; scanning a mutable download path and hashing it later creates a verification-to-use race.

Step 1: Verify the scanner profile and capture one immutable snapshot
# File: supply_chain/model_static_admission.py
from __future__ import annotations

import argparse
import hashlib
import json
import math
import os
import re
import shutil
import stat
import subprocess
import tempfile
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

POLICY_KEY = Path("/opt/aidefend/trust/model-artifact-profile.pub")
SHA256 = re.compile(r"[0-9a-f]{64}")


def strict_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    value: dict[str, Any] = {}
    for key, item in pairs:
        if key in value:
            raise ValueError(f"duplicate JSON key: {key}")
        value[key] = item
    return value


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def run(args: list[str], timeout_seconds: float) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        args, check=False, capture_output=True, text=True, timeout=timeout_seconds
    )


def load_policy(
    path: Path, bundle: Path, timeout_seconds: float
) -> tuple[dict[str, Any], str]:
    payload = path.read_bytes()
    with tempfile.NamedTemporaryFile(prefix="model-profile-", delete=False) as handle:
        handle.write(payload)
        handle.flush()
        os.fsync(handle.fileno())
        snapshot = Path(handle.name)
    try:
        subprocess.run(
            [
                "cosign", "verify-blob", "--key", str(POLICY_KEY),
                "--bundle", str(bundle), str(snapshot),
            ],
            check=True, capture_output=True, text=True, timeout=timeout_seconds,
        )
    finally:
        snapshot.unlink(missing_ok=True)
    profile = json.loads(
        payload.decode("utf-8", errors="strict"),
        object_pairs_hook=strict_object,
        parse_constant=lambda value: (_ for _ in ()).throw(ValueError(f"non-finite JSON number: {value}")),
    )
    fields = {
        "schema_version", "policy_version", "allowed_extensions",
        "pickle_like_extensions", "max_artifact_bytes",
    }
    if not isinstance(profile, dict) or set(profile) != fields:
        raise ValueError("artifact profile schema mismatch")
    if profile["schema_version"] != "aidefend.model-artifact-profile.v1":
        raise ValueError("unsupported artifact profile")
    if not isinstance(profile["policy_version"], str) or not profile["policy_version"]:
        raise ValueError("policy_version is required")
    for field in ("allowed_extensions", "pickle_like_extensions"):
        values = profile[field]
        if (
            not isinstance(values, list) or not values
            or len(values) != len(set(values))
            or any(not isinstance(value, str) or not value.startswith(".") or value != value.lower() for value in values)
        ):
            raise ValueError(f"invalid {field}")
    if not set(profile["pickle_like_extensions"]).issubset(profile["allowed_extensions"]):
        raise ValueError("pickle-like extensions must be allowed only through this scan path")
    if type(profile["max_artifact_bytes"]) is not int or profile["max_artifact_bytes"] < 1:
        raise ValueError("max_artifact_bytes must be positive")
    profile["allowed_extensions"] = set(profile["allowed_extensions"])
    profile["pickle_like_extensions"] = set(profile["pickle_like_extensions"])
    return profile, hashlib.sha256(payload).hexdigest()


def capture_snapshot(source: Path, destination: Path, maximum_bytes: int) -> str:
    flags = os.O_RDONLY | getattr(os, "O_BINARY", 0)
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    descriptor = os.open(source, flags)
    try:
        opened = os.fstat(descriptor)
        if not stat.S_ISREG(opened.st_mode) or opened.st_size < 1 or opened.st_size > maximum_bytes:
            raise ValueError("artifact must be one non-empty regular file within the signed size cap")
        digest = hashlib.sha256()
        with os.fdopen(descriptor, "rb", closefd=False) as source_handle, destination.open("xb") as target:
            remaining = opened.st_size
            while remaining:
                chunk = source_handle.read(min(1024 * 1024, remaining))
                if not chunk:
                    raise RuntimeError("artifact ended while snapshot was being captured")
                target.write(chunk)
                digest.update(chunk)
                remaining -= len(chunk)
            if source_handle.read(1):
                raise RuntimeError("artifact grew while snapshot was being captured")
            target.flush()
            os.fsync(target.fileno())
        after = os.fstat(descriptor)
        if (opened.st_dev, opened.st_ino, opened.st_size, opened.st_mtime_ns) != (
            after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns
        ):
            raise RuntimeError("artifact changed while snapshot was being captured")
        os.chmod(destination, 0o400)
        return digest.hexdigest()
    finally:
        os.close(descriptor)


@dataclass(frozen=True)
class StaticVerdict:
    schema_version: str
    artifact_sha256: str
    artifact_size: int
    extension: str
    policy_version: str
    policy_sha256: str
    modelscan_version: str
    fickling_version: str | None
    decision: str
    reasons: list[str]
    observed_at: str


def admit(
    source: Path, admitted_path: Path, report_path: Path, profile: dict[str, Any],
    policy_sha256: str, timeout_seconds: float, report_limit: int,
) -> StaticVerdict:
    extension = source.suffix.lower()
    if extension not in profile["allowed_extensions"]:
        raise PermissionError(f"extension is not allowed: {extension}")
    admitted_path.parent.mkdir(parents=True, exist_ok=True)
    if admitted_path.exists():
        raise FileExistsError("admitted output already exists")
    with tempfile.TemporaryDirectory(prefix="model-admission-", dir=admitted_path.parent) as directory:
        snapshot = Path(directory) / ("artifact" + extension)
        artifact_sha256 = capture_snapshot(source, snapshot, profile["max_artifact_bytes"])
        modelscan_version_result = run(["modelscan", "--version"], timeout_seconds)
        if modelscan_version_result.returncode != 0:
            raise RuntimeError("cannot identify ModelScan release")
        modelscan_report = Path(directory) / "modelscan.json"
        modelscan = run(
            ["modelscan", "-p", str(snapshot), "-r", "json", "-o", str(modelscan_report)],
            timeout_seconds,
        )
        reasons = []
        if modelscan.returncode != 0:
            reasons.append(f"modelscan_blocked_or_failed:{modelscan.returncode}")
        fickling_version = None
        if extension in profile["pickle_like_extensions"]:
            version = run(["fickling", "--version"], timeout_seconds)
            if version.returncode != 0:
                raise RuntimeError("cannot identify Fickling release")
            fickling_version = (version.stdout or version.stderr).strip()[:report_limit]
            fickling = run(["fickling", "--check-safety", "-p", str(snapshot)], timeout_seconds)
            if fickling.returncode != 0:
                reasons.append(f"fickling_blocked_or_failed:{fickling.returncode}")
        verdict = StaticVerdict(
            schema_version="aidefend.model-static-admission.v1",
            artifact_sha256=artifact_sha256,
            artifact_size=snapshot.stat().st_size,
            extension=extension,
            policy_version=profile["policy_version"],
            policy_sha256=policy_sha256,
            modelscan_version=(modelscan_version_result.stdout or modelscan_version_result.stderr).strip()[:report_limit],
            fickling_version=fickling_version,
            decision="BLOCK" if reasons else "ALLOW",
            reasons=reasons,
            observed_at=datetime.now(timezone.utc).isoformat(),
        )
        report_path.parent.mkdir(parents=True, exist_ok=True)
        report_path.write_text(json.dumps(asdict(verdict), indent=2, sort_keys=True) + "\n", encoding="utf-8")
        if reasons:
            return verdict
        with snapshot.open("rb") as source_handle, admitted_path.open("xb") as target:
            shutil.copyfileobj(source_handle, target, length=1024 * 1024)
            target.flush()
            os.fsync(target.fileno())
        os.chmod(admitted_path, 0o400)
        if sha256_file(admitted_path) != artifact_sha256:
            admitted_path.unlink(missing_ok=True)
            raise RuntimeError("admitted copy differs from scanned snapshot")
        return verdict


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--artifact", type=Path, required=True)
    parser.add_argument("--profile", type=Path, required=True)
    parser.add_argument("--profile-bundle", type=Path, required=True)
    parser.add_argument("--admitted-artifact", type=Path, required=True)
    parser.add_argument("--report", type=Path, required=True)
    parser.add_argument("--timeout-seconds", type=float, required=True)
    parser.add_argument("--report-max-characters", type=int, required=True)
    args = parser.parse_args()
    if not math.isfinite(args.timeout_seconds) or args.timeout_seconds <= 0 or args.report_max_characters < 1:
        raise ValueError("runtime limits are invalid")
    profile, policy_sha256 = load_policy(args.profile, args.profile_bundle, args.timeout_seconds)
    verdict = admit(
        args.artifact, args.admitted_artifact, args.report, profile,
        policy_sha256, args.timeout_seconds, args.report_max_characters,
    )
    return 0 if verdict.decision == "ALLOW" else 1


if __name__ == "__main__":
    raise SystemExit(main())
Step 2: Sign and hand off the decision
set -euo pipefail
python supply_chain/model_static_admission.py   --artifact downloads/candidate.bin   --profile policy/model-artifact-profile.json   --profile-bundle policy/model-artifact-profile.sigstore.json   --admitted-artifact quarantine/admitted/candidate.bin   --report evidence/model-static-admission.json   --timeout-seconds "$MODEL_SCANNER_TIMEOUT_SECONDS"   --report-max-characters "$MODEL_SCANNER_REPORT_MAX_CHARACTERS"
cosign sign-blob --yes --key "$ADMISSION_SIGNING_KEY_URI"   --bundle evidence/model-static-admission.sigstore.json   evidence/model-static-admission.json

The loader receives only quarantine/admitted/candidate.bin together with its recorded digest. Unsupported formats, oversize or changing input, scanner failure, unsafe findings, missing scanner version, copy drift, or evidence-signing failure blocks load and promotion. Pickle-derived or custom-loader artifacts still require the separate dynamic sandbox verdict owned by AID-I-001.005.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-H-003.009: Model Artifact Static Malware & Unsafe-Serialization Scanning | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service