"""M16 static security and supply-chain guards. A security review is a point-in-time result; these tests turn its conclusions into properties the build enforces. Each one encodes something the M16 review verified by hand, so a later milestone cannot quietly reintroduce it. """ from __future__ import annotations import ast import re from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[2] SOURCE_ROOTS = [ ROOT / "backend" / "src", ROOT / "node-agent" / "src", ROOT / "runtime-worker" / "src", ] def python_sources() -> list[Path]: files: list[Path] = [] for root in SOURCE_ROOTS: if root.is_dir(): files.extend(path for path in root.rglob("*.py") if "__pycache__" not in path.parts) return sorted(files) def parsed_sources() -> list[tuple[Path, ast.Module]]: return [(path, ast.parse(path.read_text("utf-8"))) for path in python_sources()] def _callee_name(node: ast.Call) -> str: """Dotted name of a call target, or "" when the receiver is itself an expression. `model.to("cuda").eval()` is PyTorch switching to inference mode, not the builtin `eval`. A method call on an expression has no resolvable dotted name, so it must not collapse to its final attribute. """ target = node.func parts: list[str] = [] while isinstance(target, ast.Attribute): parts.append(target.attr) target = target.value if isinstance(target, ast.Name): parts.append(target.id) return ".".join(reversed(parts)) return "" # --------------------------------------------------------------------- execution surfaces def test_no_source_file_executes_a_shell() -> None: """`shell=True` turns every interpolated value into a command injection candidate.""" offenders: list[str] = [] for path, tree in parsed_sources(): for node in ast.walk(tree): if not isinstance(node, ast.Call): continue for keyword in node.keywords: if keyword.arg == "shell" and not ( isinstance(keyword.value, ast.Constant) and keyword.value.value is False ): offenders.append(f"{path.relative_to(ROOT)}:{node.lineno}") assert offenders == [], f"shell execution found: {offenders}" @pytest.mark.parametrize( "callee", ["os.system", "os.popen", "eval", "exec", "pickle.loads", "pickle.load"] ) def test_no_source_file_calls_an_arbitrary_execution_primitive(callee: str) -> None: offenders: list[str] = [] for path, tree in parsed_sources(): for node in ast.walk(tree): if isinstance(node, ast.Call) and _callee_name(node) == callee: offenders.append(f"{path.relative_to(ROOT)}:{node.lineno}") assert offenders == [], f"{callee} found: {offenders}" def test_no_subprocess_call_passes_a_string_command_line() -> None: """A string argv can be handed to a shell; a list of arguments never is. A variable holding a list is fine and common, so this rejects the shapes that are genuinely dangerous: a literal string, an f-string, or a string built by concatenation or formatting. """ dangerous = (ast.Constant, ast.JoinedStr, ast.BinOp) offenders: list[str] = [] for path, tree in parsed_sources(): for node in ast.walk(tree): if not isinstance(node, ast.Call): continue name = _callee_name(node) if not name.startswith("subprocess.") or name.endswith( ("TimeoutExpired", "CalledProcessError", "SubprocessError") ): continue if not node.args: offenders.append(f"{path.relative_to(ROOT)}:{node.lineno} (no argv)") continue argv = node.args[0] if isinstance(argv, dangerous) or ( isinstance(argv, ast.Call) and _callee_name(argv).endswith((".format", ".join")) ): offenders.append(f"{path.relative_to(ROOT)}:{node.lineno} ({type(argv).__name__})") assert offenders == [], f"subprocess calls with a string command line: {offenders}" def test_every_subprocess_argv_is_a_list_at_runtime() -> None: """The allowlisted callers build their argv as a list literal before passing it.""" for relative in ( "backend/src/modelforge_api/services/recovery.py", "backend/src/modelforge_api/services/recovery_postgres.py", ): source = (ROOT / relative).read_text("utf-8") tree = ast.parse(source) for node in ast.walk(tree): if not (isinstance(node, ast.Call) and _callee_name(node) == "subprocess.run"): continue argv = node.args[0] if isinstance(argv, ast.Name): assigned = [ statement for statement in ast.walk(tree) if isinstance(statement, ast.Assign) and any( isinstance(target, ast.Name) and target.id == argv.id for target in statement.targets ) ] assert assigned, f"{relative}:{node.lineno} argv {argv.id} is never assigned" assert any( isinstance(statement.value, ast.List) for statement in assigned ), f"{relative}:{node.lineno} argv {argv.id} is not built as a list" else: assert isinstance(argv, ast.List), f"{relative}:{node.lineno}" def test_the_only_subprocess_users_are_the_recovery_plane() -> None: """Subprocess use is allowlisted, so a new one has to be a deliberate decision.""" allowed = { "backend/src/modelforge_api/services/recovery.py", "backend/src/modelforge_api/services/recovery_postgres.py", } users = { str(path.relative_to(ROOT)).replace("\\", "/") for path, tree in parsed_sources() for node in ast.walk(tree) if isinstance(node, ast.Call) and _callee_name(node).startswith("subprocess.") } assert users <= allowed, f"unexpected subprocess users: {sorted(users - allowed)}" def test_the_product_exposes_no_chaos_or_command_route() -> None: """Fault injection is a test harness concern; a control plane must not offer it as an API.""" routes = ROOT / "backend" / "src" / "modelforge_api" / "api" / "routes" # Anchored on whole path segments: an "evaluation" route is a first-class M6 feature, while a # "/chaos" or "/exec" segment would be a command surface. forbidden = re.compile( r"[\"']/[^\"']*/(chaos|shell|exec|command|debug)(/|[\"'])", re.IGNORECASE ) offenders = [ f"{path.relative_to(ROOT)}:{index}" for path in routes.rglob("*.py") for index, line in enumerate(path.read_text("utf-8").splitlines(), start=1) if forbidden.search(line) ] assert offenders == [], f"suspicious route definitions: {offenders}" # --------------------------------------------------------------------- cryptography def test_backup_encryption_uses_a_reviewed_library_primitive() -> None: """ModelForge designs no cryptography; it adapts one.""" source = ( ROOT / "backend" / "src" / "modelforge_api" / "services" / "recovery_crypto.py" ).read_text("utf-8") assert "from cryptography.hazmat.primitives.ciphers.aead import AESGCM" in source assert "AES-256-GCM" in source for banned in ("import hashlib\nfrom Crypto", "def _xor", "custom_cipher", "rot13"): assert banned not in source def test_every_encrypted_chunk_uses_a_fresh_nonce() -> None: """A reused GCM nonce with the same key destroys both confidentiality and authenticity.""" source = ( ROOT / "backend" / "src" / "modelforge_api" / "services" / "recovery_crypto.py" ).read_text("utf-8") implementation = source.split("class AesGcmBackupCipher", 1)[1] encrypt = implementation.split("def encrypt_file", 1)[1].split("def decrypt_file", 1)[0] assert "nonce = os.urandom(_NONCE_BYTES)" in encrypt # The nonce is drawn inside the chunk loop, not once for the whole file. loop = encrypt.split("while chunk := reader.read(CHUNK_BYTES):", 1)[1] assert "nonce = os.urandom(_NONCE_BYTES)" in loop assert "_associated(self.key_id, index)" in loop def test_decryption_failure_leaves_no_plaintext_behind() -> None: source = ( ROOT / "backend" / "src" / "modelforge_api" / "services" / "recovery_crypto.py" ).read_text("utf-8") implementation = source.split("class AesGcmBackupCipher", 1)[1] decrypt = implementation.split("def decrypt_file", 1)[1] assert "temporary.unlink(missing_ok=True)" in decrypt assert "destination.unlink(missing_ok=True)" in decrypt assert "InvalidTag" in decrypt def test_no_encryption_key_is_ever_written_into_a_backup_manifest() -> None: source = ( ROOT / "backend" / "src" / "modelforge_api" / "services" / "recovery.py" ).read_text("utf-8") manifest = source.split("def _configuration_manifest", 1)[1].split("def ", 2)[0] assert "get_secret_value" not in manifest assert "NON_EXPORTABLE_SECRET" in manifest # --------------------------------------------------------------------- deployment posture def test_no_compose_projection_grants_privileged_mode_or_the_docker_socket() -> None: offenders: list[str] = [] for path in sorted(ROOT.glob("docker-compose*.yml")): text = path.read_text("utf-8") for marker in ("privileged: true", "docker.sock", "network_mode: host", "pid: host"): if marker in text: offenders.append(f"{path.name}: {marker}") assert offenders == [], f"unsafe deployment settings: {offenders}" def test_the_datastores_are_not_published_beyond_loopback_by_default() -> None: """A control-plane database on the LAN behind a development password is the whole platform. Asserted through the same mapping parser the projection test uses rather than against a literal string: M17 made the host ports configurable, and a literal assertion would have failed for a change that did not weaken anything. The property is "bound to loopback", not "spelled exactly this way". """ compose = ROOT / "docker-compose.yml" mappings = _published_mappings(compose) datastores = [mapping for mapping in mappings if mapping.endswith((":5432", ":6379"))] assert len(datastores) == 2, f"expected PostgreSQL and Redis mappings, found {datastores}" for mapping in datastores: assert _binds_to_loopback(mapping), f"{mapping} is published beyond loopback" # The port a compose file publishes, as "[host_ip:]host_port:container_port", where any part may be # written as a ${VAR:-default} substitution. _PUBLISHED_PORT = re.compile(r'^\s*-\s*"(?P[^"]*:\d+)"\s*$') # The API is published deliberately: every admin route is operator-authenticated and the console has # to reach it. Everything else defaults to loopback. _DELIBERATELY_PUBLISHED = ("MODELFORGE_API_BIND",) def _published_mappings(path: Path) -> list[str]: return [ match.group("mapping") for line in path.read_text("utf-8").splitlines() if (match := _PUBLISHED_PORT.match(line)) ] def _split_mapping(mapping: str) -> list[str]: """Split on the colons that separate parts, not the ones inside ${VAR:-default}.""" parts: list[str] = [] current: list[str] = [] depth = 0 index = 0 while index < len(mapping): character = mapping[index] if mapping.startswith("${", index): depth += 1 current.append(mapping[index : index + 2]) index += 2 continue if character == "}" and depth: depth -= 1 elif character == ":" and not depth: parts.append("".join(current)) current = [] index += 1 continue current.append(character) index += 1 parts.append("".join(current)) return parts def _binds_to_loopback(mapping: str) -> bool: """A published port is safe only when it names a host address that is the loopback. Two parts means "host_port:container_port", which docker publishes on every interface. """ parts = _split_mapping(mapping) if len(parts) < 3: return False host = parts[0] if host == "127.0.0.1": return True # ${MODELFORGE_X_BIND:-127.0.0.1} — the default must itself be the loopback. default = re.fullmatch(r"\$\{[A-Z0-9_]+:-([^}]*)\}", host) return bool(default and default.group(1) == "127.0.0.1") @pytest.mark.parametrize( "compose", sorted(ROOT.glob("docker-compose*.yml")), ids=lambda path: path.name ) def test_every_compose_projection_binds_its_ports_to_loopback(compose: Path) -> None: """The invariant, not one file. The first version of this asserted the contents of docker-compose.yml, which is the file that had already been fixed. docker-compose.dr.yml published a rehearsal PostgreSQL on every interface, and a DR rehearsal restores the entire control-plane database into it: from the LAN the development credentials connected to it as a superuser. A test that names the file it was written for cannot find the next occurrence, so this one reads every projection. """ for mapping in _published_mappings(compose): if any(name in mapping for name in _DELIBERATELY_PUBLISHED): continue assert _binds_to_loopback(mapping), ( f"{compose.name} publishes {mapping!r} on every interface; bind it to 127.0.0.1 " f"by default and let an operator override it deliberately" ) def test_the_api_and_web_containers_drop_capabilities() -> None: text = (ROOT / "docker-compose.yml").read_text("utf-8") assert text.count("cap_drop:") >= 2 assert text.count("no-new-privileges:true") >= 2 def test_the_console_image_serves_a_build_rather_than_a_development_server() -> None: dockerfile = (ROOT / "frontend" / "Dockerfile").read_text("utf-8") assert "npm run build" in dockerfile assert "nginx-unprivileged" in dockerfile assert 'CMD ["npm", "run", "dev"]' not in dockerfile def test_the_console_sets_its_security_headers_in_every_location() -> None: """nginx does not inherit add_header into a location that declares one of its own.""" config = (ROOT / "frontend" / "nginx.conf").read_text("utf-8") serving_blocks = [ line for line in config.splitlines() if line.strip().startswith("location ") and "deny" not in line ] includes = config.count("include /etc/nginx/conf.d/security-headers.inc;") # One include per serving location, plus the server-level default. assert includes >= len(serving_blocks), f"{includes} includes for {len(serving_blocks)} blocks" # v1 generates the include from a template at image build time so the CSP's connect-src is # derived from the same API base URL that is compiled into the bundle. headers = (ROOT / "frontend" / "security-headers.inc.template").read_text("utf-8") for header in ( "Content-Security-Policy", "X-Content-Type-Options", "X-Frame-Options", "Referrer-Policy", "Cross-Origin-Opener-Policy", "Cross-Origin-Resource-Policy", "Permissions-Policy", ): assert header in headers def test_no_dependency_is_declared_as_a_floating_latest() -> None: """`latest` makes a build unrepeatable and pulls a compromised release automatically.""" import json manifest = json.loads((ROOT / "frontend" / "package.json").read_text("utf-8")) floating = [ f"{section}:{name}" for section in ("dependencies", "devDependencies") for name, spec in manifest.get(section, {}).items() if spec in ("latest", "*", "") ] assert floating == [], f"floating dependency specifiers: {floating}" def test_the_release_images_upgrade_their_installer() -> None: for dockerfile in (ROOT / "backend" / "Dockerfile", ROOT / "node-agent" / "Dockerfile"): assert "--upgrade pip" in dockerfile.read_text("utf-8"), dockerfile def test_every_release_base_image_is_digest_pinned_and_security_updated() -> None: dockerfiles = ( ROOT / "backend" / "Dockerfile", ROOT / "frontend" / "Dockerfile", ROOT / "node-agent" / "Dockerfile", ) for dockerfile in dockerfiles: text = dockerfile.read_text("utf-8") from_lines = [line for line in text.splitlines() if line.startswith("FROM ")] assert from_lines assert all("@sha256:" in line for line in from_lines), dockerfile assert ( "apk upgrade --no-cache" in text or "apt-get upgrade --yes" in text ), dockerfile for dockerfile in (dockerfiles[0], dockerfiles[2]): text = dockerfile.read_text("utf-8") assert '"setuptools>=78.1.1"' in text, dockerfile assert '"msgpack>=1.2.1"' in text, dockerfile assert "pip check" in text, dockerfile assert "python -m pip uninstall --yes pip setuptools" in text, dockerfile def test_node_agent_runtime_is_glibc_multistage_and_keeps_nvidia_hardening() -> None: dockerfile = (ROOT / "node-agent" / "Dockerfile").read_text("utf-8") compose = (ROOT / "docker-compose.node-agent.yml").read_text("utf-8") assert dockerfile.count("FROM python:3.12-slim-trixie@sha256:") == 2 assert " AS builder" in dockerfile assert "alpine" not in dockerfile.lower() assert "USER modelforge-agent" in dockerfile assert "--uid 100" in dockerfile and "--gid 101" in dockerfile assert ( "MODELFORGE_AGENT_ACCELERATOR_MODE: ${MODELFORGE_AGENT_ACCELERATOR_MODE:-nvidia}" in compose ) assert "read_only: true" in compose assert "cap_drop:\n - ALL" in compose assert "no-new-privileges:true" in compose assert "driver: nvidia" in compose and "capabilities: [gpu]" in compose def test_release_builder_uses_reproducible_image_exports() -> None: from importlib.util import module_from_spec, spec_from_file_location spec = spec_from_file_location( "release_build_supply_chain", ROOT / "scripts" / "release_build.py" ) assert spec and spec.loader release_build = module_from_spec(spec) spec.loader.exec_module(release_build) assert release_build.source_date_epoch("2026-08-28T22:00:00Z") == 1787954400 source = (ROOT / "scripts" / "release_build.py").read_text("utf-8") assert '"--metadata-file"' in source assert 'build_metadata.get("containerimage.digest")' in source with pytest.raises(ValueError, match="UTC offset"): release_build.source_date_epoch("2026-08-28T22:00:00") assert ( release_build.manifest_digest("modelforge-api@sha256:" + "a" * 64) == "sha256:" + "a" * 64 ) assert release_build.manifest_digest(None) is None script = (ROOT / "scripts" / "release_build.py").read_text("utf-8") assert '"--provenance=false"' in script assert '"--sbom=false"' in script assert "SOURCE_DATE_EPOCH" in script def test_sbom_accepts_an_unlabelled_image(monkeypatch: pytest.MonkeyPatch) -> None: from importlib.util import module_from_spec, spec_from_file_location spec = spec_from_file_location("m16_sbom_unlabelled", ROOT / "scripts" / "m16_sbom.py") assert spec and spec.loader sbom = module_from_spec(spec) spec.loader.exec_module(sbom) def fake_docker(*args: str) -> str: rendered = " ".join(args) if "Config.Labels}}" in rendered: return "null" if "RepoDigests" in rendered: return "example@sha256:" + "b" * 64 if '"base"' in rendered: return "" if "{{.Id}}" in rendered: return "sha256:" + "c" * 64 if "{{.Created}}" in rendered: return "2026-08-28T22:00:00Z" raise AssertionError(rendered) monkeypatch.setattr(sbom, "docker", fake_docker) monkeypatch.setattr(sbom, "git", lambda *args: "origin") result = sbom.image_provenance("example:1.0", "d" * 40) assert result["oci"] == {} def test_the_sbom_and_provenance_are_present_and_bound_to_a_commit() -> None: import json sbom_path = ROOT / "docs" / "security" / "sbom" / "modelforge-cyclonedx.json" provenance_path = ROOT / "docs" / "security" / "sbom" / "image-provenance.json" assert sbom_path.is_file() and provenance_path.is_file() sbom = json.loads(sbom_path.read_text("utf-8")) assert sbom["bomFormat"] == "CycloneDX" assert len(sbom["components"]) > 100 assert all("purl" in component for component in sbom["components"]) provenance = json.loads(provenance_path.read_text("utf-8")) assert len(provenance["images"]) >= 3 assert re.fullmatch(r"[0-9a-f]{40}", provenance["source_commit"]) assert all(item["image_id"] for item in provenance["images"])