518 lines
20 KiB
Python
518 lines
20 KiB
Python
"""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<mapping>[^"]*:\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"])
|