#!/usr/bin/env python3 """Pure negative fixtures: no container engine, privileges or network access.""" import copy import importlib.util from pathlib import Path import unittest import tempfile SCRATCH = tempfile.TemporaryDirectory() RESOLVER = Path(SCRATCH.name) / "empty-resolv.conf" RESOLVER.write_bytes(b"") spec = importlib.util.spec_from_file_location("validator", Path(__file__).with_name("validate-cloud-test-fixture.py")) validator = importlib.util.module_from_spec(spec) spec.loader.exec_module(validator) NAME = "archy-cloud-nextcloud-contract-synthetic" IMAGE = "a" * 64 def fixture(): return [{"Name": NAME, "Image": IMAGE, "Config": {"Labels": {"org.archipelago.cloud-fixture": "nextcloud29-v1"}}, "EffectiveCaps": ["CAP_" + c for c in sorted(validator.CAPABILITIES)], "BoundingCaps": ["CAP_" + c for c in sorted(validator.CAPABILITIES)], "HostConfig": {"NetworkMode": "none", "PortBindings": {}, "ReadonlyRootfs": True, "Privileged": False, "SecurityOpt": ["no-new-privileges"], "IpcMode": "private", "Memory": 3 * validator.GIB, "MemorySwap": 6 * validator.GIB, "PidsLimit": 256, "NanoCpus": 2_000_000_000, "CpuQuota": 200000, "CpuPeriod": 100000, "Tmpfs": {p: f"rw,size={size}" for p, size in validator.SCRATCH.items()}}, "Mounts": [{"Type": "bind", "Destination": "/contract.php", "RW": False, "Source": "/synthetic/nextcloud-upstream-contract.php"}, {"Type": "bind", "Destination": "/etc/resolv.conf", "RW": False, "Source": str(RESOLVER)}], "State": {"Running": True, "Pid": 12345}}] class Validation(unittest.TestCase): def test_only_validated_pid_is_returned(self): self.assertEqual(validator.validate(fixture(), NAME, "sha256:" + IMAGE), 12345) def test_host_and_shared_networks_are_refused(self): for mode in ["host", "bridge", "pasta", "container:other", "", None]: with self.subTest(mode=mode): item = fixture(); item[0]["HostConfig"]["NetworkMode"] = mode with self.assertRaises(ValueError): validator.validate(item, NAME, IMAGE) def test_host_exposure_privilege_and_unbounded_resources_are_refused(self): changes = {"PortBindings": {"80/tcp": [{"HostPort": "8000"}]}, "PublishAllPorts": True, "Privileged": True, "ReadonlyRootfs": False, "PidMode": "host", "IpcMode": "host", "Memory": 0, "PidsLimit": -1, "Devices": ["synthetic"], "VolumesFrom": ["live"]} for key, value in changes.items(): with self.subTest(key=key): item = fixture(); item[0]["HostConfig"][key] = value with self.assertRaises(ValueError): validator.validate(item, NAME, IMAGE) def test_large_missing_malformed_and_unlimited_limits_are_refused(self): changes = {"Memory": [None, True, 1.0, -1, 3 * validator.GIB + 1, 2 ** 63 - 1], "MemorySwap": [None, True, 0, -1, 6 * validator.GIB + 1], "PidsLimit": [None, True, 1.0, 0, -1, 257, 2 ** 63 - 1], "NanoCpus": [None, True, -1, 2_000_000_001], "CpuQuota": [None, True, -1, 200001], "CpuPeriod": [None, True, 0, 1_000_001]} for key, values in changes.items(): for value in values: with self.subTest(key=key, value=value): item = fixture(); item[0]["HostConfig"][key] = value with self.assertRaises(ValueError): validator.validate(item, NAME, IMAGE) item = fixture(); item[0]["HostConfig"].update(NanoCpus=0, CpuQuota=0) with self.assertRaises(ValueError): validator.validate(item, NAME, IMAGE) item = fixture(); item[0]["HostConfig"]["NanoCpus"] = 0 self.assertEqual(validator.validate(item, NAME, IMAGE), 12345) def test_expanded_capabilities_and_unbounded_scratch_are_refused(self): for key in ("BoundingCaps", "EffectiveCaps"): for value in (None, ["CAP_SYS_ADMIN"], ["CAP_NET_ADMIN"], ["ALL"]): item = fixture(); item[0][key] = value with self.assertRaises(ValueError): validator.validate(item, NAME, IMAGE) for value in (None, [], ["no-new-privileges", "seccomp=unconfined"]): item = fixture(); item[0]["HostConfig"]["SecurityOpt"] = value with self.assertRaises(ValueError): validator.validate(item, NAME, IMAGE) for value in ("rw", "rw,size=0", "rw,size=3g", "rw,size=1g,size=2g", "rw,size=unlimited"): item = fixture(); item[0]["HostConfig"]["Tmpfs"]["/var/www/html"] = value with self.assertRaises(ValueError): validator.validate(item, NAME, IMAGE) item = fixture(); item[0]["HostConfig"]["Tmpfs"]["/home"] = "rw,size=1m" with self.assertRaises(ValueError): validator.validate(item, NAME, IMAGE) def test_additional_or_writable_host_mounts_are_refused(self): for mount in [{"Type": "volume", "Destination": "/data"}, {"Type": "bind", "Source": "/home", "Destination": "/home", "RW": False}, {"Type": "bind", "Source": "/run/podman/podman.sock", "Destination": "/socket", "RW": True}]: item = fixture(); item[0]["Mounts"].append(mount) with self.assertRaises(ValueError): validator.validate(item, NAME, IMAGE) for key, value in [("RW", True), ("Source", "/secret"), ("Destination", "/home")]: item = fixture(); item[0]["Mounts"][0][key] = value with self.assertRaises(ValueError): validator.validate(item, NAME, IMAGE) def test_host_resolver_data_is_never_admitted(self): try: RESOLVER.write_text("nameserver 192.0.2.1\n") with self.assertRaises(ValueError): validator.validate(fixture(), NAME, IMAGE) finally: RESOLVER.write_bytes(b"") item = fixture(); item[0]["Mounts"][1]["Source"] = "/etc/resolv.conf" with self.assertRaises(ValueError): validator.validate(item, NAME, IMAGE) item = fixture(); item[0]["Mounts"].pop() with self.assertRaises(ValueError): validator.validate(item, NAME, IMAGE) def test_identity_and_liveness_are_required(self): for key, value in [("Name", "live-nextcloud"), ("Image", "b" * 64), ("Config", {}), ("State", {"Running": False, "Pid": 12345}), ("State", {"Running": True, "Pid": 1}), ("State", {"Running": True, "Pid": "12345"})]: item = fixture(); item[0][key] = value with self.assertRaises(ValueError): validator.validate(item, NAME, IMAGE) for name, image in [("live-app", IMAGE), (NAME, "nextcloud:latest")]: with self.assertRaises(ValueError): validator.validate(fixture(), name, image) for items in [[], fixture() + copy.deepcopy(fixture())]: with self.assertRaises(ValueError): validator.validate(items, NAME, IMAGE) if __name__ == "__main__": unittest.main()