#!/usr/bin/env python3 """Validate the bounded Nextcloud29 fixture contract, not arbitrary networking. The launcher must supply a reviewed image/probe, not treat labels as provenance. Limits below come from the existing 2-CPU / 3-GiB / 256-PID fixture recipe. The runner must recheck network namespace identity before executing test code. """ import json import os import re import stat import sys GIB = 1024 ** 3 CAPABILITIES = {"CHOWN", "DAC_OVERRIDE", "FOWNER", "SETUID", "SETGID", "NET_BIND_SERVICE"} SCRATCH = {"/var/www/html": 2 * GIB, "/run": 32 * 1024 ** 2, "/run/lock": 16 * 1024 ** 2, "/tmp": 128 * 1024 ** 2} def validate(items, name, image): def require(value, message): if not value: raise ValueError(message) def integer(value, low, high): return type(value) is int and low <= value <= high require(re.fullmatch(r"archy-cloud-nextcloud-contract-[A-Za-z0-9_-]+", name), "Invalid fixture name") require(re.fullmatch(r"(?:sha256:)?[a-f0-9]{64}", image), "Immutable fixture image required") require(isinstance(items, list) and len(items) == 1, "One fixture required") item = items[0] require(item.get("Name", "").lstrip("/") == name, "Fixture name mismatch") require(item.get("Image", "").removeprefix("sha256:") == image.removeprefix("sha256:"), "Fixture image mismatch") config = item.get("Config", {}) host = item.get("HostConfig", {}) require(config.get("Labels", {}).get("org.archipelago.cloud-fixture") == "nextcloud29-v1", "Fixture label required") require(host.get("NetworkMode") == "none", "Fixture must have network=none") require(not host.get("PortBindings") and not host.get("PublishAllPorts"), "Host ports forbidden") require(host.get("Privileged") is False and host.get("ReadonlyRootfs") is True, "Read-only unprivileged fixture required") require(host.get("PidMode", "") in ("", "private") and host.get("IpcMode") == "private", "Private process and IPC namespaces required") require(integer(host.get("Memory"), 1, 3 * GIB), "Fixture memory exceeds the reviewed bound") require(integer(host.get("MemorySwap"), host["Memory"], 6 * GIB), "Fixture swap exceeds the reviewed bound") require(integer(host.get("PidsLimit"), 1, 256), "Fixture PID limit exceeds the reviewed bound") nano, quota, period = (host.get(k, 0) for k in ("NanoCpus", "CpuQuota", "CpuPeriod")) require(integer(nano, 0, 2_000_000_000), "Invalid fixture CPU cap") require(type(quota) is int and quota >= 0 and type(period) is int and period >= 0, "Invalid fixture CPU quota") if quota: require(1000 <= period <= 1_000_000 and 1000 <= quota <= 2 * period, "Fixture CPU quota exceeds the reviewed bound") require(nano > 0 or quota > 0, "Positive bounded fixture CPU cap required") require(host.get("SecurityOpt") in (["no-new-privileges"], ["no-new-privileges=true"]), "Fixture security policy differs from the reviewed recipe") for key in ("EffectiveCaps", "BoundingCaps"): caps = item.get(key) require(isinstance(caps, list) and len(caps) <= len(CAPABILITIES) and all(isinstance(c, str) and c.removeprefix("CAP_") in CAPABILITIES for c in caps), "Fixture has unreviewed capabilities") require(not host.get("Devices") and not host.get("VolumesFrom"), "Host devices/volumes forbidden") scratch = host.get("Tmpfs") require(isinstance(scratch, dict) and set(scratch) == set(SCRATCH), "Unexpected fixture scratch mounts") for destination, options in scratch.items(): require(isinstance(options, str), "Invalid fixture scratch options") sizes = [option[5:] for option in options.split(',') if option.startswith('size=')] require(len(sizes) == 1, "One explicit scratch size required") match = re.fullmatch(r"([0-9]+)([kKmMgG]?)", sizes[0]) require(match is not None, "Invalid scratch size") size = int(match[1]) * {"": 1, "k": 1024, "m": 1024 ** 2, "g": GIB}[match[2].lower()] require(0 < size <= SCRATCH[destination], "Fixture scratch exceeds reviewed bound") mounts = item.get("Mounts", []) require(len([m for m in mounts if m.get("Type") == "bind"]) == 2, "Read-only probe and empty resolver required") require({m.get("Destination") for m in mounts if m.get("Type") == "bind"} == {"/contract.php", "/etc/resolv.conf"}, "Unexpected fixture bind destinations") for mount in mounts: if mount.get("Type") == "tmpfs": require(mount.get("Destination") in SCRATCH, "Unreviewed scratch destination") continue require(mount.get("Type") == "bind" and mount.get("RW") is False, "Only reviewed read-only files may be bound") if mount.get("Destination") == "/etc/resolv.conf": source = mount.get("Source", "") require(source.endswith("/empty-resolv.conf"), "Unexpected resolver scratch") try: metadata = os.stat(source, follow_symlinks=False) except OSError: raise ValueError("Resolver scratch is unavailable") from None require(stat.S_ISREG(metadata.st_mode) and metadata.st_size == 0 and metadata.st_nlink == 1, "Resolver must be an empty regular scratch file") else: require(mount.get("Source", "").endswith("/nextcloud-upstream-contract.php"), "Unexpected fixture probe") state = item.get("State", {}) pid = state.get("Pid") require(state.get("Running") is True and type(pid) is int and pid > 1, "Running fixture required") return pid def validate_cgroup(text): lines = text.splitlines() if len(text) > 256 or len(lines) != 4: raise ValueError("Missing bounded cgroup evidence") parts = lines[:3] + lines[3].split() if len(parts) != 5 or any(re.fullmatch(r"[0-9]+", p) is None for p in parts): raise ValueError("Unlimited or malformed cgroup evidence") memory, swap, pids, quota, period = map(int, parts) if not (0 < memory <= 3 * GIB and 0 <= swap <= 3 * GIB and 0 < pids <= 256 and 1000 <= period <= 1_000_000 and 1000 <= quota <= 2 * period): raise ValueError("Actual cgroup limits exceed the reviewed fixture bounds") if __name__ == "__main__": try: if sys.argv[1:] == ["--cgroup"]: validate_cgroup(sys.stdin.read(257)) print("Actual fixture cgroup bounds verified") sys.exit(0) if len(sys.argv) != 3: raise ValueError("Expected fixture name and immutable image") print(validate(json.load(sys.stdin), sys.argv[1], sys.argv[2])) except (ValueError, TypeError, KeyError, AttributeError) as error: print(f"Cloud fixture refused: {error}", file=sys.stderr) sys.exit(2)