diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f8195e1..389730d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -19,6 +19,10 @@ on: description: Optional exact public Native source overlay SHA for candidate qualification type: string default: "" + published_activity_acceptance: + description: Run the reviewed published SDK/Server activity identity tuple + type: boolean + default: false permissions: contents: read @@ -28,6 +32,31 @@ concurrency: cancel-in-progress: ${{ github.event_name == 'pull_request' }} jobs: + published-activity-acceptance: + if: github.event_name == 'workflow_dispatch' && inputs.published_activity_acceptance + needs: [action-policy, lint] + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 + with: + persist-credentials: false + - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6 + with: + python-version: "3.12.16" + - name: Bootstrap patched pip trust + uses: durable-workflow/.github/.github/actions/bootstrap-patched-pip@75587fc33ff40c34f32ef777293dad56bb6f48c7 # clean-trust + with: + clients: pip,python + - name: Run the published activity identity case + run: python scripts/ci/published-activity-acceptance.py + - name: Retain published tuple and scenario evidence + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + with: + name: published-activity-acceptance-${{ github.sha }} + path: published-activity-results/ + retention-days: 7 action-policy: name: Central action policy preflight if: ${{ github.server_url == 'https://github.com' }} @@ -104,7 +133,9 @@ jobs: with: clients: pip,python - run: pip install -e '.[dev]' - - run: ruff check src/ tests/ + - run: >- + ruff check src/ tests/ scripts/ci/activity_acceptance_*.py + scripts/ci/verify_activity_install.py scripts/ci/published-activity-acceptance.py - run: mypy src/durable_workflow/ - name: Enforce load-calibrated Avro Value regression budget run: python benchmarks/avro_value.py --enforce diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index c80ab55..e7a2ce9 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -268,7 +268,7 @@ receipt = yield ctx.start_child_workflow( ) ``` -## Activity execution identity (unreleased) +## Activity execution identity Inside an activity, `activity.context().info.activity_execution_id` exposes the logical execution ID received from Server. `workflow_id` and `run_id` identify @@ -310,9 +310,8 @@ validated admission receipt's execution ID. Inline local activities have no admitted execution ID and expose `None`, while retaining their parent workflow and run IDs. Server-free mocks do not create Server activity identity. -This source addition is unreleased and is outside the current typed-value -release candidate. Published-package retry identity and execution-guarantees -guide acceptance remain tracked in +This accessor is available in Python SDK 2.15.0 and later. Published-Server retry +identity and execution-guarantees guide acceptance remain tracked in [the shared identity issue](https://github.com/durable-workflow/.github/issues/211). ## Local activities @@ -381,6 +380,13 @@ local attempts short; use `ctx.schedule_activity(...)` for separately queued work. Local activities are yielded one at a time; they are not members of a parallel workflow group. +For local activities, `max_attempts` counts all attempts, including the first. +When a nonempty `backoff_seconds` list runs out, later retries repeat its final +entry. For example, five attempts with `[1, 2]` wait `1, 2, 2, 2` seconds before +attempts two through five. An empty list adds no retry wait. A committed local +result replays its recorded attempts without rerunning the handler, including +older records made with a zero-delay tail. + ## Worker sessions A worker session routes related activities to one capable worker while its diff --git a/scripts/ci/activity_acceptance_consumer.py b/scripts/ci/activity_acceptance_consumer.py new file mode 100644 index 0000000..c0dcb11 --- /dev/null +++ b/scripts/ci/activity_acceptance_consumer.py @@ -0,0 +1,266 @@ +"""Install a closed wheel set, then await admission to the ordinary retry case.""" + +from __future__ import annotations + +import base64 +import hashlib +import importlib.machinery +import importlib.metadata +import ipaddress +import json +import os +import stat +import subprocess +import sys +import time +import urllib.request +from pathlib import Path +from urllib.parse import urlunsplit + +DATA = Path("/run/acceptance-data") +WHEELS = Path("/run/acceptance-wheels") +ENVIRONMENT = Path("/run/acceptance-env") +LIMIT = 64 * 1024 * 1024 +JUNIT = DATA / "identity.xml" +CAPTURED_VERIFIER = "" + + +def receive(maximum): + data = sys.stdin.buffer.readline(maximum + 1) + if not data.endswith(b"\n") or len(data) > maximum: + raise ValueError("consumer input exceeds its bound") + return json.loads(data) + + +def write(directory, name, item): + if not name or "/" in name or name in {".", ".."}: + raise ValueError("consumer input filename differs") + data = base64.b64decode(item["base64"], validate=True) + if hashlib.sha256(data).hexdigest() != item["sha256"]: + raise ValueError("consumer input digest differs") + descriptor = os.open(directory / name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) + try: + with os.fdopen(descriptor, "wb", closefd=False) as stream: + stream.write(data) + finally: + os.close(descriptor) + + +def run(arguments, environment, input_data=None): + return subprocess.run( + arguments, env=environment, input=input_data, capture_output=True, text=True, timeout=90, check=True + ).stdout + + +def validate_setup(packet): + if set(packet) != {"files", "wheels", "wheel_name", "wheel_sha256", "sdk_version"}: + raise ValueError("consumer setup shape differs") + expected_files = {"requirements.txt", "verify_activity_install.py", "test_activity_identity.py", "conftest.py"} + if set(packet["files"]) != expected_files or not 1 <= len(packet["wheels"]) <= 30: + raise ValueError("consumer setup membership differs") + if packet["wheel_name"] not in packet["wheels"]: + raise ValueError("approved SDK wheel is missing") + if packet["wheels"][packet["wheel_name"]]["sha256"] != packet["wheel_sha256"]: + raise ValueError("SDK wheel packet differs from approval") + + +def await_runtime(number): + if type(number) is not int or not 0 <= number <= 0xFFFFFFFF: + raise ValueError("consumer endpoint must be an unsigned IPv4 integer") + # The controller binds this numeric endpoint to the two inspected IDs. + address = ipaddress.IPv4Address(number) + if ( + not address.is_private + or address.is_loopback + or address.is_unspecified + or address.is_multicast + or address.is_reserved + or address.is_link_local + ): + raise ValueError("consumer endpoint must be a private numeric IPv4 address") + octets = [(number >> shift) & 255 for shift in (24, 16, 8, 0)] + numeric_host = ".".join(str(octet) for octet in octets) + url = urlunsplit(("http", numeric_host + ":8080", "", "", "")) + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + try: + with opener.open(url + "/api/health", timeout=2) as response: + if response.status == 200: + return url + except OSError: + time.sleep(0.2) + raise TimeoutError("published runtime readiness") + + +def retained_verifier(): + if not CAPTURED_VERIFIER: + raise ValueError("controller did not bind the Git-captured verifier") + namespace = {"__name__": "captured_sdk_inventory"} + exec(compile(CAPTURED_VERIFIER, "", "exec"), namespace) + return namespace["verify_sdk"] + + +def inventory(verifier, python, packet): + # The base interpreter, captured verifier and installer template remain in + # bootstrap memory. SDK execution cannot replace the terminal checker file. + from pip._vendor.distlib.scripts import SCRIPT_TEMPLATE + + site = ( + ENVIRONMENT + / "lib" + / ("python" + str(sys.version_info.major) + "." + str(sys.version_info.minor)) + / "site-packages" + ) + spec = importlib.machinery.PathFinder.find_spec("durable_workflow", [str(site)]) + if spec is None or spec.origin is None: + raise ValueError("installed SDK origin is unavailable") + wheel = base64.b64decode(packet["wheels"][packet["wheel_name"]]["base64"], validate=True) + result = verifier(site, spec.origin, wheel, packet["wheel_sha256"], python=python, script_template=SCRIPT_TEMPLATE) + versions = [ + item.version + for item in importlib.metadata.distributions(path=[str(site)]) + if item.metadata["Name"].lower().replace("_", "-") == "durable-workflow" + ] + if versions != [packet["sdk_version"]]: + raise ValueError("installed SDK version differs") + return result + + +def main(): + packet = prepare_inputs() + verifier = retained_verifier() + environment, python = prepare_environment() + before = inventory(verifier, python, packet) + ready = { + "sdk_inventory": before, + "python": sys.version, + "environment": str(ENVIRONMENT), + "distributions": run( + [ + str(python), + "-I", + "-B", + "-c", + "import importlib.metadata,json; print(json.dumps(sorted((d.metadata['Name']," + "d.version) for d in importlib.metadata.distributions())))", + ], + environment, + ), + } + print("ACTIVITY_CONSUMER_READY=" + json.dumps(ready, sort_keys=True), flush=True) + admission = receive(8192) + if set(admission) != {"runtime_ipv4", "token"}: + raise ValueError("consumer admission shape differs") + admission["server_url"] = await_runtime(admission["runtime_ipv4"]) + output = execute_case(verifier, python, packet, before, environment, admission) + print("ACTIVITY_CONSUMER_RESULT=" + json.dumps(output, sort_keys=True), flush=True) + # Remain attached while the controller rechecks the same two IDs and IP. + if receive(8192) != {"release": True}: + raise ValueError("consumer release shape differs") + if not output["complete"]: + raise RuntimeError("ordinary activity case failed") + + +def prepare_inputs(): + if os.getuid() != 1000: + raise ValueError("consumer must be unprivileged") + for directory in (DATA, WHEELS, ENVIRONMENT): + info = directory.lstat() + if not stat.S_ISDIR(info.st_mode) or info.st_uid != 1000 or info.st_mode & 0o077 or list(directory.iterdir()): + raise ValueError("consumer scratch is not empty and private") + packet = receive(LIMIT) + validate_setup(packet) + for name, item in packet["files"].items(): + write(DATA, name, item) + for name, item in packet["wheels"].items(): + if not name.endswith(".whl"): + raise ValueError("consumer accepts wheels only") + write(WHEELS, name, item) + return packet + + +def prepare_environment(): + environment = { + "PATH": os.defpath, + "PYTHONDONTWRITEBYTECODE": "1", + "PYTHONNOUSERSITE": "1", + "PIP_CONFIG_FILE": os.devnull, + "PIP_DISABLE_PIP_VERSION_CHECK": "1", + "PIP_NO_CACHE_DIR": "1", + "PIP_NO_INPUT": "1", + "TMPDIR": str(DATA), + } + python = ENVIRONMENT / "bin/python" + run([sys.executable, "-I", "-B", "-m", "venv", "--without-pip", "--copies", str(ENVIRONMENT)], environment) + run( + [ + sys.executable, + "-I", + "-B", + "-m", + "pip", + "--python", + str(python), + "install", + "--no-index", + "--find-links", + str(WHEELS), + "--only-binary=:all:", + "--require-hashes", + "--no-compile", + "-r", + str(DATA / "requirements.txt"), + ], + environment, + ) + return environment, python + + +def execute_case(verifier, python, packet, before, environment, admission): + environment.update( + { + "DURABLE_WORKFLOW_SERVER_URL": admission["server_url"], + "DURABLE_WORKFLOW_AUTH_TOKEN": admission["token"], + "DURABLE_WORKFLOW_EXPECT_INSTALLED_VERSION": packet["sdk_version"], + "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1", + } + ) + result = subprocess.run( + [ + str(python), + "-I", + "-B", + "-m", + "pytest", + "-c", + "/dev/null", + "-p", + "pytest_asyncio.plugin", + "-p", + "no:cacheprovider", + "--asyncio-mode=auto", + "--import-mode=importlib", + "-v", + "--capture=tee-sys", + str(DATA / "test_activity_identity.py"), + "--junitxml=" + str(JUNIT), + ], + env=environment, + capture_output=True, + text=True, + timeout=90, + ) + after = inventory(verifier, python, packet) + if after != before: + raise ValueError("installed SDK inventory changed") + return { + "complete": result.returncode == 0, + "sdk_inventory": after, + "case_output": result.stdout + result.stderr, + "junit": JUNIT.read_text() if JUNIT.is_file() else "", + } + + +if __name__ == "__main__": + main() diff --git a/scripts/ci/activity_acceptance_output.py b/scripts/ci/activity_acceptance_output.py new file mode 100644 index 0000000..8a2ab25 --- /dev/null +++ b/scripts/ci/activity_acceptance_output.py @@ -0,0 +1,94 @@ +"""Private no-follow result files retained independently of runtime cleanup.""" + +from __future__ import annotations + +import json +import os +import stat +from pathlib import Path + +RESULT_NAME = "result.json" + + +class Output: + def __init__(self, path): + self.path = Path(path) + self.path.mkdir(mode=0o700) + self.directory = os.open(self.path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + self.identity = os.fstat(self.directory) + self.files = {} + self.file_identities = {} + self.reserve(RESULT_NAME) + + def check(self): + current = self.path.lstat() + if (current.st_dev, current.st_ino) != (self.identity.st_dev, self.identity.st_ino): + raise ValueError("private output directory changed") + + def reserve(self, name): + if "/" in name or name in {"", ".", ".."}: + raise ValueError("output name must be one basename") + descriptor = os.open(name, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=self.directory) + self.files[name] = descriptor + info = os.fstat(descriptor) + self.file_identities[name] = (info.st_dev, info.st_ino, info.st_mode, info.st_uid, info.st_gid, info.st_nlink) + return descriptor + + def check_file(self, name): + descriptor = self.files[name] + current = os.stat(name, dir_fd=self.directory, follow_symlinks=False) + held = os.fstat(descriptor) + for info in (current, held): + shape = (info.st_dev, info.st_ino, info.st_mode, info.st_uid, info.st_gid, info.st_nlink) + if shape != self.file_identities[name] or not stat.S_ISREG(info.st_mode): + raise ValueError("private output file changed") + + def write(self, name, data): + self.check() + descriptor = self.files[name] if name in self.files else self.reserve(name) + self.check_file(name) + self._write(descriptor, data) + + @staticmethod + def _write(descriptor, data): + data = data.encode() if isinstance(data, str) else data + os.ftruncate(descriptor, 0) + offset = 0 + while offset < len(data): + offset += os.pwrite(descriptor, data[offset:], offset) + os.fsync(descriptor) + + def read_cid(self, name="runtime.cid"): + try: + descriptor = os.open(name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=self.directory) + except FileNotFoundError: + return "" + try: + info = os.fstat(descriptor) + if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or info.st_nlink != 1 or info.st_size > 128: + raise ValueError("CID file shape differs") + return os.pread(descriptor, 129, 0).decode("ascii") + finally: + os.close(descriptor) + + def finish(self, report): + # This held descriptor survives a path refusal; cleanup does not depend + # on traversing the output pathname to retain a terminal failure receipt. + try: + self.check() + self.check_file(RESULT_NAME) + except OSError as error: + report["output_failure"] = type(error).__name__ + report["complete"] = False + except ValueError: + report["output_failure"] = "identity_changed" + report["complete"] = False + encoded = json.dumps(report, indent=2) + "\n" + if len(encoded.encode()) > 128 * 1024: + raise ValueError("terminal result exceeded its bound") + self._write(self.files[RESULT_NAME], encoded) + + def close(self): + for descriptor in self.files.values(): + os.close(descriptor) + os.close(self.directory) diff --git a/scripts/ci/activity_acceptance_runtime.py b/scripts/ci/activity_acceptance_runtime.py new file mode 100644 index 0000000..d2695f4 --- /dev/null +++ b/scripts/ci/activity_acceptance_runtime.py @@ -0,0 +1,410 @@ +"""Exact container and network authority for the ordinary published SDK cell.""" + +from __future__ import annotations + +import ipaddress +import re +from pathlib import Path + +OWNER = "dev.durable-workflow.fixture-owner" +RUN = "dev.durable-workflow.fixture-run" +ROLE = "dev.durable-workflow.fixture-role" +PRIVATE_64M = "noexec,nosuid,nodev,uid=1000,gid=1000,mode=0700,size=64m" +TMPFS = { + "runtime": { + "/run/runtime-tmp": "noexec,nosuid,nodev,uid=1000,gid=1000,mode=0700,size=128m", + "/app/database": "noexec,nosuid,nodev,uid=1000,gid=1000,mode=0700,size=128m", + "/app/storage": PRIVATE_64M, + }, + "consumer": { + "/run/acceptance-data": PRIVATE_64M, + "/run/acceptance-wheels": PRIVATE_64M, + "/run/acceptance-env": "exec,nosuid,nodev,uid=1000,gid=1000,mode=0700,size=256m", + }, +} +ENVIRONMENT = { + "TMPDIR": "/run/runtime-tmp", + "DB_CONNECTION": "sqlite", + "DB_DATABASE": "/app/database/activity-identity.sqlite", + "DW_AUTH_TOKEN": "published-activity-test-token", + "DW_RUNTIME_CREDENTIALS_ENABLED": "false", +} + + +def full_id(value): + return isinstance(value, str) and re.fullmatch(r"[0-9a-f]{64}", value) is not None + + +def created_identity(stdout, cidfile): + """Retain both valid channels for cleanup, even when admission must refuse.""" + captured, printed = cidfile.strip(), stdout.strip() + identity = None + if full_id(captured): + identity = captured + elif full_id(printed): + identity = printed + candidates = sorted({value for value in (captured, printed) if full_id(value)}) + return identity, bool(identity and printed == identity and captured == identity), candidates + + +def require_owner(record, identity, nonce, role): + labels = record["Config"]["Labels"] or {} + if ( + not full_id(identity) + or record["Id"] != identity + or labels.get(OWNER) != "sdk-python" + or labels.get(RUN) != nonce + or labels.get(ROLE) != role + ): + raise ValueError("container cleanup ownership differs") + + +def verify_network(record, identity, nonce): + expected = { + "Id": identity, + "Driver": "bridge", + "Internal": True, + "Attachable": False, + "Ingress": False, + "EnableIPv6": False, + } + labels = record["Labels"] or {} + if ( + not full_id(identity) + or any(record.get(key) != value for key, value in expected.items()) + or labels.get(OWNER) != "sdk-python" + or labels.get(RUN) != nonce + or labels.get(ROLE) != "network" + ): + raise ValueError("owned internal network differs") + + +def expected_command(image, role, bootstrap): + if role == "consumer": + return ["/usr/local/bin/python3"], ["-I", "-B", "-c", bootstrap], "/" + return image["Config"].get("Entrypoint"), image["Config"].get("Cmd"), image["Config"].get("WorkingDir", "") + + +def bind_local_image(index_record, platform_record, index_digest, child_digest, config_digest): + if platform_record["Architecture"] != "amd64" or platform_record["Os"] != "linux": + raise ValueError("local platform image differs") + for record in (index_record, platform_record): + if not any(value.rsplit("@", 1)[-1] == index_digest for value in record["RepoDigests"]): + raise ValueError("local image lacks the selected immutable index") + descriptor = platform_record.get("Descriptor") + if index_record["Id"] == config_digest and platform_record["Id"] == config_digest: + store = "classic" + elif index_record["Id"] == index_digest and platform_record["Id"] == child_digest: + store = "containerd" + else: + raise ValueError("local image IDs do not bind the selected manifest/config") + if (store == "containerd" or descriptor is not None) and ( + descriptor is None + or descriptor.get("digest") != child_digest + or descriptor.get("platform") != {"architecture": "amd64", "os": "linux"} + ): + raise ValueError("local platform descriptor differs") + return { + **platform_record, + "CreateReference": index_record["Id"], + "PlatformConfigDigest": config_digest, + "ImageStore": store, + } + + +def environment_values(entries): + values = {} + for entry in entries: + key, value = entry.split("=", 1) + values[key] = value + return values + + +def verify_config(config, identity, nonce, image, role, bootstrap): + environment = environment_values(image["Config"]["Env"]) + if role == "runtime": + environment.update(ENVIRONMENT) + received = environment_values(config["Env"]) + if len(received) != len(config["Env"]) or received != environment: + raise ValueError("container environment differs") + entrypoint, command, directory = expected_command(image, role, bootstrap) + labels = {**(image["Config"].get("Labels") or {}), OWNER: "sdk-python", RUN: nonce, ROLE: role} + expected = { + "Hostname": identity[:12], + "Domainname": "", + "Image": image.get("CreateReference", image["Id"]), + "User": "1000:1000", + "Entrypoint": entrypoint, + "Cmd": command, + "WorkingDir": directory, + "Tty": False, + "OpenStdin": role == "consumer", + "StdinOnce": role == "consumer", + "AttachStdin": role == "consumer", + "AttachStdout": True, + "AttachStderr": True, + "Labels": labels, + "Volumes": image["Config"].get("Volumes"), + "ExposedPorts": image["Config"].get("ExposedPorts"), + "Healthcheck": image["Config"].get("Healthcheck"), + "StopSignal": image["Config"].get("StopSignal", ""), + "StopTimeout": image["Config"].get("StopTimeout"), + "Shell": image["Config"].get("Shell"), + "OnBuild": image["Config"].get("OnBuild"), + "ArgsEscaped": image["Config"].get("ArgsEscaped", False), + "NetworkDisabled": False, + "MacAddress": "", + } + if set(config) - (set(expected) | {"Env"}): + raise ValueError("container has an unreviewed Config field") + optional = { + "ExposedPorts": None, + "Healthcheck": None, + "StopSignal": "", + "StopTimeout": None, + "Shell": None, + "OnBuild": None, + "ArgsEscaped": False, + "NetworkDisabled": False, + "MacAddress": "", + } + if (set(expected) - set(optional)) - set(config): + raise ValueError("container Config lacks a required observation") + for key, value in expected.items(): + received_value = config.get(key, optional[key]) if key in optional else config[key] + if received_value != value: + raise ValueError("container Config differs from the closed image/role projection") + + +def verify_container(record, identity, nonce, image, network, child_digest, role, bootstrap="", *, cidfile=""): + require_owner(record, identity, nonce, role) + verify_config(record["Config"], identity, nonce, image, role, bootstrap) + if record["Image"] != image.get("CreateReference", image["Id"]): + raise ValueError("container image identity differs") + descriptor = record.get("ImageManifestDescriptor") + if descriptor is not None and descriptor.get("digest") != child_digest: + raise ValueError("container platform child differs") + if image.get("ImageStore") == "containerd" and ( + descriptor is None or descriptor.get("platform") != {"os": "linux", "architecture": "amd64"} + ): + raise ValueError("containerd platform descriptor is missing or different") + verify_host(record["HostConfig"], role, network, cidfile=cidfile) + if any(mount["Type"] != "tmpfs" or mount["Destination"] not in TMPFS[role] for mount in record["Mounts"]): + raise ValueError("container has an undeclared mount") + + +MASKED_PATHS = { + "/proc/acpi", + "/proc/asound", + "/proc/interrupts", + "/proc/kcore", + "/proc/keys", + "/proc/latency_stats", + "/proc/sched_debug", + "/proc/scsi", + "/proc/timer_list", + "/proc/timer_stats", + "/sys/devices/virtual/powercap", + "/sys/firmware", +} +READONLY_PATHS = {"/proc/bus", "/proc/fs", "/proc/irq", "/proc/sys", "/proc/sysrq-trigger"} + + +def expected_masked_paths(): + # The fixed local daemon also masks the kernel's CPU thermal-throttle paths. + return MASKED_PATHS | {str(path) for path in Path("/sys/devices/system/cpu").glob("cpu[0-9]*/thermal_throttle")} + + +def verify_protected_paths(host): + for field, expected in (("MaskedPaths", expected_masked_paths()), ("ReadonlyPaths", READONLY_PATHS)): + actual = host.get(field) + if not isinstance(actual, list) or len(actual) != len(set(actual)) or set(actual) != expected: + raise ValueError("container protected system paths differ") + + +def verify_host(host, role, network, *, cidfile=""): + expected = { + "NetworkMode": network, + "ReadonlyRootfs": True, + "Privileged": False, + "NanoCpus": 1_000_000_000, + "Memory": 1_073_741_824, + "MemorySwap": 1_073_741_824, + "PidsLimit": 256, + "Init": True, + "CapDrop": ["ALL"], + "SecurityOpt": ["no-new-privileges"], + "Tmpfs": TMPFS[role], + "PublishAllPorts": False, + "AutoRemove": False, + "ContainerIDFile": cidfile, + "ConsoleSize": [0, 0], + "MemorySwappiness": None, + "ShmSize": 67_108_864, + "Runtime": "runc", + "IpcMode": "private", + "CgroupnsMode": "private", + "RestartPolicy": {"Name": "no", "MaximumRetryCount": 0}, + "LogConfig": {"Type": "json-file", "Config": {"max-size": "1m", "max-file": "1"}}, + } + if any(host.get(key) != value for key, value in expected.items()): + raise ValueError("container resource/network/security configuration differs") + if "OomKillDisable" not in host or (host["OomKillDisable"] is not False and host["OomKillDisable"] is not None): + raise ValueError("container disables the OOM killer or lacks its lifecycle value") + verify_protected_paths(host) + zero_fields = ( + "IOMaximumIOps", + "IOMaximumBandwidth", + "CpuShares", + "CpuPeriod", + "CpuQuota", + "CpuRealtimePeriod", + "CpuRealtimeRuntime", + "OomScoreAdj", + "MemoryReservation", + "BlkioWeight", + "CpuCount", + "CpuPercent", + "KernelMemory", + "KernelMemoryTCP", + ) + if any(host.get(key, 0) != 0 for key in zero_fields): + raise ValueError("container has an undeclared resource override") + forbidden = ( + "Annotations", + "Binds", + "VolumesFrom", + "Devices", + "DeviceRequests", + "CapAdd", + "ExtraHosts", + "PortBindings", + "PidMode", + "UTSMode", + "UsernsMode", + "Links", + "Mounts", + "VolumeDriver", + "Isolation", + "GroupAdd", + "Sysctls", + "DeviceCgroupRules", + "CgroupParent", + "Cgroup", + "CpusetCpus", + "CpusetMems", + "Ulimits", + "Dns", + "DnsOptions", + "DnsSearch", + "StorageOpt", + "BlkioWeightDevice", + "BlkioDeviceReadBps", + "BlkioDeviceWriteBps", + "BlkioDeviceReadIOps", + "BlkioDeviceWriteIOps", + ) + known = set(expected) | set(zero_fields) | set(forbidden) | {"OomKillDisable", "MaskedPaths", "ReadonlyPaths"} + if set(host) - known: + raise ValueError("container has an unreviewed HostConfig field") + if any(host.get(key) for key in forbidden): + raise ValueError("container has an undeclared host resource") + + +def endpoint(record, network): + endpoints = list(record["NetworkSettings"]["Networks"].values()) + if len(endpoints) != 1 or endpoints[0]["NetworkID"] != network: + raise ValueError("container network attachment differs") + address = ipaddress.ip_address(endpoints[0]["IPAddress"]) + if address.version != 4 or not address.is_private or address.is_loopback or address.is_unspecified: + raise ValueError("runtime address is not a private numeric IPv4 endpoint") + return str(address) + + +def verify_attachments(network, runtime_record, consumer_record): + identities = {runtime_record["Id"], consumer_record["Id"]} + if set(network["Containers"]) != identities: + raise ValueError("internal network has missing or foreign attachments") + address = endpoint(runtime_record, network["Id"]) + consumer_address = endpoint(consumer_record, network["Id"]) + if address == consumer_address: + raise ValueError("container addresses are not distinct") + for record in (runtime_record, consumer_record): + actual = endpoint(record, network["Id"]) + if network["Containers"][record["Id"]]["IPv4Address"].split("/", 1)[0] != actual: + raise ValueError("network and container endpoint observations differ") + subnets = [ipaddress.ip_network(item["Subnet"]) for item in network["IPAM"]["Config"]] + if not any(ipaddress.ip_address(address) in subnet for subnet in subnets): + raise ValueError("runtime address differs from network IPAM") + return address + + +def create_arguments(image, network, nonce, cidfile, role, bootstrap=""): + arguments = [ + "docker", + "create", + "--pull=never", + "--platform", + "linux/amd64", + "--cidfile", + str(cidfile), + "--name", + "sdk-python-activity-" + role + "-" + nonce[:12], + ] + for key, value in ((OWNER, "sdk-python"), (RUN, nonce), (ROLE, role)): + arguments.extend(["--label", key + "=" + value]) + arguments.extend( + [ + "--init", + "--read-only", + "--network", + network, + "--cpus", + "1", + "--memory", + "1g", + "--memory-swap", + "1g", + "--pids-limit", + "256", + "--user", + "1000:1000", + "--cap-drop", + "ALL", + "--security-opt", + "no-new-privileges", + "--ipc", + "private", + "--cgroupns", + "private", + "--restart", + "no", + "--runtime", + "runc", + "--log-driver", + "json-file", + "--log-opt", + "max-size=1m", + "--log-opt", + "max-file=1", + ] + ) + for target, options in TMPFS[role].items(): + arguments.extend(["--tmpfs", target + ":" + options]) + if role == "consumer": + return [ + *arguments, + "-i", + "--workdir", + "/", + "--entrypoint", + "/usr/local/bin/python3", + image.get("CreateReference", image["Id"]), + "-I", + "-B", + "-c", + bootstrap, + ] + for key, value in ENVIRONMENT.items(): + arguments.extend(["-e", key + "=" + value]) + return [*arguments, image.get("CreateReference", image["Id"])] diff --git a/scripts/ci/activity_acceptance_session.py b/scripts/ci/activity_acceptance_session.py new file mode 100644 index 0000000..7f768c7 --- /dev/null +++ b/scripts/ci/activity_acceptance_session.py @@ -0,0 +1,95 @@ +"""Bounded stdin/stdout exchange with the closed, already-owned consumer.""" + +from __future__ import annotations + +import json +import os +import selectors +import subprocess +import time +from collections import deque + + +class Session: + def __init__(self, identity, environment): + self.process = subprocess.Popen( + ["docker", "start", "-a", "-i", identity], + env=environment, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + self.selector = selectors.DefaultSelector() + os.set_blocking(self.process.stdin.fileno(), False) + os.set_blocking(self.process.stdout.fileno(), False) + self.selector.register(self.process.stdout, selectors.EVENT_READ) + self.deadline = time.monotonic() + 240 + self.output = bytearray() + self.pending = bytearray() + self.lines = deque() + self.eof = False + + def read_available(self): + chunk = os.read(self.process.stdout.fileno(), 65536) + if not chunk: + self.eof = True + self.selector.unregister(self.process.stdout) + return + self.output.extend(chunk) + if len(self.output) > 2 * 1024 * 1024: + raise ValueError("consumer output exceeds its bound") + self.pending.extend(chunk) + while b"\n" in self.pending: + line, _, tail = self.pending.partition(b"\n") + self.pending = bytearray(tail) + self.lines.append(line.decode()) + + def events(self): + remaining = self.deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("closed consumer deadline") + return self.selector.select(min(remaining, 1)) + + def send(self, value): + data = json.dumps(value, separators=(",", ":")).encode() + b"\n" + if len(data) > 64 * 1024 * 1024: + raise ValueError("consumer setup exceeds its bound") + self.selector.register(self.process.stdin, selectors.EVENT_WRITE) + offset = 0 + try: + while offset < len(data): + for key, _ in self.events(): + if key.fileobj is self.process.stdout: + self.read_available() + else: + offset += os.write(self.process.stdin.fileno(), data[offset : offset + 65536]) + if self.eof: + raise RuntimeError("consumer exited before receiving input") + finally: + self.selector.unregister(self.process.stdin) + + def receive(self, marker): + while True: + while self.lines: + line = self.lines.popleft() + if line.startswith(marker): + return json.loads(line[len(marker) :]) + if self.eof: + raise RuntimeError("consumer exited without its bounded result") + for _key, _ in self.events(): + self.read_available() + + def finish(self): + self.process.stdin.close() + while not self.eof: + for _key, _ in self.events(): + self.read_available() + return self.process.wait(timeout=10) + + def close(self): + if self.process.poll() is None: + self.process.kill() + self.process.wait(timeout=10) + self.process.stdin.close() + self.process.stdout.close() + self.selector.close() diff --git a/scripts/ci/published-activity-acceptance.md b/scripts/ci/published-activity-acceptance.md new file mode 100644 index 0000000..21f6d4c --- /dev/null +++ b/scripts/ci/published-activity-acceptance.md @@ -0,0 +1,50 @@ +The ordinary activity-identity acceptance cell installs the exact reviewed +registry wheel and uses the immutable published Server tuple in +`tests/fixtures/published-activity-acceptance.json`. It reads identity inside +real activity callbacks, deliberately fails the first attempt, and checks the +logical execution/workflow/run identity across the Server retry while task and +attempt identities change. + +After source review and the normal PR gates, run it in GitHub Actions: + +```bash +gh workflow run ci.yml --ref main -f published_activity_acceptance=true +``` + +The checked-in release approval supplies the SDK wheel digest. The complete +dependency set, including pytest and its plugin, is hash locked. Downloads +happen before the SDK consumer starts. The consumer installs those captured +wheels without network access into a fresh private environment, checks every +SDK wheel member, complete distribution metadata and installed RECORD, and uses +isolated Python with explicit pytest plugins. Installer-created metadata and +console scripts have exact expected bytes. The checker stays in the trusted +bootstrap's memory through the case. The checkout is never installed as the SDK. + +The published native runtime and official Python consumer share one owned +internal network. Neither publishes a port or receives a host mount. The +controller checks both full container IDs, owner/role/run labels, image config +digests, resource limits and network attachments before admitting SDK activity +and after the result. The numeric runtime endpoint must remain unchanged. Each +container has a read-only root, an unprivileged user, all capabilities dropped, +no new privileges, one CPU, 1 GiB RAM with no additional swap and 256 PIDs. +Writable temporary storage stays inside that memory limit. + +Config and HostConfig use closed projections. Healthchecks must match the +published image, and protected system paths must match the local daemon's +reviewed defaults, including kernel-exposed thermal-throttle paths. Docker 29's +observed false/null OOM-disable transition is accepted; true still refuses. +Platform-specific inspection binds containerd's index/child identities or a +classic store's config identity to the selected immutable manifest chain. + +Central action-policy and lint gates precede execution. The controller has a +six-minute execution bound, followed by independent cleanup of each captured +ID and then the network. Label-bound reconciliation covers uncertain create +responses. A foreign replacement is preserved. A reserved result descriptor +retains a terminal failure even if the output path refuses further use. + +The seven-day artifact contains the selected tuple, complete wheel hashes, +dependency versions, SDK inventory, before/after container and network facts, +actual callback observations, JUnit, logs and cleanup result. An installed +package or a source integration pass alone is not this published-tuple proof. +The cell does not qualify local/prepared/session activity behavior or external +side-effect recovery. diff --git a/scripts/ci/published-activity-acceptance.py b/scripts/ci/published-activity-acceptance.py new file mode 100644 index 0000000..7e7bb2b --- /dev/null +++ b/scripts/ci/published-activity-acceptance.py @@ -0,0 +1,479 @@ +"""Run one exact published SDK retry-identity case inside an owned internal network.""" + +from __future__ import annotations + +import base64 +import hashlib +import ipaddress +import json +import os +import re +import signal +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path +from uuid import uuid4 +from xml.etree import ElementTree + +try: + from scripts.ci import activity_acceptance_output as output_files + from scripts.ci import activity_acceptance_runtime as runtime + from scripts.ci.activity_acceptance_session import Session +except ModuleNotFoundError as error: + if error.name != "scripts": + raise + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from scripts.ci import activity_acceptance_output as output_files + from scripts.ci import activity_acceptance_runtime as runtime + from scripts.ci.activity_acceptance_session import Session + +ROOT = Path(__file__).resolve().parents[2] +OUTPUT = ROOT / "published-activity-results" +REQUIREMENTS = "requirements.txt" +VERIFIER_FILENAME = "verify_activity_install.py" +OWNER_VALUE = "=sdk-python" +LABEL_FILTER = "label=" +ID_FORMAT = "{{.ID}}" +# No inherited context, registry credentials, proxy, SSH or remote Docker endpoint. +# This kernel directory is private to each CLI process and contains no config.json. +DOCKER_ENVIRONMENT = { + "PATH": os.defpath, + "DOCKER_HOST": "unix:///var/run/docker.sock", + "DOCKER_CONFIG": "/proc/self/fd", +} + + +def call(*arguments, check=True, **kwargs): + if arguments[0] == "docker": + kwargs["env"] = DOCKER_ENVIRONMENT + return subprocess.run(arguments, check=check, capture_output=True, text=True, timeout=180, **kwargs) + + +def cleanup_call(*arguments, check=True): + return subprocess.run(arguments, check=check, capture_output=True, text=True, timeout=10, env=DOCKER_ENVIRONMENT) + + +def deadline(_signum, _frame): + raise TimeoutError("published cell reached its six-minute execution bound") + + +def verify_distributions(ready, requirements): + expected = sorted( + (name.lower().replace("_", "-"), version) + for name, version in re.findall(r"^([A-Za-z0-9_.-]+)==([^\s]+)", requirements.decode(), re.MULTILINE) + ) + actual = sorted((name.lower().replace("_", "-"), version) for name, version in json.loads(ready["distributions"])) + if not expected or actual != expected: + raise ValueError("consumer environment differs from the complete dependency lock") + + +def verify_case_result(result): + if not result["complete"]: + raise ValueError("published retry case reported failure") + root = ElementTree.fromstring(result["junit"]) + cases = list(root.iter("testcase")) + expected = "test_retry_preserves_received_execution_id_and_changes_attempt_id" + if len(cases) != 1 or cases[0].get("name") != expected: + raise ValueError("published retry case inventory differs") + if any(cases[0].find(tag) is not None for tag in ("failure", "error", "skipped")): + raise ValueError("published retry case did not pass") + + +def engine(): + observed = json.loads(call("docker", "info", "--format", "{{json .}}").stdout) + if observed["OSType"] != "linux" or observed["Architecture"] not in {"x86_64", "amd64"}: + raise ValueError("published cell requires the local Linux amd64 engine") + return {key: observed[key] for key in ("ID", "ServerVersion", "OSType", "Architecture")} + + +def packet_file(data): + return {"base64": base64.b64encode(data).decode(), "sha256": hashlib.sha256(data).hexdigest()} + + +def bind_bootstrap(captured): + source = captured["bootstrap.py"].decode() + placeholder = 'CAPTURED_VERIFIER = ""' + if source.count(placeholder) != 1: + raise ValueError("consumer verifier binding placeholder differs") + return source.replace(placeholder, "CAPTURED_VERIFIER = " + repr(captured[VERIFIER_FILENAME].decode()), 1) + + +def capture_source(report): + commit = call("git", "rev-parse", "HEAD", cwd=ROOT).stdout.strip() + if re.fullmatch(r"[0-9a-f]{40}", commit) is None or os.environ.get("GITHUB_SHA", commit) != commit: + raise ValueError("checkout differs from the selected Actions commit") + paths = { + REQUIREMENTS: "scripts/ci/published-activity-requirements.txt", + VERIFIER_FILENAME: "scripts/ci/verify_activity_install.py", + "test_activity_identity.py": "tests/integration/test_activity_identity.py", + "conftest.py": "tests/integration/conftest.py", + "bootstrap.py": "scripts/ci/activity_acceptance_consumer.py", + "tuple.json": "tests/fixtures/published-activity-acceptance.json", + } + captured = { + name: call("git", "show", commit + ":" + path, cwd=ROOT).stdout.encode() for name, path in paths.items() + } + report["source_commit"] = commit + report["source_files"] = {paths[name]: hashlib.sha256(data).hexdigest() for name, data in captured.items()} + return captured + + +def download_wheels(selected, captured, files): + approval_path = ".github/release-candidates/" + selected["sdk_version"] + ".json" + approval = json.loads(call("git", "show", "HEAD:" + approval_path, cwd=ROOT).stdout) + wheel_name = next(name for name in approval["files"] if name.endswith(".whl")) + files.write(REQUIREMENTS, captured[REQUIREMENTS]) + files.check() + wheelhouse = OUTPUT / "wheels" + wheelhouse.mkdir(mode=0o700) + environment = { + "PATH": os.defpath, + "PIP_CONFIG_FILE": os.devnull, + "PIP_DISABLE_PIP_VERSION_CHECK": "1", + "PYTHONDONTWRITEBYTECODE": "1", + "PIP_INDEX_URL": "https://pypi.org/simple", + **{key: os.environ[key] for key in ("SSL_CERT_FILE", "PIP_CERT", "REQUESTS_CA_BUNDLE") if key in os.environ}, + } + result = call( + sys.executable, + "-I", + "-m", + "pip", + "download", + "--require-hashes", + "--only-binary=:all:", + "--dest", + str(wheelhouse), + "-r", + str(OUTPUT / REQUIREMENTS), + env=environment, + ) + files.write("download.log", result.stdout + result.stderr) + files.check() + wheels = {} + for path in sorted(wheelhouse.iterdir()): + if path.is_symlink() or not path.is_file() or path.suffix != ".whl": + raise ValueError("downloaded wheel membership differs") + wheels[path.name] = packet_file(path.read_bytes()) + if wheels[wheel_name]["sha256"] != approval["files"][wheel_name]: + raise ValueError("registry wheel differs from the independently reviewed release") + return { + "files": { + name: packet_file(captured[name]) + for name in (REQUIREMENTS, VERIFIER_FILENAME, "test_activity_identity.py", "conftest.py") + }, + "wheels": wheels, + "wheel_name": wheel_name, + "wheel_sha256": approval["files"][wheel_name], + "sdk_version": selected["sdk_version"], + } + + +def verify_published_image(selected, role, files): + prefix = "server" if role == "runtime" else "python" + image, child_digest = selected[prefix + "_image"], selected[prefix + "_amd64_manifest"] + call("docker", "pull", "--platform", "linux/amd64", image) + index_text = call("docker", "buildx", "imagetools", "inspect", "--raw", image).stdout + child_text = call( + "docker", "buildx", "imagetools", "inspect", "--raw", image.split("@", 1)[0] + "@" + child_digest + ).stdout + index, child = json.loads(index_text), json.loads(child_text) + platforms = [ + item["digest"] + for item in index["manifests"] + if item.get("platform") == {"architecture": "amd64", "os": "linux"} + ] + if platforms != [child_digest]: + raise ValueError("published index differs from the selected platform child") + local_index = json.loads(call("docker", "image", "inspect", image).stdout)[0] + local_platform = json.loads(call("docker", "image", "inspect", "--platform", "linux/amd64", image).stdout)[0] + observed = runtime.bind_local_image( + local_index, local_platform, image.rsplit("@", 1)[-1], child_digest, child["config"]["digest"] + ) + if role == "runtime": + if observed["Config"]["Labels"]["org.opencontainers.image.revision"] != selected["server_source"]: + raise ValueError("runtime source differs") + elif "PYTHON_VERSION=" + selected["python_version"] not in observed["Config"]["Env"]: + raise ValueError("consumer interpreter image differs") + files.write(role + "-image.json", json.dumps(observed, indent=2)) + files.write(role + "-local-index.json", json.dumps(local_index, indent=2)) + files.write(role + "-index.json", index_text) + files.write(role + "-manifest.json", child_text) + return observed + + +def absent(result, identity): + return result.returncode == 1 and result.stderr.strip().lower() == "error: no such object: " + identity + + +class OwnedDocker: + def __init__(self, nonce, files, report): + self.nonce, self.files, self.report = nonce, files, report + self.containers = {"runtime": set(), "consumer": set()} + self.networks = set() + self.network = None + self.identities = {} + + def inspect(self, kind, identity): + command = ["docker", "network", "inspect", identity] if kind == "network" else ["docker", "inspect", identity] + return json.loads(call(*command).stdout)[0] + + def labels(self, role): + return [ + "--label", + runtime.OWNER + OWNER_VALUE, + "--label", + runtime.RUN + "=" + self.nonce, + "--label", + runtime.ROLE + "=" + role, + ] + + def create_network(self): + result = call( + "docker", + "network", + "create", + "--internal", + "--driver", + "bridge", + *self.labels("network"), + "sdk-python-activity-" + self.nonce, + check=False, + ) + identity = result.stdout.strip() + if runtime.full_id(identity): + self.networks.add(identity) + result.check_returncode() + if not runtime.full_id(identity): + raise ValueError("network create did not return a full ID") + runtime.verify_network(self.inspect("network", identity), identity, self.nonce) + self.network = identity + self.report["network_id"] = identity + + def create(self, role, image, child, bootstrap): + self.files.check() + cid_name = role + ".cid" + result = call( + *runtime.create_arguments(image, self.network, self.nonce, OUTPUT / cid_name, role, bootstrap), check=False + ) + identity, agrees, candidates = runtime.created_identity(result.stdout, self.files.read_cid(cid_name)) + self.containers[role].update(candidates) + result.check_returncode() + if not agrees: + raise ValueError("container create channels disagree or omit the full ID") + self.files.check() + self.identities[role] = identity + self.report[role + "_id"] = identity + self.verify(role, image, child, bootstrap) + return identity + + def verify(self, role, image, child, bootstrap): + identity = self.identities[role] + record = self.inspect("container", identity) + runtime.verify_container( + record, + identity, + self.nonce, + image, + self.network, + child, + role, + bootstrap, + cidfile=str(OUTPUT / (role + ".cid")), + ) + return record + + def boundary(self, images, selected, bootstrap, phase): + network = self.inspect("network", self.network) + runtime.verify_network(network, self.network, self.nonce) + records = { + role: self.verify(role, images[role], selected[prefix + "_amd64_manifest"], bootstrap) + for role, prefix in (("runtime", "server"), ("consumer", "python")) + } + if any(not record["State"]["Running"] or record["State"]["OOMKilled"] for record in records.values()): + raise ValueError("both owned containers must remain running without OOM") + url = runtime.verify_attachments(network, records["runtime"], records["consumer"]) + self.files.write(phase + "-authority.json", json.dumps({"network": network, **records}, indent=2)) + return url + + def discover(self): + filters = [ + "--filter", + LABEL_FILTER + runtime.OWNER + OWNER_VALUE, + "--filter", + LABEL_FILTER + runtime.RUN + "=" + self.nonce, + ] + for role in self.containers: + result = cleanup_call( + "docker", + "ps", + "-a", + "--no-trunc", + *filters, + "--filter", + LABEL_FILTER + runtime.ROLE + "=" + role, + "--format", + ID_FORMAT, + ) + self.containers[role].update(value for value in result.stdout.splitlines() if runtime.full_id(value)) + result = cleanup_call( + "docker", + "network", + "ls", + "--no-trunc", + *filters, + "--filter", + LABEL_FILTER + runtime.ROLE + "=network", + "--format", + ID_FORMAT, + ) + self.networks.update(value for value in result.stdout.splitlines() if runtime.full_id(value)) + + def remove_container(self, role, identity): + result = cleanup_call("docker", "inspect", identity, check=False) + if absent(result, identity): + return {"absent": True} + result.check_returncode() + record = json.loads(result.stdout)[0] + runtime.require_owner(record, identity, self.nonce, role) + try: + logs = cleanup_call("docker", "logs", identity, check=False) + self.files.write(role + "-" + identity[:12] + ".log", logs.stdout + logs.stderr) + except (OSError, ValueError): + self.report["log_retention_failed"] = True + cleanup_call("docker", "rm", "-f", "-v", identity) + result = cleanup_call("docker", "inspect", identity, check=False) + return { + "absent": absent(result, identity), + "oom_killed": record["State"]["OOMKilled"], + "exit_code": record["State"]["ExitCode"], + } + + def remove_network(self, identity): + record = json.loads(cleanup_call("docker", "network", "inspect", identity).stdout)[0] + runtime.verify_network(record, identity, self.nonce) + if record["Containers"]: + raise ValueError("network retains an attachment; refusing removal") + cleanup_call("docker", "network", "rm", identity) + result = cleanup_call( + "docker", "network", "ls", "--no-trunc", "--filter", "id=" + identity, "--format", ID_FORMAT + ) + return {"absent": not result.stdout.strip()} + + def cleanup(self): + outcomes = {} + try: + self.discover() + except Exception as error: + outcomes["reconciliation"] = {"error": type(error).__name__} + self.cleanup_network_boundaries(outcomes) + # Each removal is independent: a refused output path or one foreign ID + # cannot disable cleanup of the other captured owned objects. + for role, identities in self.containers.items(): + for identity in sorted(identities): + self.cleanup_one(outcomes, identity, self.remove_container, role, identity) + for identity in sorted(self.networks): + self.cleanup_one(outcomes, identity, self.remove_network, identity) + self.cleanup_one(outcomes, "remaining-owned-labels", self.verify_label_absence) + self.report["cleanup"] = outcomes + self.report["complete"] = self.report["complete"] and all( + value.get("absent") and not value.get("oom_killed") for value in outcomes.values() + ) + + @staticmethod + def cleanup_one(outcomes, key, action, *arguments): + try: + outcomes[key] = action(*arguments) + except Exception as error: + outcomes[key] = {"error": type(error).__name__} + + def cleanup_network_boundaries(self, outcomes): + for identity in sorted(self.networks): + try: + record = json.loads(cleanup_call("docker", "network", "inspect", identity).stdout)[0] + runtime.verify_network(record, identity, self.nonce) + known = set().union(*self.containers.values()) + if set(record["Containers"] or {}) - known: + raise ValueError("foreign network attachment before cleanup") + except Exception as error: + outcomes["network-boundary-" + identity] = {"error": type(error).__name__} + + def verify_label_absence(self): + filters = [ + "--filter", + LABEL_FILTER + runtime.OWNER + OWNER_VALUE, + "--filter", + LABEL_FILTER + runtime.RUN + "=" + self.nonce, + ] + containers = cleanup_call("docker", "ps", "-a", "--no-trunc", *filters, "--format", ID_FORMAT) + networks = cleanup_call("docker", "network", "ls", "--no-trunc", *filters, "--format", ID_FORMAT) + return {"absent": not containers.stdout.strip() and not networks.stdout.strip()} + + +def run_case(owned, selected, images, packet, bootstrap, report): + owned.create_network() + runtime_id = owned.create("runtime", images["runtime"], selected["server_amd64_manifest"], bootstrap) + call("docker", "start", runtime_id) + consumer_id = owned.create("consumer", images["consumer"], selected["python_amd64_manifest"], bootstrap) + session = Session(consumer_id, DOCKER_ENVIRONMENT) + try: + session.send(packet) + ready = session.receive("ACTIVITY_CONSUMER_READY=") + verify_distributions(ready, base64.b64decode(packet["files"][REQUIREMENTS]["base64"])) + url = owned.boundary(images, selected, bootstrap, "before-sdk") + session.send({"runtime_ipv4": int(ipaddress.IPv4Address(url)), "token": runtime.ENVIRONMENT["DW_AUTH_TOKEN"]}) + result = session.receive("ACTIVITY_CONSUMER_RESULT=") + verify_case_result(result) + if owned.boundary(images, selected, bootstrap, "after-sdk") != url: + raise ValueError("runtime endpoint changed during the case") + if ready["sdk_inventory"] != result["sdk_inventory"]: + raise ValueError("installed SDK bytes changed during the case") + owned.files.write("sdk-inventory.json", json.dumps(result["sdk_inventory"], indent=2)) + owned.files.write("identity.log", result["case_output"]) + owned.files.write("identity.xml", result["junit"]) + report["consumer_ready"] = {key: value for key, value in ready.items() if key != "sdk_inventory"} + session.send({"release": True}) + code = session.finish() + report["consumer_exit"] = code + report["complete"] = bool(result["complete"] and code == 0) + finally: + session.close() + owned.files.write("consumer-transport.log", bytes(session.output)) + + +def main(): + files = output_files.Output(OUTPUT) + report = {"complete": False, "started_at": datetime.now(timezone.utc).isoformat()} + owned = OwnedDocker(uuid4().hex, files, report) + signal.signal(signal.SIGALRM, deadline) + signal.alarm(360) + try: + captured = capture_source(report) + selected = json.loads(captured["tuple.json"]) + report["tuple"] = selected + report["engine"] = engine() + packet = download_wheels(selected, captured, files) + report["wheel_sha256"] = packet["wheel_sha256"] + report["wheel_set"] = {name: item["sha256"] for name, item in packet["wheels"].items()} + images = {role: verify_published_image(selected, role, files) for role in ("runtime", "consumer")} + bootstrap = bind_bootstrap(captured) + report["consumer_bootstrap_sha256"] = hashlib.sha256(bootstrap.encode()).hexdigest() + run_case(owned, selected, images, packet, bootstrap, report) + if engine() != report["engine"]: + raise ValueError("Docker engine authority changed") + except Exception as error: + report["failure"] = type(error).__name__ + ": " + str(error)[:1000] + report["complete"] = False + finally: + signal.alarm(0) + owned.cleanup() + report["finished_at"] = datetime.now(timezone.utc).isoformat() + files.finish(report) + files.close() + if not report["complete"]: + raise RuntimeError("published activity acceptance or cleanup is incomplete; see result.json") + + +if __name__ == "__main__": + main() diff --git a/scripts/ci/published-activity-requirements.in b/scripts/ci/published-activity-requirements.in new file mode 100644 index 0000000..022eb2a --- /dev/null +++ b/scripts/ci/published-activity-requirements.in @@ -0,0 +1,3 @@ +durable-workflow==2.17.2 +pytest==9.2.0 +pytest-asyncio==1.5.0 diff --git a/scripts/ci/published-activity-requirements.txt b/scripts/ci/published-activity-requirements.txt new file mode 100644 index 0000000..12e0481 --- /dev/null +++ b/scripts/ci/published-activity-requirements.txt @@ -0,0 +1,83 @@ +anyio==4.16.0 \ + --hash=sha256:59bb03251907983d1603d27c3087ddb1e93299ea2163adcdba37899a4fb12ffb \ + --hash=sha256:c66ae634e0b8228396b801741beb3294326d4b6102f7bba4041c0dfb58c29026 +certifi==2026.10.23 \ + --hash=sha256:a9895e6e2758b371e0545e321a3a8fd1fcfd9829ada8824a936dfb16aaaf36cd \ + --hash=sha256:e9b54de76e530bb6cf0ccd036787103acb58fd42ba930ea8bf948ddbaa2f61d1 +durable-workflow==2.17.2 \ + --hash=sha256:76feb6ce60eb92710c718ece3fe63bb9bfb3b1cafc7fa0dbdce0a9e0b8321956 \ + --hash=sha256:7dc77d3882fa5cbddbeaee6a4cc63b05285e280523c8ea40dc31e03cc12a40bc +fastavro==1.13.1 \ + --hash=sha256:01810229c86dcec75da8cc08f18f509e7a1883681c5c83c69f85589998440624 \ + --hash=sha256:045af8ab8fec214e3ff6241fed32c5124582888d5dce1da3ef3fa48629bd25b2 \ + --hash=sha256:0723398cd2b246a47bb6f44cb8230f158391c59e998f79687ba256cfa37127d7 \ + --hash=sha256:0994c545a4e2038b6d0b3ca54214d9573024e659fc5e618c4577329c89b9e016 \ + --hash=sha256:142e97f126358d910fc1d54742f8129f7c8ddee5d6c6c2da4ac8440483d03964 \ + --hash=sha256:28305b4e0764f362cffe5bb6993021d584c050d49256f153d1f46ee4fb188ba8 \ + --hash=sha256:2c44e98f32f59478ff0636b0415859327775a62433c2a184541595fb806ef33c \ + --hash=sha256:2f56a127d71e45083306d2650efff827cad0f4b0744dd42cb69c631d77943b1d \ + --hash=sha256:300a3c13dfa4ae7940224021dd5d41ea9fbad0a7bfa446e3f4176a969d18e596 \ + --hash=sha256:3fbe18a47dc1ea35bcdf01c16b7c9fe0dbeb22aa0e57e75d8c4dcd7b57395ea6 \ + --hash=sha256:3fd052bf63c097a34da732eba9f4eea179ae1104664e58c2404b48768b3d550f \ + --hash=sha256:46ff9c48be24798e1926eaa3733f80967439cd7f1c7514e32c64714cb6c405d9 \ + --hash=sha256:47ddd4d831eced3765b0f98d597bea8e07973b62be5aefce75ff7fc12fdb0f9e \ + --hash=sha256:5678573fd7a01d7b91099e9aa5ceb4a12f94979b421a710ae079c07c6470c864 \ + --hash=sha256:59a3ade141eb59cf723bede90a7cce0b1f9d49c642fe19d34737b421ac385495 \ + --hash=sha256:6bc39e1b87893307df49c6117cb2525e216af02da6b292d78685396366a41205 \ + --hash=sha256:6f05aa2539bf7a19e9eb3bdaf6580c4d0f082a8230f641eaf9c84e4bcf0e6bc4 \ + --hash=sha256:73fc8234e0dd162b69374bb66bbfb37dd6eac48d4e43c4c8609d2ffafb92797f \ + --hash=sha256:754a483d1f161545da76b3d6a3155b7e37477f1e149f00ccfff740d9ec5c143e \ + --hash=sha256:78251e44f96079b1d884b1977eeadee5a18b32098a42aa950a6914e5b6ec6e16 \ + --hash=sha256:783d3fa1a0b1cf785893788b276e674f69824d104498f7aee2d80f5fb73f619e \ + --hash=sha256:7db91731ae8f77e638525245a5b74c673c6ef1b1d3b1e64b91a5232cb4e34f6e \ + --hash=sha256:8ceecd6896adbc57c9e59ee3295c8016ae372f17df9787c4d1ba5a73209d723a \ + --hash=sha256:8f12f7f8154fbae11bad499ad93fbff08764c390acd43461ca4f7dc7807925b8 \ + --hash=sha256:90049246bc000da01715194e038da1121a24288c702a8482cc660069a41aacba \ + --hash=sha256:950f2e260f65c7e6135288c142b078d06d2f1c90fc52f91a14c08e5f8811bf06 \ + --hash=sha256:9be0b06f90784f5e04bfb29a467c698ab1f88409c0db4821bbc4d86d583bc82a \ + --hash=sha256:9f53c6e3179ef6c35724e5193c69bda85d001d987bbfb487a171fa04f526bd7c \ + --hash=sha256:a06d21d9ef55a9ab56eb869713ee88371b05da9fd9600a44170649eab71c6310 \ + --hash=sha256:a1b96aceb181a699dcadd1b0dad7026047ee62f606d1df36ca5a52acd4fe9dc3 \ + --hash=sha256:aef0ba9b7b9c0b6febeb4c14da9f13957dc02bc522ca4ab01d226c4d0dcde08a \ + --hash=sha256:bf36a4391f62b3c8292ff8461def7192738eb9311edd26c6d730788e92ee2560 \ + --hash=sha256:d596200f71c5706e931708ab4cb6f39decbdebe660453c54707a36e7a66b4aba \ + --hash=sha256:db65955d681266091392756ea80728b7f002e038b0c45f88873897b95c7963a0 \ + --hash=sha256:deab9d233ca9e3b03021c5b87a7807a1986a0375ef64975cbee9ad104e7eb3ea \ + --hash=sha256:e3d7e0850230a9af977184dd0677e2bc6341659835d55a73a2fa76c7d2d2d65e \ + --hash=sha256:f4126ba2e1097e42e5f911f16efca9df62ec54d40c27e18ff304c017c32a8af9 \ + --hash=sha256:f59980a60ecc1bce5a9a0f95116bd05928936514f199e127770b7afc7d423842 \ + --hash=sha256:ffa147df1278b8a849586da1f2b520e856e78ea797edc4c974c8bb1e6b4bfd66 \ + --hash=sha256:ffa4b0b942e3aa7e66cc97a1862a2da6a3fce3dbcbd17a9b4be6ff1c33c93976 +h11==0.16.0 \ + --hash=sha256:4e35b956cf45792e4caa5885e69fba00bdbc6ffafbfa020300e549b208ee5ff1 \ + --hash=sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86 +httpcore==1.0.9 \ + --hash=sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55 \ + --hash=sha256:6e34463af53fd2ab5d807f399a9b45ea31c3dfa2276f15a2c3f00afff6e176e8 +httpx==0.28.1 \ + --hash=sha256:75e98c5f16b0f35b567856f597f06ff2270a374470a5c2392242528e3e3e42fc \ + --hash=sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad +idna==3.20 \ + --hash=sha256:a7db850025b95ded1eae8a46181a1a6c56c92c96f0e2b005d9ff8dc0210cab44 \ + --hash=sha256:ab7ae7122974553370f0bdb919e1a960b2cd1bc1ef0276416d896db81c14582c +iniconfig==2.3.1 \ + --hash=sha256:67f4b9c50da0dedf52af349e7749a80a9057a5031199791b906c3bb3ae878960 \ + --hash=sha256:9121e2c1fdb355232495be3194c8dfe87ccc2d5dee45947b78e68f499790d7a7 +packaging==26.3 \ + --hash=sha256:94edc256424af38762eb31306eed28beb9f0efc50a8837492c9d6fd6004aed79 \ + --hash=sha256:d7193f7c8e4e93f444fde0262bf90af30e16fa0ad0ad44cb553c87339b23cd1c +pluggy==1.7.0 \ + --hash=sha256:7dd7b0d8832ba3cb632c306926ded123429211b83641b35dc5c41ad2d34f9bec \ + --hash=sha256:d1eaa46ebb595891b860ab086b4d09c8588af65ebd4361b8e8f4bb8920b90ba8 +pygments==2.21.0 \ + --hash=sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9 \ + --hash=sha256:610ca751c9bc2492b38eb9a38a7fbc93edbbb2d7182edaf34e66ae493dee5c8c +pytest==9.2.0 \ + --hash=sha256:07acdcf88ec174f907dc01af70bf52d49b014bbab03faeeade31c96b18c2e588 \ + --hash=sha256:97c4f3b1f3c7414aa36465e09b8220f1d03627993bd51b1282e193753c80557b +pytest-asyncio==1.5.0 \ + --hash=sha256:557f255ee03cc84f84a8d5f438bc9c6e0f714bf86970ae743c7ecb5a38d7e07d \ + --hash=sha256:92c8fa83ca419fd0d4bd6a34df6520dae435841d3733c319343249492adf37f7 +typing-extensions==4.16.0 \ + --hash=sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8 \ + --hash=sha256:dc983d19a509c94dba722ee6abd33940f7c05a89e243c47e907eb4db6f1a43e5 diff --git a/scripts/ci/verify_activity_install.py b/scripts/ci/verify_activity_install.py new file mode 100644 index 0000000..12f2836 --- /dev/null +++ b/scripts/ci/verify_activity_install.py @@ -0,0 +1,124 @@ +"""Verify installed SDK files against the independently approved registry wheel.""" + +from __future__ import annotations + +import base64 +import configparser +import csv +import hashlib +import io +import os +import stat +import zipfile +from pathlib import Path, PurePosixPath + + +def sha256(data): + return hashlib.sha256(data).hexdigest() + + +def record_rows(text): + result = {} + for path, checksum, size in csv.reader(io.StringIO(text)): + if path in result: + raise ValueError("SDK RECORD repeats a member") + result[path] = (checksum, size) + return result + + +def member(archive, path, checksum, size): + parts = PurePosixPath(path) + if parts.is_absolute() or ".." in parts.parts or "\\" in path: + raise ValueError("SDK wheel member path is not relative") + info = archive.getinfo(path) + if stat.S_ISLNK(info.external_attr >> 16): + raise ValueError("SDK wheel member is a symlink") + content = archive.read(path) + digest = base64.urlsafe_b64encode(hashlib.sha256(content).digest()).decode().rstrip("=") + if checksum != "sha256=" + digest or int(size) != len(content): + raise ValueError("SDK wheel member differs from RECORD") + return content + + +def wheel_members(data, expected_sha256): + if sha256(data) != expected_sha256: + raise ValueError("SDK wheel differs from its reviewed digest") + with zipfile.ZipFile(io.BytesIO(data)) as archive: + names = archive.namelist() + records = [name for name in names if name.endswith(".dist-info/RECORD")] + if len(names) != len(set(names)) or len(records) != 1: + raise ValueError("SDK wheel membership differs") + record = records[0] + rows = record_rows(archive.read(record).decode()) + if set(names) != set(rows) or rows[record] != ("", ""): + raise ValueError("SDK wheel members differ from RECORD membership") + expected = {path: member(archive, path, *row) for path, row in rows.items() if path != record} + return record, expected + + +def installed_member(site, path, content): + installed = site / path + if installed.is_symlink() or not installed.is_file() or installed.read_bytes() != content: + raise ValueError("installed SDK member differs from reviewed wheel") + return {"sha256": sha256(content), "bytes": len(content)} + + +def generated_members(site, record, members, python, script_template): + metadata = str(PurePosixPath(record).parent) + generated = {metadata + "/INSTALLER": b"pip\n", metadata + "/REQUESTED": b""} + entry_points = members.get(metadata + "/entry_points.txt") + if entry_points is None: + return generated + if python is None or script_template is None: + raise ValueError("console scripts require the trusted base installer's template") + parser = configparser.ConfigParser() + parser.read_string(entry_points.decode()) + if parser.sections() != ["console_scripts"]: + raise ValueError("SDK entry point groups differ") + for name, target in parser.items("console_scripts"): + if not name or "/" in name or "\\" in name or name in {".", ".."}: + raise ValueError("SDK console script name differs") + module, function = target.split(":", 1) + body = script_template % {"module": module, "import_name": function.split(".")[0], "func": function} + relative = os.path.relpath(Path(python).parent / name, site) + generated[relative] = ("#!" + str(python) + "\n" + body).encode() + return generated + + +def verify_membership(site, record, expected): + roots = (site / "durable_workflow", site / PurePosixPath(record).parent) + actual = { + str(path.relative_to(site)) for root in roots for path in root.rglob("*") if path.is_file() or path.is_symlink() + } + distribution = {path for path in expected if not path.startswith("../")} | {record} + if actual != distribution: + raise ValueError("installed SDK has missing or extra distribution files") + + +def verify_installed_record(site, record, expected): + installed_record = site / record + if installed_record.is_symlink() or not installed_record.is_file(): + raise ValueError("installed SDK RECORD is unavailable") + installed_rows = record_rows(installed_record.read_text()) + if set(installed_rows) != set(expected) | {record} or installed_rows[record] != ("", ""): + raise ValueError("installed SDK RECORD membership differs") + for path, data in expected.items(): + digest = base64.urlsafe_b64encode(bytes.fromhex(data["sha256"])).decode().rstrip("=") + if installed_rows[path] != ("sha256=" + digest, str(data["bytes"])): + raise ValueError("installed SDK RECORD differs from the reviewed distribution") + expected[record] = {"sha256": sha256(installed_record.read_bytes()), "bytes": installed_record.stat().st_size} + return expected + + +def verify_sdk(site, origin, wheel, expected_sha256, *, python=None, script_template=None): + site = Path(site).resolve() + if Path(origin).resolve() != site / "durable_workflow/__init__.py": + raise ValueError("SDK import origin differs from the fresh environment") + record, members = wheel_members(wheel, expected_sha256) + generated = generated_members(site, record, members, python, script_template) + if set(generated) & set(members): + raise ValueError("installer entries overlap reviewed wheel members") + members.update(generated) + verify_membership(site, record, members) + expected = {path: installed_member(site, path, content) for path, content in members.items()} + return verify_installed_record(site, record, expected) diff --git a/tests/fixtures/published-activity-acceptance.json b/tests/fixtures/published-activity-acceptance.json new file mode 100644 index 0000000..47c7440 --- /dev/null +++ b/tests/fixtures/published-activity-acceptance.json @@ -0,0 +1,11 @@ +{ + "schema": "sdk-python.published-activity-acceptance/v1", + "sdk_version": "2.17.2", + "server_image": "ghcr.io/durable-workflow/server@sha256:8ffcdb2a36af3129a1987849aafe8715fe0d3bf9a9648467f0b5cb8eeef43ba9", + "server_source": "5614274d2629d03f92932a07dfee0fb10349ddd9", + "server_amd64_manifest": "sha256:813c9337c44be7d497efd29e1b463546519e488aa796b5ecf9dfc6f4f10d2128", + "server_qualification": "https://github.com/durable-workflow/server/pull/1209#issuecomment-6858753154", + "python_image": "docker.io/library/python@sha256:935c3c5b12357a7e10caa9c1e6ca5692344d97cf229832ef584be2cec9fc03df", + "python_amd64_manifest": "sha256:0f086103b026048cc50cb5da25ec435c023c5241a83fc44daea593377abc61c4", + "python_version": "3.12.16" +} diff --git a/tests/integration/test_activity_identity.py b/tests/integration/test_activity_identity.py new file mode 100644 index 0000000..ec984aa --- /dev/null +++ b/tests/integration/test_activity_identity.py @@ -0,0 +1,92 @@ +"""Observe logical identity inside real activity callbacks across Server retry.""" + +from __future__ import annotations + +import importlib.metadata +import json +import os +import sys +from pathlib import Path +from typing import Any +from uuid import uuid4 + +import durable_workflow +from durable_workflow import ActivityRetryPolicy, Client, Worker, activity, workflow +from durable_workflow.workflow import WorkflowContext + + +@workflow.defn(name="tests.python-activity-retry-identity") +class ActivityRetryIdentityWorkflow: + def run(self, ctx: WorkflowContext) -> Any: + return ( + yield ctx.schedule_activity( + "tests.python-retry-identity", + [], + retry_policy=ActivityRetryPolicy(max_attempts=2, backoff_seconds=[1]), + ) + ) + + +async def test_retry_preserves_received_execution_id_and_changes_attempt_id( + server_url: str, + server_token: str, +) -> None: + installed_version = os.environ.get("DURABLE_WORKFLOW_EXPECT_INSTALLED_VERSION") + if installed_version: + assert importlib.metadata.version("durable-workflow") == installed_version + assert Path(durable_workflow.__file__).is_relative_to(Path(sys.prefix)) + observed: list[dict[str, Any]] = [] + + @activity.defn(name="tests.python-retry-identity") + async def inspect_identity() -> dict[str, Any]: + info = activity.context().info + value = { + "execution_id": info.activity_execution_id, + "workflow_id": info.workflow_id, + "run_id": info.run_id, + "task_id": info.task_id, + "attempt_id": info.activity_attempt_id, + "attempt_number": info.attempt_number, + } + observed.append(value) + if info.attempt_number == 1: + raise RuntimeError("retry the first identity observation") + return value + + queue = "python-activity-identity-" + uuid4().hex + async with Client(server_url, token=server_token, namespace="default") as client: + worker = Worker( + client, + task_queue=queue, + workflows=[ActivityRetryIdentityWorkflow], + activities=[inspect_identity], + ) + handle = await client.start_workflow( + workflow_type="tests.python-activity-retry-identity", + workflow_id=queue, + task_queue=queue, + input=[], + ) + try: + await worker.run_until(workflow_id=handle.workflow_id, timeout=45) + result = await handle.result(timeout=10) + finally: + await worker.stop() + + assert len(observed) == 2 + first, second = observed + assert [first["attempt_number"], second["attempt_number"]] == [1, 2] + assert isinstance(first["execution_id"], str) + assert first["execution_id"].strip() + assert first["execution_id"] == second["execution_id"] + for value in observed: + for name in ("task_id", "attempt_id"): + assert isinstance(value[name], str) + assert value[name].strip() + assert first["attempt_id"] != second["attempt_id"] + assert first["task_id"] != second["task_id"] + assert {value["workflow_id"] for value in observed} == {handle.workflow_id} + assert {value["run_id"] for value in observed} == {handle.run_id} + assert first["execution_id"] not in {first["attempt_id"], second["attempt_id"], first["task_id"], second["task_id"]} + assert result == second + print(json.dumps({"scope": "activity callback retry identity", "observed": observed}, sort_keys=True)) diff --git a/tests/test_activity_acceptance_authority.py b/tests/test_activity_acceptance_authority.py new file mode 100644 index 0000000..f910618 --- /dev/null +++ b/tests/test_activity_acceptance_authority.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +import importlib.util +import json +import os +from pathlib import Path + +import pytest + +ROOT = Path(__file__).parents[1] / "scripts/ci" + + +def load(name): + spec = importlib.util.spec_from_file_location(name, ROOT / (name + ".py")) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +runtime = load("activity_acceptance_runtime") +files = load("activity_acceptance_output") + + +def test_malformed_create_stdout_does_not_discard_the_full_owned_cid(): + identity, agrees, candidates = runtime.created_identity("malformed stdout", "a" * 64 + "\n") + assert identity == "a" * 64 + assert agrees is False + assert candidates == ["a" * 64] + + +def test_disagreeing_creation_channels_preserve_both_candidates_without_admitting_start(): + identity, agrees, candidates = runtime.created_identity("a" * 64, "b" * 64) + assert identity == "b" * 64 + assert agrees is False + assert candidates == ["a" * 64, "b" * 64] + + +def test_private_output_refuses_a_preexisting_result_symlink(tmp_path): + output = files.Output(tmp_path / "result") + target = tmp_path / "foreign" + target.write_text("preserved") + (output.path / "link.json").symlink_to(target) + try: + with pytest.raises(FileExistsError): + output.write("link.json", "replacement") + assert target.read_text() == "preserved" + finally: + output.close() + + +def test_terminal_receipt_uses_the_reserved_descriptor_after_directory_replacement(tmp_path): + output = files.Output(tmp_path / "result") + original = tmp_path / "original" + output.path.rename(original) + output.path.mkdir() + try: + report = {"complete": True} + output.finish(report) + assert report["complete"] is False + assert json.loads((original / "result.json").read_text())["output_failure"] == "identity_changed" + assert not (output.path / "result.json").exists() + finally: + output.close() + + +def test_replaced_result_cannot_become_a_successful_terminal_receipt(tmp_path): + output = files.Output(tmp_path / "result") + (output.path / "result.json").rename(output.path / "retained.json") + (output.path / "result.json").write_text("foreign") + try: + report = {"complete": True} + output.finish(report) + assert report["complete"] is False + original = os.pread(output.files["result.json"], 4096, 0) + assert json.loads(original)["complete"] is False + assert (output.path / "result.json").read_text() == "foreign" + finally: + output.close() diff --git a/tests/test_activity_acceptance_consumer.py b/tests/test_activity_acceptance_consumer.py new file mode 100644 index 0000000..e9fcf66 --- /dev/null +++ b/tests/test_activity_acceptance_consumer.py @@ -0,0 +1,155 @@ +from __future__ import annotations + +import importlib.util +import ipaddress +import os +import subprocess +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +ROOT = Path(__file__).parents[1] / "scripts/ci" + + +def load(name): + spec = importlib.util.spec_from_file_location(name, ROOT / (name + ".py")) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +consumer = load("activity_acceptance_consumer") +transport = load("activity_acceptance_session") + + +@pytest.mark.parametrize( + "url", + [ + "http://172.29.0.2:8080", + "foreign.example", + None, + True, + -1, + 2**32, + int(ipaddress.IPv4Address("127.0.0.1")), + int(ipaddress.IPv4Address("8.8.8.8")), + int(ipaddress.IPv4Address("0.0.0.0")), + int(ipaddress.IPv4Address("224.0.0.1")), + int(ipaddress.IPv4Address("169.254.1.1")), + ], +) +def test_consumer_refuses_unselected_endpoint_before_any_request(url, monkeypatch): + monkeypatch.setattr(consumer.urllib.request, "build_opener", lambda *_args: pytest.fail("unexpected request")) + with pytest.raises(ValueError): + consumer.await_runtime(url) + + +def test_consumer_uses_empty_proxies_for_its_numeric_internal_readiness_request(monkeypatch): + seen = [] + + class Response: + status = 200 + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def handler(proxies): + seen.append(proxies) + return "empty proxy handler" + + def open_url(url, timeout): + seen.append((url, timeout)) + return Response() + + monkeypatch.setattr(consumer.urllib.request, "ProxyHandler", handler) + monkeypatch.setattr(consumer.urllib.request, "build_opener", lambda _handler: SimpleNamespace(open=open_url)) + consumer.await_runtime(int(ipaddress.IPv4Address("172.29.0.2"))) + assert seen == [{}, ("http://172.29.0.2:8080/api/health", 2)] + + +def test_installer_has_no_ambient_environment_or_network_fallback(monkeypatch): + calls = [] + monkeypatch.setenv("PYTHONPATH", "/foreign/src") + monkeypatch.setenv("HTTPS_PROXY", "http://foreign") + monkeypatch.setenv("PIP_EXTRA_INDEX_URL", "https://foreign") + monkeypatch.setattr(consumer, "run", lambda arguments, environment: calls.append((arguments, environment))) + environment, _python = consumer.prepare_environment() + assert "PYTHONPATH" not in environment + assert "HTTPS_PROXY" not in environment + assert "PIP_EXTRA_INDEX_URL" not in environment + command = calls[-1][0] + assert "--no-index" in command + assert "--require-hashes" in command + assert "--no-compile" in command + assert command[1:3] == ["-I", "-B"] + + +def test_stdio_protocol_stays_open_for_authority_recheck_and_explicit_release(monkeypatch): + original = subprocess.Popen + # Only inert JSON messages run here; there is no SDK, package or API input. + source = """ +import json,sys +assert json.loads(sys.stdin.readline()) == {'setup': 'captured'} +print('READY=' + json.dumps({'complete': True}), flush=True) +assert json.loads(sys.stdin.readline()) == {'admit': True} +print('RESULT=' + json.dumps({'complete': True}), flush=True) +assert json.loads(sys.stdin.readline()) == {'release': True} +""" + + def launch(_arguments, **kwargs): + return original([sys.executable, "-I", "-B", "-c", source], **kwargs) + + monkeypatch.setattr(transport.subprocess, "Popen", launch) + session = transport.Session("a" * 64, {"PATH": os.defpath}) + try: + session.send({"setup": "captured"}) + assert session.receive("READY=") == {"complete": True} + session.send({"admit": True}) + assert session.receive("RESULT=") == {"complete": True} + assert session.process.poll() is None + session.send({"release": True}) + assert session.finish() == 0 + finally: + session.close() + + +def test_input_digest_substitution_refuses_before_writing(tmp_path): + with pytest.raises(ValueError, match="digest differs"): + consumer.write(tmp_path, "input.py", {"base64": "cHJpbnQoJ3JlcGxhY2VtZW50Jyk=", "sha256": "0" * 64}) + assert list(tmp_path.iterdir()) == [] + + +def test_closed_consumer_refuses_sdk_wheel_omission_or_substitution(): + packet = { + "files": dict.fromkeys( + ("requirements.txt", "verify_activity_install.py", "test_activity_identity.py", "conftest.py"), {} + ), + "wheels": {"sdk.whl": {"sha256": "a" * 64}}, + "wheel_name": "sdk.whl", + "wheel_sha256": "b" * 64, + "sdk_version": "test", + } + with pytest.raises(ValueError, match="differs from approval"): + consumer.validate_setup(packet) + packet["wheel_name"] = "absent.whl" + with pytest.raises(ValueError, match="wheel is missing"): + consumer.validate_setup(packet) + + +def test_sdk_writable_verifier_replacement_cannot_change_the_retained_checker(monkeypatch, tmp_path): + original = "def verify_sdk(*args, **kwargs): return {'checker': 'Git-captured'}\n" + monkeypatch.setattr(consumer, "CAPTURED_VERIFIER", original) + checker = consumer.retained_verifier() + data = tmp_path / "verify_activity_install.py" + data.write_text("def verify_sdk(*args, **kwargs): return {'checker': 'substituted'}\n") + predecessor = {} + exec(compile(data.read_text(), str(data), "exec"), predecessor) + assert predecessor["verify_sdk"]() == {"checker": "substituted"} + assert checker() == {"checker": "Git-captured"} diff --git a/tests/test_activity_install_inventory.py b/tests/test_activity_install_inventory.py new file mode 100644 index 0000000..b27b7ad --- /dev/null +++ b/tests/test_activity_install_inventory.py @@ -0,0 +1,152 @@ +from __future__ import annotations + +import base64 +import csv +import hashlib +import importlib.util +import io +import zipfile +from pathlib import Path + +import pytest + +PATH = Path(__file__).parents[1] / "scripts/ci/verify_activity_install.py" +SPEC = importlib.util.spec_from_file_location("verify_activity_install", PATH) +assert SPEC is not None +assert SPEC.loader is not None +inventory = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(inventory) + + +def installed_fixture(tmp_path): + site = tmp_path / "site" + package = site / "durable_workflow" + package.mkdir(parents=True) + content = b"VALUE = 'reviewed wheel'\n" + (package / "__init__.py").write_bytes(content) + name = "durable_workflow/__init__.py" + record = "durable_workflow-1.0.dist-info/RECORD" + table = io.StringIO() + writer = csv.writer(table) + digest = base64.urlsafe_b64encode(hashlib.sha256(content).digest()).decode().rstrip("=") + writer.writerow([name, "sha256=" + digest, len(content)]) + writer.writerow([record, "", ""]) + (site / record).parent.mkdir() + (site / record).write_text(table.getvalue()) + installed_table = io.StringIO(table.getvalue()) + installed_table.seek(0, 2) + installed_writer = csv.writer(installed_table) + for extra, data in (("INSTALLER", b"pip\n"), ("REQUESTED", b"")): + path = str(Path(record).parent / extra) + (site / path).write_bytes(data) + checksum = base64.urlsafe_b64encode(hashlib.sha256(data).digest()).decode().rstrip("=") + installed_writer.writerow([path, "sha256=" + checksum, len(data)]) + (site / record).write_text(installed_table.getvalue()) + wheel = tmp_path / "sdk.whl" + with zipfile.ZipFile(wheel, "w") as archive: + archive.writestr(name, content) + archive.writestr(record, table.getvalue()) + return site, package / "__init__.py", wheel, hashlib.sha256(wheel.read_bytes()).hexdigest() + + +def test_installed_member_matches_the_independently_reviewed_wheel(tmp_path): + site, origin, wheel, digest = installed_fixture(tmp_path) + result = inventory.verify_sdk(site, origin, wheel.read_bytes(), digest) + assert "durable_workflow/__init__.py" in result + assert "durable_workflow-1.0.dist-info/RECORD" in result + + +@pytest.mark.parametrize("change", ["replace", "extra", "wrong_origin", "wrong_wheel"]) +def test_version_and_origin_alone_cannot_approve_substituted_or_extra_sdk_bytes(tmp_path, change): + site, origin, wheel, digest = installed_fixture(tmp_path) + if change == "replace": + origin.write_text("VALUE = 'substituted module'\n") + elif change == "extra": + origin.with_name("extra.py").write_text("VALUE = 'ambient file'\n") + elif change == "wrong_origin": + origin = tmp_path / "checkout/src/durable_workflow/__init__.py" + else: + digest = "0" * 64 + data = wheel.read_bytes() + with pytest.raises(ValueError): + inventory.verify_sdk(site, origin, data, digest) + + +def test_repeated_installed_record_row_cannot_hide_a_replaced_member(tmp_path): + site, origin, wheel, digest = installed_fixture(tmp_path) + record = next(site.glob("*.dist-info/RECORD")) + record.write_text(record.read_text() + record.read_text().splitlines()[0] + "\n") + data = wheel.read_bytes() + with pytest.raises(ValueError, match="RECORD repeats"): + inventory.verify_sdk(site, origin, data, digest) + + +@pytest.mark.parametrize("change", ["extra_metadata", "foreign_entry_points", "extra_record_row", "installer_bytes"]) +def test_unreviewed_distribution_members_and_record_rows_are_refused(tmp_path, change): + site, origin, wheel, digest = installed_fixture(tmp_path) + metadata = next(site.glob("*.dist-info")) + if change == "extra_metadata": + (metadata / "ambient.py").write_text("unreviewed = True\n") + elif change == "foreign_entry_points": + (metadata / "entry_points.txt").write_text("[pytest11]\nambient = foreign:plugin\n") + elif change == "installer_bytes": + (metadata / "INSTALLER").write_text("foreign installer\n") + else: + record = metadata / "RECORD" + record.write_text(record.read_text() + "foreign.py,sha256=unreviewed,1\n") + data = wheel.read_bytes() + with pytest.raises(ValueError): + inventory.verify_sdk(site, origin, data, digest) + + +def test_only_expected_installer_entries_and_exact_generated_console_script_are_admitted(tmp_path): + site, origin, old_wheel, _old_digest = installed_fixture(tmp_path) + metadata = "durable_workflow-1.0.dist-info" + entry_points = (metadata + "/entry_points.txt", b"[console_scripts]\nreviewed-cli = durable_workflow:main\n") + python = tmp_path / "bin/python" + python.parent.mkdir() + template = "from %(module)s import %(import_name)s\n%(func)s()\n" + content = ("#!" + str(python) + "\nfrom durable_workflow import main\nmain()\n").encode() + script = python.parent / "reviewed-cli" + script.write_bytes(content) + (site / entry_points[0]).write_bytes(entry_points[1]) + record = metadata + "/RECORD" + with zipfile.ZipFile(old_wheel) as archive: + members = {name: archive.read(name) for name in archive.namelist() if name != record} + members[entry_points[0]] = entry_points[1] + table = io.StringIO() + writer = csv.writer(table) + for name, data in members.items(): + checksum = base64.urlsafe_b64encode(hashlib.sha256(data).digest()).decode().rstrip("=") + writer.writerow([name, "sha256=" + checksum, len(data)]) + writer.writerow([record, "", ""]) + wheel = tmp_path / "console.whl" + with zipfile.ZipFile(wheel, "w") as archive: + for name, data in members.items(): + archive.writestr(name, data) + archive.writestr(record, table.getvalue()) + for name, data in ( + (metadata + "/INSTALLER", b"pip\n"), + (metadata + "/REQUESTED", b""), + ("../bin/reviewed-cli", content), + ): + checksum = base64.urlsafe_b64encode(hashlib.sha256(data).digest()).decode().rstrip("=") + writer.writerow([name, "sha256=" + checksum, len(data)]) + (site / record).write_text(table.getvalue()) + data = wheel.read_bytes() + digest = hashlib.sha256(data).hexdigest() + result = inventory.verify_sdk(site, origin, data, digest, python=python, script_template=template) + assert "../bin/reviewed-cli" in result + # Matching attacker-updated RECORD rows cannot approve changed wrappers. + changed = b"print('unreviewed console code')\n" + script.write_bytes(changed) + checksum = base64.urlsafe_b64encode(hashlib.sha256(changed).digest()).decode().rstrip("=") + rows = list(csv.reader((site / record).read_text().splitlines())) + for row in rows: + if row[0] == "../bin/reviewed-cli": + row[1:] = ["sha256=" + checksum, str(len(changed))] + table = io.StringIO() + csv.writer(table).writerows(rows) + (site / record).write_text(table.getvalue()) + with pytest.raises(ValueError, match="member differs"): + inventory.verify_sdk(site, origin, data, digest, python=python, script_template=template) diff --git a/tests/test_published_activity_acceptance.py b/tests/test_published_activity_acceptance.py new file mode 100644 index 0000000..8407b9f --- /dev/null +++ b/tests/test_published_activity_acceptance.py @@ -0,0 +1,543 @@ +from __future__ import annotations + +import importlib.util +import json +from copy import deepcopy +from pathlib import Path +from types import SimpleNamespace + +import pytest + +PATH = Path(__file__).parents[1] / "scripts/ci/published-activity-acceptance.py" +SPEC = importlib.util.spec_from_file_location("published_activity_acceptance", PATH) +assert SPEC is not None +assert SPEC.loader is not None +runner = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(runner) +runtime = runner.runtime + + +def configuration(role="runtime"): + image = { + "Id": "sha256:" + "f" * 64, + "Config": { + "Env": ["PATH=/usr/local/bin:/usr/bin:/bin"], + "Entrypoint": ["/app/server"], + "Cmd": None, + "WorkingDir": "/app", + }, + } + entrypoint, command, directory = runtime.expected_command(image, role, "captured source") + environment = list(image["Config"]["Env"]) + if role == "runtime": + environment += [key + "=" + value for key, value in runtime.ENVIRONMENT.items()] + host = { + "NetworkMode": "c" * 64, + "ReadonlyRootfs": True, + "Privileged": False, + "NanoCpus": 1_000_000_000, + "Memory": 1_073_741_824, + "MemorySwap": 1_073_741_824, + "PidsLimit": 256, + "Init": True, + "CapDrop": ["ALL"], + "SecurityOpt": ["no-new-privileges"], + "Tmpfs": deepcopy(runtime.TMPFS[role]), + "PublishAllPorts": False, + "AutoRemove": False, + "OomKillDisable": False, + "ContainerIDFile": "", + "ConsoleSize": [0, 0], + "MaskedPaths": sorted(runtime.expected_masked_paths()), + "ReadonlyPaths": sorted(runtime.READONLY_PATHS), + "MemorySwappiness": None, + "ShmSize": 67_108_864, + "Runtime": "runc", + "IpcMode": "private", + "CgroupnsMode": "private", + "RestartPolicy": {"Name": "no", "MaximumRetryCount": 0}, + "LogConfig": {"Type": "json-file", "Config": {"max-size": "1m", "max-file": "1"}}, + } + record = { + "Id": "a" * 64, + "Image": image["Id"], + "Config": { + "Labels": {runtime.OWNER: "sdk-python", runtime.RUN: "ours", runtime.ROLE: role}, + "Env": environment, + "Image": image["Id"], + "User": "1000:1000", + "Entrypoint": entrypoint, + "Cmd": command, + "WorkingDir": directory, + "Tty": False, + "OpenStdin": role == "consumer", + "StdinOnce": role == "consumer", + "AttachStdin": role == "consumer", + "AttachStdout": True, + "AttachStderr": True, + "Hostname": "a" * 12, + "Domainname": "", + "Volumes": None, + }, + "HostConfig": host, + "Mounts": [], + "ImageManifestDescriptor": {"digest": "sha256:" + "d" * 64}, + } + return record, image + + +@pytest.mark.parametrize("role", ["runtime", "consumer"]) +def test_exact_closed_container_configuration_is_admitted(role): + record, image = configuration(role) + runtime.verify_container(record, "a" * 64, "ours", image, "c" * 64, "sha256:" + "d" * 64, role, "captured source") + arguments = runtime.create_arguments(image, "c" * 64, "ours", "captured.cid", role, "captured source") + assert "--pull=never" in arguments + assert "-p" not in arguments + assert "--publish" not in arguments + + +@pytest.mark.parametrize("role", ["runtime", "consumer"]) +@pytest.mark.parametrize( + "change", + [ + "image", + "child", + "user", + "command", + "owner", + "nonce", + "role", + "proxy", + "writable", + "swap", + "caps", + "network", + "host_mount", + "port", + "ipc", + "extra_env", + ], +) +def test_mutated_authority_or_container_profile_refuses_before_start(role, change): + record, image = configuration(role) + config, host = record["Config"], record["HostConfig"] + if change == "image": + record["Image"] = "sha256:" + "e" * 64 + elif change == "child": + record["ImageManifestDescriptor"]["digest"] = "sha256:" + "e" * 64 + elif change == "user": + config["User"] = "0:0" + elif change == "command": + config["Entrypoint"] = ["/bin/sh"] + elif change in {"owner", "nonce", "role"}: + config["Labels"][{"owner": runtime.OWNER, "nonce": runtime.RUN, "role": runtime.ROLE}[change]] = "foreign" + elif change in {"proxy", "extra_env"}: + config["Env"].append("HTTPS_PROXY=http://foreign" if change == "proxy" else "PYTHONPATH=/checkout/src") + elif change == "writable": + host["ReadonlyRootfs"] = False + elif change == "swap": + host["MemorySwap"] = -1 + elif change == "caps": + host["CapAdd"] = ["SYS_ADMIN"] + elif change == "network": + host["NetworkMode"] = "host" + elif change == "host_mount": + record["Mounts"] = [{"Type": "bind", "Destination": "/checkout"}] + elif change == "port": + host["PortBindings"] = {"8080/tcp": [{"HostIp": "127.0.0.1", "HostPort": "1234"}]} + else: + host["IpcMode"] = "host" + with pytest.raises(ValueError): + runtime.verify_container( + record, "a" * 64, "ours", image, "c" * 64, "sha256:" + "d" * 64, role, "captured source" + ) + + +def network_fixture(): + network = { + "Id": "c" * 64, + "Driver": "bridge", + "Internal": True, + "Attachable": False, + "Ingress": False, + "EnableIPv6": False, + "Labels": {runtime.OWNER: "sdk-python", runtime.RUN: "ours", runtime.ROLE: "network"}, + "IPAM": {"Config": [{"Subnet": "172.29.0.0/16"}]}, + "Containers": {}, + } + records = [] + for identity, address in (("a" * 64, "172.29.0.2"), ("b" * 64, "172.29.0.3")): + records.append( + { + "Id": identity, + "NetworkSettings": { + "Networks": { + "owned": {"NetworkID": network["Id"], "IPAddress": address}, + } + }, + } + ) + network["Containers"][identity] = {"IPv4Address": address + "/16"} + return network, records + + +def test_endpoint_is_bound_to_exact_two_ids_network_and_numeric_ip(): + network, records = network_fixture() + runtime.verify_network(network, "c" * 64, "ours") + assert runtime.verify_attachments(network, *records) == "172.29.0.2" + + +@pytest.mark.parametrize( + "change", + [ + "foreign", + "missing", + "extra_network", + "wrong_network", + "wrong_ip", + "loopback", + "hostname", + "public", + "wrong_subnet", + ], +) +def test_endpoint_substitution_or_foreign_attachment_is_refused(change): + network, records = network_fixture() + endpoint = records[0]["NetworkSettings"]["Networks"]["owned"] + if change == "foreign": + network["Containers"]["e" * 64] = {"IPv4Address": "172.29.0.4/16"} + elif change == "missing": + del network["Containers"]["a" * 64] + elif change == "extra_network": + records[0]["NetworkSettings"]["Networks"]["foreign"] = deepcopy(endpoint) + elif change == "wrong_network": + endpoint["NetworkID"] = "e" * 64 + elif change == "wrong_ip": + endpoint["IPAddress"] = "172.29.0.4" + elif change == "wrong_subnet": + network["IPAM"]["Config"][0]["Subnet"] = "172.30.0.0/16" + else: + endpoint["IPAddress"] = {"loopback": "127.0.0.1", "hostname": "foreign.example", "public": "8.8.8.8"}[change] + with pytest.raises(ValueError): + runtime.verify_attachments(network, *records) + + +@pytest.mark.parametrize( + "key,value", + [("Internal", False), ("Attachable", True), ("Ingress", True), ("EnableIPv6", True), ("Driver", "host")], +) +def test_network_refuses_a_less_restricted_profile(key, value): + network, _records = network_fixture() + network[key] = value + with pytest.raises(ValueError): + runtime.verify_network(network, "c" * 64, "ours") + + +def test_uncertain_create_is_reconciled_by_owner_role_nonce_and_full_id(monkeypatch, tmp_path): + files = runner.output_files.Output(tmp_path / "result") + owned = runner.OwnedDocker("ours", files, {"complete": False}) + seen = [] + + def call(*arguments): + seen.append(arguments) + role = arguments[-3] + identities = { + "label=" + runtime.ROLE + "=runtime": "a" * 64, + "label=" + runtime.ROLE + "=consumer": "b" * 64, + "label=" + runtime.ROLE + "=network": "c" * 64, + } + return SimpleNamespace(stdout=identities[role] + "\n") + + monkeypatch.setattr(runner, "cleanup_call", call) + try: + owned.discover() + assert owned.containers == {"runtime": {"a" * 64}, "consumer": {"b" * 64}} + assert owned.networks == {"c" * 64} + for arguments in seen: + assert "label=" + runtime.OWNER + "=sdk-python" in arguments + assert "label=" + runtime.RUN + "=ours" in arguments + finally: + files.close() + + +def test_cleanup_refuses_a_foreign_same_name_replacement_without_removing_it(monkeypatch, tmp_path): + files = runner.output_files.Output(tmp_path / "result") + owned = runner.OwnedDocker("ours", files, {"complete": False}) + record, _image = configuration() + record["Name"] = "/reused-display-name" + record["Config"]["Labels"][runtime.RUN] = "foreign" + calls = [] + + def call(*arguments, **_kwargs): + calls.append(arguments) + return SimpleNamespace(stdout=json.dumps([record]), returncode=0, stderr="", check_returncode=lambda: None) + + monkeypatch.setattr(runner, "cleanup_call", call) + try: + with pytest.raises(ValueError, match="cleanup ownership differs"): + owned.remove_container("runtime", "a" * 64) + assert calls == [("docker", "inspect", "a" * 64)] + finally: + files.close() + + +def test_partial_cleanup_failure_cannot_disable_other_owned_removals(monkeypatch, tmp_path): + files = runner.output_files.Output(tmp_path / "result") + report = {"complete": True} + owned = runner.OwnedDocker("ours", files, report) + owned.containers = {"runtime": {"a" * 64}, "consumer": {"b" * 64}} + owned.networks = {"c" * 64} + calls = [] + network, _records = network_fixture() + network["Containers"] = {} + monkeypatch.setattr(owned, "discover", lambda: None) + monkeypatch.setattr(runner, "cleanup_call", lambda *_args: SimpleNamespace(stdout=json.dumps([network]))) + + def remove(role, identity): + calls.append(identity) + if role == "runtime": + raise ValueError("reserved output path refused") + return {"absent": True} + + monkeypatch.setattr(owned, "remove_container", remove) + monkeypatch.setattr(owned, "remove_network", lambda identity: calls.append(identity) or {"absent": True}) + try: + owned.cleanup() + assert calls == ["a" * 64, "b" * 64, "c" * 64] + assert report["complete"] is False + finally: + files.close() + + +@pytest.mark.parametrize( + "actual", + [ + [["durable-workflow", "2.17.2"], ["pytest", "9.2.0"], ["ambient-plugin", "1.0"]], + [["durable-workflow", "2.17.2"], ["pytest", "9.1.0"]], + [["durable-workflow", "2.17.2"]], + ], +) +def test_complete_lock_rejects_ambient_missing_or_changed_distributions(actual): + lock = b"durable-workflow==2.17.2 \\\n --hash=sha256:abc\npytest==9.2.0 \\\n --hash=sha256:def\n" + ready = {"distributions": json.dumps(actual)} + with pytest.raises(ValueError, match="complete dependency lock"): + runner.verify_distributions(ready, lock) + + +def test_complete_lock_accepts_only_the_exact_private_environment(): + runner.verify_distributions( + {"distributions": json.dumps([["pytest", "9.2.0"], ["durable_workflow", "2.17.2"]])}, + b"durable-workflow==2.17.2\npytest==9.2.0\n", + ) + + +def test_malformed_create_retains_owned_cid_for_cleanup_and_terminal_failure(monkeypatch, tmp_path): + files = runner.output_files.Output(tmp_path / "result") + report = {"complete": False} + owned = runner.OwnedDocker("ours", files, report) + owned.network = "c" * 64 + monkeypatch.setattr(runner, "OUTPUT", files.path) + (files.path / "runtime.cid").write_text("a" * 64 + "\n") + result = SimpleNamespace(stdout="malformed create output", check_returncode=lambda: None) + monkeypatch.setattr(runner, "call", lambda *_args, **_kwargs: result) + monkeypatch.setattr(owned, "verify", lambda *_args: pytest.fail("uncertain create must not start")) + monkeypatch.setattr(owned, "discover", lambda: None) + monkeypatch.setattr(owned, "verify_label_absence", lambda: {"absent": True}) + removed = [] + monkeypatch.setattr( + owned, "remove_container", lambda role, identity: removed.append((role, identity)) or {"absent": True} + ) + try: + _record, image = configuration() + with pytest.raises(ValueError, match="create channels disagree"): + owned.create("runtime", image, "sha256:" + "d" * 64, "captured source") + owned.cleanup() + files.finish(report) + assert removed == [("runtime", "a" * 64)] + receipt = json.loads((files.path / "result.json").read_text()) + assert receipt["complete"] is False + assert receipt["cleanup"]["a" * 64]["absent"] is True + finally: + files.close() + + +def case_result(): + return { + "complete": True, + "junit": '' + "", + } + + +def test_complete_case_requires_the_actual_single_retry_test_to_pass(): + runner.verify_case_result(case_result()) + + +@pytest.mark.parametrize("change", ["skip", "failure", "error", "wrong_case", "empty", "reported_failure"]) +def test_exit_zero_cannot_qualify_a_skipped_missing_or_different_case(change): + result = case_result() + if change in {"skip", "failure", "error"}: + tag = "skipped" if change == "skip" else change + result["junit"] = result["junit"].replace("/>", "><" + tag + "/>") + elif change == "wrong_case": + result["junit"] = result["junit"].replace( + "test_retry_preserves_received_execution_id_and_changes_attempt_id", "unrelated_case" + ) + elif change == "empty": + result["junit"] = '' + else: + result["complete"] = False + with pytest.raises(ValueError): + runner.verify_case_result(result) + + +@pytest.mark.parametrize( + "field,value", + [ + ("GroupAdd", ["0"]), + ("OomKillDisable", True), + ("Sysctls", {"net.ipv4.ip_forward": "1"}), + ("DeviceCgroupRules", ["c *:* rwm"]), + ("CgroupParent", "foreign"), + ("Ulimits", [{"Name": "nofile", "Soft": 999999, "Hard": 999999}]), + ("Dns", ["8.8.8.8"]), + ("DnsSearch", ["foreign"]), + ("DnsOptions", ["use-vc"]), + ("CpuQuota", -1), + ("CpusetCpus", "0"), + ("MemoryReservation", 999999), + ("OomScoreAdj", -1000), + ("BlkioDeviceReadBps", [{"Path": "/dev/sda", "Rate": 1}]), + ], +) +def test_undeclared_group_oom_sysctl_device_cgroup_dns_and_resource_overrides_refuse(field, value): + record, image = configuration("consumer") + record["HostConfig"][field] = value + with pytest.raises(ValueError): + runtime.verify_container( + record, "a" * 64, "ours", image, "c" * 64, "sha256:" + "d" * 64, "consumer", "captured source" + ) + + +@pytest.mark.parametrize( + "field,value", + [ + ("Healthcheck", {"Test": ["CMD", "unreviewed"]}), + ("StopSignal", "SIGKILL"), + ("StopTimeout", 999), + ("Shell", ["/bin/sh", "-c"]), + ("NetworkDisabled", True), + ("MacAddress", "02:00:00:00:00:01"), + ("Volumes", {"/foreign": {}}), + ("ExposedPorts", {"9999/tcp": {}}), + ("OnBuild", ["RUN unreviewed"]), + ("AttachStdin", False), + ("StdinOnce", False), + ("Hostname", "foreign"), + ("UnreviewedRunHook", "foreign"), + ], +) +def test_unreviewed_effectful_config_fields_are_refused(field, value): + record, image = configuration("consumer") + record["Config"][field] = value + with pytest.raises(ValueError): + runtime.verify_container( + record, "a" * 64, "ours", image, "c" * 64, "sha256:" + "d" * 64, "consumer", "captured source" + ) + + +@pytest.mark.parametrize( + "field,value", + [ + ("MaskedPaths", []), + ("ReadonlyPaths", []), + ("IOMaximumIOps", 1), + ("IOMaximumBandwidth", 1), + ("UnreviewedHostFlag", True), + ("ContainerIDFile", "/foreign/container.cid"), + ], +) +def test_protected_paths_io_limits_and_unknown_host_fields_are_closed(field, value): + record, image = configuration("consumer") + record["HostConfig"][field] = value + with pytest.raises(ValueError): + runtime.verify_container( + record, "a" * 64, "ours", image, "c" * 64, "sha256:" + "d" * 64, "consumer", "captured source" + ) + + +@pytest.mark.parametrize("value", [False, None]) +def test_observed_docker29_false_null_oom_equivalence_keeps_the_killer_enabled(value): + record, image = configuration("consumer") + record["HostConfig"]["OomKillDisable"] = value + runtime.verify_container( + record, "a" * 64, "ours", image, "c" * 64, "sha256:" + "d" * 64, "consumer", "captured source" + ) + + +@pytest.mark.parametrize("value", [True, 0, 1, "false", "missing"]) +def test_oom_enabled_equivalence_does_not_admit_true_wrong_types_or_missing_evidence(value): + record, image = configuration("consumer") + if value == "missing": + del record["HostConfig"]["OomKillDisable"] + else: + record["HostConfig"]["OomKillDisable"] = value + with pytest.raises(ValueError): + runtime.verify_container( + record, "a" * 64, "ours", image, "c" * 64, "sha256:" + "d" * 64, "consumer", "captured source" + ) + + +def test_only_the_reviewed_image_healthcheck_is_admitted(): + record, image = configuration("runtime") + image["Config"]["Healthcheck"] = {"Test": ["CMD", "/app/reviewed-health"]} + record["Config"]["Healthcheck"] = deepcopy(image["Config"]["Healthcheck"]) + runtime.verify_container(record, "a" * 64, "ours", image, "c" * 64, "sha256:" + "d" * 64, "runtime") + del record["Config"]["Healthcheck"] + with pytest.raises(ValueError): + runtime.verify_container(record, "a" * 64, "ours", image, "c" * 64, "sha256:" + "d" * 64, "runtime") + + +def image_binding_fixture(store): + index_digest, child_digest, config_digest = ["sha256:" + value * 64 for value in ("a", "b", "c")] + record = { + "Id": config_digest, + "RepoDigests": ["example/image@" + index_digest], + "Architecture": "amd64", + "Os": "linux", + "Config": {}, + } + index, platform = deepcopy(record), deepcopy(record) + if store == "containerd": + index["Id"], platform["Id"] = index_digest, child_digest + platform["Descriptor"] = {"digest": child_digest, "platform": {"architecture": "amd64", "os": "linux"}} + return index, platform, index_digest, child_digest, config_digest + + +@pytest.mark.parametrize("store", ["classic", "containerd"]) +def test_image_store_identity_binds_immutable_index_platform_and_config(store): + result = runtime.bind_local_image(*image_binding_fixture(store)) + assert result["ImageStore"] == store + assert result["PlatformConfigDigest"] == "sha256:" + "c" * 64 + + +@pytest.mark.parametrize( + "mutation", ["index_id", "child_id", "descriptor", "platform", "repo_digest", "missing_descriptor"] +) +def test_platform_image_binding_refuses_an_ambiguous_or_foreign_local_object(mutation): + index, platform, index_digest, child_digest, config_digest = image_binding_fixture("containerd") + if mutation == "index_id": + index["Id"] = "sha256:" + "e" * 64 + elif mutation == "child_id": + platform["Id"] = "sha256:" + "e" * 64 + elif mutation == "descriptor": + platform["Descriptor"]["digest"] = "sha256:" + "e" * 64 + elif mutation == "platform": + platform["Descriptor"]["platform"]["architecture"] = "arm64" + elif mutation == "repo_digest": + platform["RepoDigests"] = ["example/image@sha256:" + "e" * 64] + else: + del platform["Descriptor"] + with pytest.raises(ValueError): + runtime.bind_local_image(index, platform, index_digest, child_digest, config_digest)