"""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") SYNTHETIC_RESOURCE_UUID = "contract-test-resource" ROUTER_PRIORITY_LABELS = { f"traefik.http.routers.http-0-{SYNTHETIC_RESOURCE_UUID}-site.priority": "500", f"traefik.http.routers.https-0-{SYNTHETIC_RESOURCE_UUID}-site.priority": "500", } SOURCE_ROUTER_PRIORITY_TEMPLATES = { "traefik.http.routers.http-0-${UUID}-site.priority=500", "traefik.http.routers.https-0-${UUID}-site.priority=500", } 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], env: dict[str, str] | None = None ) -> subprocess.CompletedProcess[str]: return subprocess.run( command, cwd=ROOT, env=env, 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}") cls.render_command = [ COMPOSE_BIN, "-f", str(COMPOSE_FILE), "config", "--format", "json" ] cls.render_env = os.environ.copy() cls.render_env["COOLIFY_RESOURCE_UUID"] = SYNTHETIC_RESOURCE_UUID rendered = run(cls.render_command, env=cls.render_env) 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_rendered_traefik_router_priorities_are_exact(self) -> None: rendered_labels = self.service().get("labels") self.assertIsInstance(rendered_labels, dict) labels = rendered_labels if isinstance(rendered_labels, dict) else {} self.assertEqual(labels, ROUTER_PRIORITY_LABELS) self.assertTrue(all(isinstance(value, str) for value in labels.values())) def test_priority_labels_use_source_list_syntax_and_uuid_interpolation(self) -> None: source_label_items = { match.group(1) for match in re.finditer( r'^\s*-\s*["\']?(traefik\.(?:http\.)?[^"\']+?)["\']?\s*$', self.source, re.MULTILINE, ) } required_interpolation = re.compile( r"\$\{COOLIFY_RESOURCE_UUID:\?[^}\r\n]+\}" ) normalized_items = { required_interpolation.sub("${UUID}", item) for item in source_label_items } self.assertEqual(normalized_items, SOURCE_ROUTER_PRIORITY_TEMPLATES) self.assertTrue( all(required_interpolation.search(item) for item in source_label_items), "each Traefik label must use required COOLIFY_RESOURCE_UUID interpolation", ) self.assertNotIn(SYNTHETIC_RESOURCE_UUID, self.source) mapping_labels = [ (key, line) for _, key, line in source_mapping_keys(self.source) if key.startswith("traefik.") ] self.assertEqual(mapping_labels, [], "Traefik labels must use Compose list syntax") def test_resource_uuid_interpolation_fails_closed_when_missing_or_empty(self) -> None: missing_env = os.environ.copy() missing_env.pop("COOLIFY_RESOURCE_UUID", None) empty_env = os.environ.copy() empty_env["COOLIFY_RESOURCE_UUID"] = "" for case, env in (("missing", missing_env), ("empty", empty_env)): with self.subTest(case=case): rendered = run(self.render_command, env=env) self.assertNotEqual( rendered.returncode, 0, f"Compose config accepted a {case} COOLIFY_RESOURCE_UUID", ) self.assertIn("COOLIFY_RESOURCE_UUID", rendered.stderr) 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() ) meaningful = meaningful.replace( "${COOLIFY_RESOURCE_UUID:?required}", "$COOLIFY_RESOURCE_UUID" ) 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()