test: define hardened Compose contract

tests/run.py

@@ -0,0 +1,20 @@
+"""Compile the contract tests, then execute unittest discovery."""
+
+from pathlib import Path
+import py_compile
+import sys
+import unittest
+
+
+TEST_DIR = Path(__file__).resolve().parent
+
+
+def main() -> int:
+    for test_file in sorted(TEST_DIR.glob("test_*.py")):
+        py_compile.compile(str(test_file), doraise=True)
+    suite = unittest.defaultTestLoader.discover(str(TEST_DIR), pattern="test_*.py")
+    return 0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_compose_contract.py

@@ -0,0 +1,203 @@
+"""Executable contract for the hardened, Coolify-compatible Compose service."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import unittest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+COMPOSE_FILE = ROOT / "compose.yaml"
+DOCKERFILE = ROOT / "Dockerfile"
+COMPOSE_BIN = os.environ.get("COMPOSE_BIN", "/opt/data/tmp/docker-compose-birweb")
+
+DOCKERFILE_SHA256 = "748bd362b43475779d241074a1955ae880967c923a7f68c693cb1c67f8b242dd"
+BASE_IMAGE = (
+    "nginxinc/nginx-unprivileged:1.27-alpine@"
+    "sha256:65e3e85dbaed8ba248841d9d58a899b6197106c23cb0ff1a132b7bfe0547e4c0"
+)
+HEALTH_COMMAND = "wget -qO- http://127.0.0.1:8080/health.json || exit 1"
+TMPFS_TARGETS = {"/tmp", "/var/cache/nginx", "/var/run"}
+
+
+def run(command: list[str]) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        command,
+        cwd=ROOT,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        check=False,
+    )
+
+
+def source_mapping_keys(source: str) -> list[tuple[int, str, int]]:
+    """Return (indent, key, line) for YAML block-mapping keys.
+
+    Compose rendering can discard extension fields or normalize some forbidden
+    constructs. This deliberately small scanner inspects source structure while
+    ignoring blank lines, comments, sequence values, and colons inside quotes.
+    The contract file is expected to use ordinary block-style Compose YAML.
+    """
+    keys: list[tuple[int, str, int]] = []
+    key_pattern = re.compile(r"^([A-Za-z_][A-Za-z0-9_.-]*|['\"][^'\"]+['\"]):(?:\s|$)")
+    for line_number, raw_line in enumerate(source.splitlines(), 1):
+        content = raw_line.lstrip(" ")
+        if not content or content.startswith("#") or content.startswith("-"):
+            continue
+        match = key_pattern.match(content)
+        if match:
+            keys.append((len(raw_line) - len(content), match.group(1).strip("'\""), line_number))
+    return keys
+
+
+def parse_size(value: object) -> int:
+    match = re.fullmatch(r"([0-9]+(?:\.[0-9]+)?)([kKmMgG])?[bB]?", str(value))
+    if not match:
+        raise AssertionError(f"invalid Compose memory size: {value!r}")
+    amount = float(match.group(1))
+    multiplier = {None: 1, "k": 1024, "m": 1024**2, "g": 1024**3}[lower(match.group(2))]
+    return int(amount * multiplier)
+
+
+def lower(value: str | None) -> str | None:
+    return value.lower() if value else None
+
+
+class ComposeContract(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        version = run([COMPOSE_BIN, "version"])
+        if version.returncode != 0:
+            raise AssertionError(
+                f"verified Compose binary is not executable: {COMPOSE_BIN}\n"
+                f"{version.stderr or version.stdout}"
+            )
+        if not COMPOSE_FILE.is_file():
+            raise AssertionError(f"RED: required Compose file is missing: {COMPOSE_FILE}")
+        rendered = run(
+            [COMPOSE_BIN, "compose", "-f", str(COMPOSE_FILE), "config", "--format", "json"]
+        )
+        if rendered.returncode != 0:
+            raise AssertionError(f"Compose config rendering failed:\n{rendered.stderr}")
+        try:
+            cls.config = json.loads(rendered.stdout)
+        except json.JSONDecodeError as error:
+            raise AssertionError(f"Compose did not emit valid JSON: {error}") from error
+        cls.source = COMPOSE_FILE.read_text(encoding="utf-8")
+
+    def service(self) -> dict[str, object]:
+        services = self.config.get("services")
+        self.assertIsInstance(services, dict)
+        self.assertEqual(set(services), {"site"})
+        return services["site"]
+
+    def test_exactly_one_service_and_required_build(self) -> None:
+        service = self.service()
+        self.assertEqual(service.get("build", {}).get("context"), str(ROOT))
+        self.assertEqual(service.get("build", {}).get("dockerfile"), "Dockerfile")
+        self.assertNotIn("image", service)
+        self.assertNotIn("container_name", service)
+
+    def test_runtime_identity_and_hardening(self) -> None:
+        service = self.service()
+        self.assertEqual(service.get("user"), "101:101")
+        self.assertIs(service.get("read_only"), True)
+        self.assertEqual(service.get("cap_drop"), ["ALL"])
+        self.assertEqual(service.get("security_opt"), ["no-new-privileges:true"])
+        for forbidden in ("privileged", "network_mode", "devices", "cap_add"):
+            self.assertNotIn(forbidden, service)
+
+    def test_no_host_ports_and_only_container_port_8080(self) -> None:
+        service = self.service()
+        self.assertNotIn("ports", service)
+        self.assertEqual(service.get("expose"), ["8080"])
+
+    def test_tmpfs_targets_options_and_size_caps(self) -> None:
+        tmpfs = self.service().get("tmpfs")
+        self.assertIsInstance(tmpfs, list)
+        parsed: dict[str, set[str]] = {}
+        for entry in tmpfs:
+            target, separator, options = str(entry).partition(":")
+            self.assertTrue(separator, f"tmpfs entry lacks mount options: {entry!r}")
+            parsed[target] = set(options.split(","))
+        self.assertEqual(set(parsed), TMPFS_TARGETS)
+        for target, options in parsed.items():
+            self.assertTrue({"rw", "noexec", "nosuid"}.issubset(options), target)
+            size_options = [option for option in options if option.startswith("size=")]
+            self.assertEqual(len(size_options), 1, target)
+            self.assertEqual(len(options), 4, f"unexpected tmpfs options for {target}: {options}")
+            size = parse_size(size_options[0].split("=", 1)[1])
+            self.assertGreater(size, 0, target)
+            self.assertLessEqual(size, 64 * 1024**2, f"unbounded tmpfs size for {target}")
+
+    def test_healthcheck_and_restart_are_exact(self) -> None:
+        service = self.service()
+        self.assertEqual(
+            service.get("healthcheck"),
+            {
+                "test": ["CMD-SHELL", HEALTH_COMMAND],
+                "interval": "10s",
+                "timeout": "2s",
+                "retries": 3,
+            },
+        )
+        self.assertEqual(service.get("restart"), "unless-stopped")
+
+    def test_resource_limits_are_conservative_and_portable(self) -> None:
+        service = self.service()
+        limits = service.get("deploy", {}).get("resources", {}).get("limits", {})
+        self.assertEqual(set(limits), {"cpus", "memory"})
+        cpus = float(limits["cpus"])
+        self.assertGreater(cpus, 0)
+        self.assertLessEqual(cpus, 1.0)
+        memory = parse_size(limits["memory"])
+        self.assertGreaterEqual(memory, 16 * 1024**2)
+        self.assertLessEqual(memory, 512 * 1024**2)
+
+    def test_no_secrets_bind_mounts_or_forbidden_source_keys(self) -> None:
+        service = self.service()
+        self.assertNotIn("secrets", self.config)
+        self.assertNotIn("secrets", service)
+        self.assertNotIn("volumes", service)
+        forbidden = {
+            "secrets", "volumes", "ports", "image", "container_name",
+            "privileged", "network_mode", "devices", "cap_add",
+            "pid", "ipc", "uts", "cgroup", "cgroup_parent",
+        }
+        found = [(key, line) for _, key, line in source_mapping_keys(self.source) if key in forbidden]
+        self.assertEqual(found, [], f"forbidden Compose source keys: {found}")
+        # Keep the source auditable by the structural scanner: flow mappings and
+        # YAML merge/alias tricks could otherwise hide keys from line-based review.
+        meaningful = "\n".join(
+            line.split("#", 1)[0] for line in self.source.splitlines()
+        )
+        self.assertNotRegex(meaningful, r"[{}]|(?:^|\s)[&*][A-Za-z0-9_-]+|<<\s*:")
+
+
+class DockerfileContract(unittest.TestCase):
+    def test_dockerfile_is_byte_for_byte_locked(self) -> None:
+        digest = hashlib.sha256(DOCKERFILE.read_bytes()).hexdigest()
+        self.assertEqual(digest, DOCKERFILE_SHA256)
+
+    def test_immutable_image_identity_and_runtime_contract(self) -> None:
+        instructions = [line.strip() for line in DOCKERFILE.read_text(encoding="utf-8").splitlines()]
+        self.assertEqual(instructions[0], f"FROM {BASE_IMAGE}")
+        self.assertEqual([line for line in instructions if line.startswith("USER ")], ["USER 101"])
+        self.assertEqual([line for line in instructions if line.startswith("EXPOSE ")], ["EXPOSE 8080"])
+        self.assertEqual(
+            [line for line in instructions if line.startswith("HEALTHCHECK ")],
+            [
+                "HEALTHCHECK --interval=10s --timeout=2s --retries=3 "
+                f"CMD {HEALTH_COMMAND}"
+            ],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()