121 lines
7.0 KiB
Python
121 lines
7.0 KiB
Python
#!/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()
|