From f94407e5875611456c325d1cbfed61c6010c1fc0 Mon Sep 17 00:00:00 2001 From: BirSite Automation Date: Sun, 19 Jul 2026 17:28:59 +0000 Subject: [PATCH] test: define hardened Compose contract --- tests/run.py | 20 ++++ tests/test_compose_contract.py | 203 +++++++++++++++++++++++++++++++++ 2 files changed, 223 insertions(+) create mode 100644 tests/run.py create mode 100644 tests/test_compose_contract.py diff --git a/tests/run.py b/tests/run.py new file mode 100644 index 0000000..9c878c1 --- /dev/null +++ b/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()) diff --git a/tests/test_compose_contract.py b/tests/test_compose_contract.py new file mode 100644 index 0000000..1c20ac8 --- /dev/null +++ b/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()