Phase 8: Production AI Security Architecture · 60 min · pip-audit · hashlib · importlib.metadata
Supply Chain Security & SBOM for ML
You didn't write the weights, the base image, or half your dependency tree -- but you're the one who ships them to production.
What you will learn
- Run pip-audit (or explain and simulate its behavior when unavailable) to detect known-vulnerable dependencies in an ML pipeline's package tree
- Build a SHA-256 model weight hashing and verification utility that detects a tampered or substituted checkpoint before it's loaded into a serving process
- Generate a minimal CycloneDX-shaped SBOM for a pip environment, including model weights as first-class components
- Explain dependency confusion and compromised-package attack patterns specific to ML pipelines, and where in the pipeline each control in this lesson would catch them
The Problem
Every earlier phase of this course assumed a specific model, a specific guardrail library, a specific set of Python packages, and asked "how do we secure what runs on top of these." This lesson asks the question underneath that one: how do you know the model, the packages, and the container you're building on top of are what you think they are?
This isn't a hypothetical layer. OWASP's GenAI/LLM Top 10 dedicates two categories to it directly — LLM03 (training data poisoning) and LLM05 (supply chain vulnerabilities) — because ML pipelines have a wider and less-audited supply chain than typical application code. A normal web app pulls in npm or pip packages. An ML pipeline pulls in those, plus multi-gigabyte model weight files downloaded from a hub, plus base container images with CUDA/driver stacks, plus adapter/LoRA weights that can be swapped into a base model post-training. Every one of those is an opportunity for something malicious to be substituted for something legitimate, and unlike a pip package (which at least has a changelog and a maintainer), a downloaded .safetensors file gives you no signal at all about whether it's what the model card says it is.
Three controls close different parts of this gap, and this lesson builds all three from scratch so you understand exactly what each one is asserting:
- Dependency vulnerability scanning (pip-audit) — catches known-CVE packages in your dependency tree.
- Model weight verification (hashing) — catches a checkpoint that's been tampered with or silently substituted after you first trusted it.
- SBOM generation — produces the artifact that lets anyone downstream of you — a security reviewer, an auditor, a customer doing vendor due diligence — answer "what exactly is in this system" without re-deriving it themselves.
Dependency scanning with pip-audit
pip-audit (https://github.com/pypa/pip-audit) is the PyPA's own supply-chain scanner. It queries the Python Packaging Advisory Database — backed by Google's OSV project — for every package in your environment or requirements.txt, and reports back CVE/PYSEC/GHSA identifiers for anything with a known vulnerability, plus the version that fixes it. No API key, no account, and it runs the same way in CI as it does on your laptop.
The lesson's code never assumes the tool is installed — it checks with shutil.which("pip-audit") first, exactly the way you'd guard an optional external tool in production code:
if shutil.which("pip-audit") is None:
# explain what would run and what the output means, don't crash
...
cmd = ["pip-audit", "--format", "json"]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
This pattern — check availability, explain the fallback, never silently no-op — matters more here than in a typical integration, because a supply-chain scanner that fails silently and reports "no vulnerabilities found" when it actually never ran is worse than not having the control at all: it produces false confidence instead of an honest gap.
A scan that didn't run is not the same as a scan that found nothing
The single most dangerous failure mode for any security scanning tool -- pip-audit, a SAST tool, a container scanner -- is silently not running and reporting success anyway, because a misconfigured CI step, a missing binary, or a network timeout looks identical to "clean" if you only check the exit code without checking that the tool actually executed. Every scanner integration in this lesson distinguishes "tool ran and found 0 issues" from "tool didn't run" as different states, and a CI gate (Lesson 3) should always treat the second as a hard failure, not a pass.
Model weight verification: hashing as a trust boundary
Once you've downloaded a model checkpoint from a hub, a vendor, or a teammate's S3 bucket, how do you know it's still the file you originally vetted the next time you deploy it? This is the exact same question a pip install hash-check answers for a package, applied to a .safetensors or .bin file:
def hash_file(path, chunk_size=1 << 20) -> str:
sha256 = hashlib.sha256()
with open(path, "rb") as fh:
while chunk := fh.read(chunk_size):
sha256.update(chunk)
return sha256.hexdigest()
Reading in fixed-size chunks matters for real usage, not just this lesson's small demo file: production model weights are often tens of gigabytes, and loading a whole checkpoint into memory just to hash it would double your peak memory usage for no reason. The chunked version streams constant memory regardless of file size.
The actual control isn't the hash function — it's the workflow around it: hash every weight file once, when you trust it (right after downloading from a verified source, or right after your own training run), commit that manifest to your model registry or repo, and re-hash and compare before every deploy. A single differing byte — from corruption, a partial download, or a deliberate substitution — produces a completely different hash, and the file must not be loaded if it doesn't match.
A team downloads a model checkpoint from a public hub, verifies its SHA-256 hash matches the model card's published hash, and deploys it. Three months later, they redeploy the exact same service without re-downloading or re-verifying the file. Is skipping re-verification on redeploy safe?
Explanation: The threat model for weight verification isn't just "was this file legitimate when I first got it" -- it's "is this file, right now, at the moment I'm about to load it into a serving process, identical to the version I trusted." A compromised build pipeline, a tampered artifact store, or even routine disk corruption can change a file between verifications without any user action. Re-hashing before every deploy costs milliseconds for a cached comparison and closes exactly this gap; skipping it because "we checked once" reintroduces the risk the control exists to remove.
SBOM generation for ML pipelines
A Software Bill of Materials is the artifact that turns "trust us, we scanned it" into something a security reviewer, auditor, or customer can independently verify. CycloneDX is the most widely adopted SBOM format, and while real pipelines typically generate one with cyclonedx-py or pip-audit --format cyclonedx-json, this lesson builds a minimal version from importlib.metadata — Python's stdlib package introspection — so you understand the actual data model underneath the tooling: every component gets a name, a version, a type, a license (when declared), and a purl (package URL) that uniquely identifies it.
The detail specific to ML pipelines: an SBOM that only lists pip packages is incomplete. What model weights does the system load, and what version/hash are they pinned to? This lesson's SBOM generator accepts extra_components specifically so model weight artifacts — identified by their verified hash from the previous section, not just a filename — appear in the same document as your pip dependency tree. This is the concrete answer to "map this system's dependencies for a compliance review": one document, one format, covering both layers of your actual supply chain.
A vendor sends you an SBOM for their ML pipeline that lists 140 pip packages with versions and licenses, but no model weight artifacts at all. Why is this SBOM incomplete for a security review of an ML system specifically (as opposed to a normal web application)?
The lesson is explicit about this gap: "an SBOM that only lists pip packages is incomplete" for ML systems because "ML pipelines pull in... multi-gigabyte model weight files downloaded from a hub... plus adapter/LoRA weights that can be swapped into a base model post-training." A pip-only SBOM answers "what code is in this system" but says nothing about "what model weights it loads," which is exactly the layer LLM05 (supply chain vulnerabilities) and LLM03 (training data poisoning) target. This is why this lesson's generate_pip_sbom() accepts extra_components for weight artifacts identified by their verified hash — without that, a reviewer has no way to even know a weight file exists, let alone confirm it's the one that was vetted.
Build
Build it
The code artifact below implements all three controls: a run_pip_audit() wrapper that shells out to the real tool via subprocess when available and explains the fallback when it isn't, a hash_file/build_weight_manifest/verify_weights chain for model weight verification (demonstrated catching a deliberately tampered file), and a generate_pip_sbom() function producing a CycloneDX-shaped SBOM including model-weight components.
"""
lesson_08_04_supply_chain_scanner.py
Three supply-chain security controls for ML pipelines, each usable standalone
or together as a pre-deploy check:
1. Dependency vulnerability scanning -- shells out to the real `pip-audit`
tool (https://github.com/pypa/pip-audit) if it's installed, via
`shutil.which` + `subprocess`. If it isn't installed, prints exactly
what it would run and what the JSON output means, so the lesson is
fully runnable without requiring the tool.
2. Model weight verification -- a from-scratch SHA-256 hashing/verification
utility for model artifact files, the same mechanism (with the same
failure modes) as verifying a pip package hash or a container image
digest, applied to a `.bin`/`.safetensors`/`.pt` weight file.
3. SBOM generation -- a minimal CycloneDX-shaped Software Bill of Materials
JSON generator for the current Python environment, built from
`importlib.metadata` with no external dependencies.
Run:
python lesson_08_04_supply_chain_scanner.py
No network calls except the optional pip-audit subprocess, which itself
queries the OSV/PyPI advisory database only if pip-audit is installed and
you choose to run it against a real environment -- the demo below runs
entirely offline against synthetic data when pip-audit isn't present.
"""
from __future__ import annotations
import hashlib
import json
import shutil
import subprocess
import sys
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
try:
import importlib.metadata as importlib_metadata
except ImportError: # pragma: no cover - Python < 3.8 fallback, not expected here
import importlib_metadata # type: ignore
# ---------------------------------------------------------------------------
# 1. Dependency vulnerability scanning via pip-audit
# ---------------------------------------------------------------------------
@dataclass
class VulnerabilityFinding:
package: str
version: str
vuln_id: str
fix_versions: list[str]
description: str
def run_pip_audit(requirements_path: str | None = None) -> tuple[bool, list[VulnerabilityFinding], str]:
"""
Run the real pip-audit tool if installed (checked via shutil.which, never
assumed). Returns (tool_available, findings, raw_report).
pip-audit (https://github.com/pypa/pip-audit) queries the Python
Packaging Advisory Database (backed by OSV) for every installed or
pinned package and reports known CVEs, without needing an API key.
"""
if shutil.which("pip-audit") is None:
explanation = (
"pip-audit is not installed in this environment.\n"
"Install with: pip install pip-audit\n\n"
"What this function would run:\n"
f" pip-audit --format json"
f"{f' -r {requirements_path}' if requirements_path else ' (against the current environment)'}\n\n"
"What the JSON output contains per finding: the package name, the\n"
"installed version, one or more vulnerability IDs (PYSEC-/CVE- or\n"
"GHSA- identifiers pulled from the OSV database), a description,\n"
"and a list of fixed versions you could upgrade to. A CI security\n"
"gate (Lesson 3) treats any non-empty findings list on a\n"
"supply-chain-critical package (e.g. a model-loading or\n"
"deserialization library) as a blocking failure, and non-critical\n"
"findings as a warning routed to a dependency-update ticket."
)
return False, [], explanation
cmd = ["pip-audit", "--format", "json"]
if requirements_path:
cmd += ["-r", requirements_path]
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120, check=False)
except (subprocess.TimeoutExpired, OSError) as exc:
return True, [], f"pip-audit invocation failed: {exc}"
findings: list[VulnerabilityFinding] = []
try:
parsed = json.loads(result.stdout or "{}")
for dep in parsed.get("dependencies", []):
for vuln in dep.get("vulns", []):
findings.append(
VulnerabilityFinding(
package=dep.get("name", "unknown"),
version=dep.get("version", "unknown"),
vuln_id=vuln.get("id", "unknown"),
fix_versions=vuln.get("fix_versions", []),
description=vuln.get("description", ""),
)
)
except json.JSONDecodeError:
return True, [], result.stdout or result.stderr or "pip-audit produced no parseable output."
return True, findings, result.stdout
def render_pip_audit_report(available: bool, findings: list[VulnerabilityFinding], raw: str) -> str:
lines = ["## Dependency Vulnerability Scan (pip-audit)", ""]
if not available:
lines.append("Tool status: NOT INSTALLED (checked via `shutil.which('pip-audit')`)")
lines.append("")
lines.append(raw)
return "\n".join(lines)
lines.append(f"Tool status: available, {len(findings)} finding(s)")
if not findings:
lines.append("- No known vulnerabilities found in scanned dependencies.")
for f in findings:
fix = ", ".join(f.fix_versions) if f.fix_versions else "no fix available yet"
lines.append(f"- [{f.vuln_id}] {f.package}=={f.version}: {f.description or 'see advisory'} (fix: {fix})")
return "\n".join(lines)
# ---------------------------------------------------------------------------
# Simulated pip-audit findings for the offline demo (used only when the real
# tool isn't installed, so the lesson still demonstrates what a finding
# looks like and how a gate would react to it).
# ---------------------------------------------------------------------------
def simulated_pip_audit_findings() -> list[VulnerabilityFinding]:
"""A synthetic but realistically-shaped finding, standing in for what
pip-audit would report against a genuinely vulnerable pinned dependency.
Used only for the offline demo path -- never presented as a real scan."""
return [
VulnerabilityFinding(
package="example-vulnerable-pkg",
version="1.2.0",
vuln_id="PYSEC-2023-9999",
fix_versions=["1.2.1", "1.3.0"],
description="Deserialization of untrusted data can lead to arbitrary code execution (synthetic example finding).",
)
]
# ---------------------------------------------------------------------------
# 2. Model weight verification and hashing
# ---------------------------------------------------------------------------
def hash_file(path: str | Path, chunk_size: int = 1 << 20) -> str:
"""SHA-256 hash of a file, read in chunks so multi-gigabyte model weight
files don't need to be loaded into memory at once."""
sha256 = hashlib.sha256()
with open(path, "rb") as fh:
while chunk := fh.read(chunk_size):
sha256.update(chunk)
return sha256.hexdigest()
@dataclass
class WeightManifestEntry:
filename: str
sha256: str
size_bytes: int
@dataclass
class WeightVerificationResult:
filename: str
expected_hash: str
actual_hash: str
verified: bool
def build_weight_manifest(model_dir: str | Path) -> list[WeightManifestEntry]:
"""
Hash every weight-shaped file in a model directory and produce a
manifest -- the artifact you'd commit alongside a model release, publish
to a model registry, or compare against a vendor-published hash before
trusting a downloaded checkpoint (the same threat pip-audit addresses
for packages: a compromised or tampered artifact silently substituted
for the legitimate one).
"""
model_dir = Path(model_dir)
manifest: list[WeightManifestEntry] = []
weight_extensions = {".bin", ".safetensors", ".pt", ".pth", ".onnx", ".gguf"}
for path in sorted(model_dir.rglob("*")):
if path.is_file() and path.suffix in weight_extensions:
manifest.append(
WeightManifestEntry(
filename=str(path.relative_to(model_dir)),
sha256=hash_file(path),
size_bytes=path.stat().st_size,
)
)
return manifest
def verify_weights(model_dir: str | Path, manifest: list[WeightManifestEntry]) -> list[WeightVerificationResult]:
"""
Re-hash every file named in a trusted manifest and compare. This is the
check you run before loading a downloaded checkpoint into a serving
process: if a single byte differs from the manifest -- whether from
corruption, a partial download, or a deliberate supply-chain
substitution -- the hash will not match, and the file must not be
loaded.
"""
model_dir = Path(model_dir)
results: list[WeightVerificationResult] = []
for entry in manifest:
file_path = model_dir / entry.filename
if not file_path.exists():
results.append(WeightVerificationResult(entry.filename, entry.sha256, "MISSING", False))
continue
actual = hash_file(file_path)
results.append(WeightVerificationResult(entry.filename, entry.sha256, actual, actual == entry.sha256))
return results
def render_weight_verification_report(results: list[WeightVerificationResult]) -> str:
lines = ["## Model Weight Verification", ""]
for r in results:
status = "VERIFIED" if r.verified else "MISMATCH -- DO NOT LOAD"
lines.append(f"- [{status}] {r.filename}")
if not r.verified:
lines.append(f" expected: {r.expected_hash}")
lines.append(f" actual: {r.actual_hash}")
all_verified = all(r.verified for r in results)
lines.append("")
lines.append(f"**Overall: {'ALL WEIGHTS VERIFIED' if all_verified else 'VERIFICATION FAILED -- BLOCK DEPLOYMENT'}**")
return "\n".join(lines)
# ---------------------------------------------------------------------------
# 3. SBOM generation for a pip environment
# ---------------------------------------------------------------------------
@dataclass
class SBOMComponent:
name: str
version: str
license: str | None = None
def generate_pip_sbom(component_type: str = "application", extra_components: list[SBOMComponent] | None = None) -> dict:
"""
Minimal CycloneDX-shaped SBOM for the current Python environment, built
entirely from importlib.metadata (stdlib, no network, no external
dependency needed to generate it). Real pipelines typically shell out to
`cyclonedx-py` or `pip-audit --format cyclonedx-json` for a fuller spec
implementation; this demonstrates the underlying data model so you
understand what's actually being asserted when a vendor asks for your
SBOM.
extra_components lets you fold in non-pip artifacts an ML pipeline SBOM
should also declare -- most importantly, model weight files with their
verified hashes from build_weight_manifest(), since "what code is in
this system" is incomplete without "what model weights it loads."
"""
components = []
for dist in importlib_metadata.distributions():
name = dist.metadata.get("Name") or dist.metadata.get("Summary") or "unknown"
version = dist.version or "unknown"
license_field = dist.metadata.get("License")
components.append(
{
"type": "library",
"name": name,
"version": version,
"license": license_field if license_field and license_field != "UNKNOWN" else None,
"purl": f"pkg:pypi/{name.lower()}@{version}",
}
)
# de-duplicate by name, keep first occurrence, sort for reproducible output
seen: set[str] = set()
deduped = []
for c in components:
key = c["name"].lower()
if key in seen:
continue
seen.add(key)
deduped.append(c)
deduped.sort(key=lambda c: c["name"].lower())
for extra in extra_components or []:
deduped.append(
{
"type": "model-weights",
"name": extra.name,
"version": extra.version,
"license": extra.license,
"purl": None,
}
)
sbom = {
"bomFormat": "CycloneDX",
"specVersion": "1.5",
"serialNumber": "urn:uuid:00000000-0000-4000-8000-000000000000",
"version": 1,
"metadata": {
"timestamp": datetime.now(timezone.utc).isoformat(),
"component": {"type": component_type, "name": "ml-pipeline"},
},
"components": deduped,
}
return sbom
def render_sbom_summary(sbom: dict) -> str:
lines = ["## SBOM Summary", ""]
libs = [c for c in sbom["components"] if c["type"] == "library"]
weights = [c for c in sbom["components"] if c["type"] == "model-weights"]
lines.append(f"- {len(libs)} pip package component(s)")
lines.append(f"- {len(weights)} model-weight component(s)")
lines.append(f"- Generated: {sbom['metadata']['timestamp']}")
return "\n".join(lines)
# ---------------------------------------------------------------------------
# Demo
# ---------------------------------------------------------------------------
if __name__ == "__main__":
import tempfile
print("=" * 70)
print("1. Dependency vulnerability scanning (pip-audit)")
print("=" * 70)
available, findings, raw = run_pip_audit()
if not available:
print(render_pip_audit_report(available, findings, raw))
print("\n--- Offline demo: what a real finding would look like ---")
demo_findings = simulated_pip_audit_findings()
print(render_pip_audit_report(True, demo_findings, "(simulated)"))
else:
print(render_pip_audit_report(available, findings, raw))
print("\n" + "=" * 70)
print("2. Model weight verification (SHA-256 manifest + re-verify)")
print("=" * 70)
with tempfile.TemporaryDirectory() as tmp:
tmp_path = Path(tmp)
# Simulate a small model artifact directory -- real weight files are
# gigabytes, but the hashing logic is identical at any size since
# hash_file() streams in fixed-size chunks.
(tmp_path / "model.safetensors").write_bytes(b"synthetic model weights v1" * 1000)
(tmp_path / "config.json").write_text('{"architecture": "demo"}')
manifest = build_weight_manifest(tmp_path)
print(f"Built manifest with {len(manifest)} weight file(s):")
for entry in manifest:
print(f" {entry.filename} sha256={entry.sha256[:16]}... ({entry.size_bytes} bytes)")
print("\nScenario A: verifying against the unmodified files")
results_ok = verify_weights(tmp_path, manifest)
print(render_weight_verification_report(results_ok))
print("\nScenario B: simulating a tampered/substituted weight file")
(tmp_path / "model.safetensors").write_bytes(b"MALICIOUS SUBSTITUTED WEIGHTS" * 500)
results_tampered = verify_weights(tmp_path, manifest)
print(render_weight_verification_report(results_tampered))
print("\n" + "=" * 70)
print("3. SBOM generation for the current pip environment")
print("=" * 70)
extra = [SBOMComponent(name="model.safetensors", version="sha256:" + hashlib.sha256(b"synthetic model weights v1" * 1000).hexdigest()[:16], license="model-license-pending-review")]
sbom = generate_pip_sbom(extra_components=extra)
print(render_sbom_summary(sbom))
print("\nFirst 5 components (of {}):".format(len(sbom["components"])))
for c in sbom["components"][:5]:
print(f" [{c['type']}] {c['name']}=={c['version']}")
print("\n" + "=" * 70)
print("Full SBOM JSON written to stdout is available via generate_pip_sbom();")
print("the lab requires saving it to sbom.json.")
print("=" * 70)
Unlock the full lesson
You've read the first 2 sections. The rest of this lesson covers Check Yourself — plus a hands-on lab, quiz, and project artifact.
Create a free account — the opening phases of 29 of 35 courses are free, no credit card.
Browse all courses · View pricing · DeVenture Academy