"""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" 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_source_defines_no_traefik_labels(self) -> None: source_label_names = { match.group(1).strip("'\"") for match in re.finditer( r'^\s*-\s*([^\s=]+)(?:=.*)?$', self.source, re.IGNORECASE | re.MULTILINE, ) } source_label_names.update( key for _, key, line in source_mapping_keys(self.source) ) traefik_labels = sorted( name for name in source_label_names if re.search(r"(?:^|[._-])traefik(?:[._-]|$)", name, re.IGNORECASE) ) # Interpolating a label key is unsafe: Coolify preprocesses list labels # into a mapping before Compose interpolation, leaving that key literal. self.assertEqual( traefik_labels, [], f"source Traefik labels are forbidden; Coolify owns them: {traefik_labels}", ) def test_compose_renders_without_resource_uuid(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.assertEqual( rendered.returncode, 0, f"Compose config rejected a {case} COOLIFY_RESOURCE_UUID:\n{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()