From 8e9a222759e904ab7b7da48c8572f5c8e03a390c Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 14:44:28 +0000 Subject: [PATCH 1/5] test(server): seed physical command-key upgrades through published APIs --- .../command-key-legacy-v1-vectors.json | 5 + scripts/ci/command_key_upgrade_seed.py | 148 ++++++++++++++++++ 2 files changed, 153 insertions(+) create mode 100644 scripts/ci/command_key_upgrade_seed.py diff --git a/resources/platform-protocol-specs/command-key-legacy-v1-vectors.json b/resources/platform-protocol-specs/command-key-legacy-v1-vectors.json index 8a3619209..e71f57442 100644 --- a/resources/platform-protocol-specs/command-key-legacy-v1-vectors.json +++ b/resources/platform-protocol-specs/command-key-legacy-v1-vectors.json @@ -126,6 +126,11 @@ } }, "physical_upgrade": { + "api_seed_driver": "scripts/ci/command_key_upgrade_seed.py", + "baseline_key_authority": { + "update": "Published 3.8 API admissions retain the raw caller request_id. These are physical legacy-key upgrade seeds.", + "signal": "Published 3.8 ordinary signal rows have request_id NULL; only bridge identities are stored. They are unkeyed controls, not physical retained caller-key proof. Do not manufacture a raw signal key or infer equality." + }, "scope": "Separate actual old-image admission and schema-upgrade cell. Do not substitute the synthetic transform or minimal orphan command-row schema tests.", "baseline": { "tag": "3.8.0", diff --git a/scripts/ci/command_key_upgrade_seed.py b/scripts/ci/command_key_upgrade_seed.py new file mode 100644 index 000000000..b0bdcd096 --- /dev/null +++ b/scripts/ci/command_key_upgrade_seed.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python3 +"""Write real 3.8 API receipts for the separately qualified legacy upgrade cell. + +Run with the published Python SDK installed for its official Avro codec. This +does not run an SDK Worker or transform database rows. Stop the disposable +baseline Server and back up its database before the manifest's upgrade step. +Published 3.8 retains ordinary update keys; its ordinary signal rows have no +stored caller key and are controls, not physical legacy-key readback proof. +""" + +import argparse +import hashlib +import importlib.metadata +import json +from pathlib import Path +import re +from urllib.error import HTTPError +from urllib.parse import urlparse +from urllib.request import Request, urlopen + + +class SeedClient: + def __init__(self, url): + parsed = urlparse(url) + if parsed.scheme != "http" or parsed.hostname != "127.0.0.1" or not parsed.port or parsed.path: + raise ValueError("seed URL must be an explicit disposable http://127.0.0.1:port") + self.url = url + self.protocol = None + + def request(self, method, path, body=None, expected=200): + headers = {"Authorization": "Bearer test-token", "X-Namespace": "default", + "Accept": "application/json", "Content-Type": "application/json"} + if path.startswith("/api/worker/"): + headers["X-Durable-Workflow-Protocol-Version"] = self.protocol + else: + headers["X-Durable-Workflow-Control-Plane-Version"] = "2" + data = None if body is None else json.dumps(body).encode() + try: + response = urlopen(Request(self.url + path, data=data, headers=headers, method=method), timeout=30) + except HTTPError as error: + response = error + with response: + raw = response.read(1_048_577) + if len(raw) > 1_048_576: + raise ValueError("seed response exceeds 1 MiB") + value = json.loads(raw) + if response.status != expected: + raise ValueError(f"{method} {path}: expected {expected}, got {response.status}: {value}") + return value + + def poll(self, worker, queue): + value = self.request("POST", "/api/worker/workflow-tasks/poll", + {"worker_id": worker, "task_queue": queue, "timeout_seconds": 0}) + task = value.get("task") + if not isinstance(task, dict) or not task.get("task_id"): + raise ValueError("seed expected a real leased workflow task") + return task + + def finish(self, task, commands): + return self.request("POST", f"/api/worker/workflow-tasks/{task['task_id']}/complete", + {"lease_owner": task["lease_owner"], + "workflow_task_attempt": task["workflow_task_attempt"], "commands": commands}) + + +def seed(client, prefix, manifest): + info = client.request("GET", "/api/cluster/info") + if info.get("version") != manifest["physical_upgrade"]["baseline"]["tag"]: + raise ValueError("physical seed must run on the manifest's published baseline version") + client.protocol = info["worker_protocol"]["version"] + # Import only for the codec. All seed admissions below use raw producer APIs. + from durable_workflow import serializer + + arguments = serializer.envelope([manifest["legacy_readback"]["original_input"]], codec="avro") + null = serializer.envelope(None, codec="avro") + for value, envelope in [([1], arguments), (None, null)]: + if serializer.decode_envelope(envelope, codec="avro") != value: + raise ValueError("official codec round-trip failed") + parameters = [{"name": "value", "type": "int", "required": True, "variadic": False, + "default_available": False, "allows_null": False}] + cases = [] + for kind in manifest["legacy_readback"]["types"]: + for state in ["open", "completed", "continued"]: + workflow = f"{prefix}{kind}-{state}" + worker, queue = workflow + "-worker", workflow + "-queue" + client.request("POST", "/api/worker/register", { + "worker_id": worker, "task_queue": queue, "runtime": "php", + "supported_workflow_types": ["echo"], "supported_activity_types": [], + "capabilities": ["workflow_updates"], + "capability_manifest": {name: {"supported": False, "minimum_protocol_version": "1.18"} + for name in ["local_activities", "worker_sessions", "sticky_execution"]}, + "workflow_command_contracts": {"echo": { + "signals": ["finish"], "signal_contracts": [{"name": "finish", "parameters": parameters}], + "updates": ["set"], "update_validators": [], + "update_contracts": [{"name": "set", "parameters": parameters}]}}}, expected=201) + started = client.request("POST", "/api/workflows", { + "workflow_id": workflow, "workflow_type": "echo", "task_queue": queue, + "input": arguments}, expected=201) + task = client.poll(worker, queue) + if task["run_id"] != started["run_id"]: + raise ValueError("seed claimed another workflow's task") + client.finish(task, [{"type": "open_signal_wait", "signal_name": "finish"}]) + name = "finish" if kind == "signal" else "set" + path = f"/api/workflows/{workflow}/{kind}/{name}" + body = {"input": arguments, "request_id": manifest["keys"]["original"]} + if kind == "update": + body["wait_for"] = "accepted" + receipt = client.request("POST", path, body, expected=202) + if state != "open": + task = client.poll(worker, queue) + if task["run_id"] != started["run_id"]: + raise ValueError("seed close claimed another workflow's task") + commands = [] + if kind == "update": + commands.append({"type": "complete_update", "update_id": receipt["update_id"], "result": null}) + commands.append({"type": "complete_workflow" if state == "completed" else "continue_as_new", + "result": null}) + client.finish(task, commands) + history = client.request("GET", f"/api/workflows/{workflow}/runs/{started['run_id']}/history") + cases.append({"kind": kind, "state": state, "workflow_id": workflow, + "path": path, "original_body": body, "receipt": receipt, "history": history}) + return {"schema": "durable-workflow.command-key-physical-seed.v1", + "scope": "Real baseline API admissions; not a synthetic transform, upgrade result or SDK Worker proof.", + "baseline": manifest["physical_upgrade"]["baseline"], "advertised_protocol": client.protocol, + "codec_package": {"name": "durable-workflow", "version": importlib.metadata.version("durable-workflow")}, + "cases": cases} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--url", required=True) + parser.add_argument("--prefix", required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + if not re.fullmatch(r"[A-Za-z0-9-]{1,32}", args.prefix): + parser.error("prefix must contain 1–32 ASCII letters, digits or hyphens") + if args.output.exists(): + parser.error("output must be a new file") + root = Path(__file__).resolve().parents[2] + raw = (root / "resources/platform-protocol-specs/command-key-legacy-v1-vectors.json").read_bytes() + result = seed(SeedClient(args.url), args.prefix, json.loads(raw)) + result["manifest_sha256"] = hashlib.sha256(raw).hexdigest() + with args.output.open("x") as output: + output.write(json.dumps(result, indent=2) + "\n") + print(f"Seeded {len(result['cases'])} real baseline receipts. Stop and back up before upgrading.") + + +if __name__ == "__main__": + main() From 60a16911c058c4ada598b19706511b3347c8afbb Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 14:48:05 +0000 Subject: [PATCH 2/5] Clarify physical unkeyed signal upgrade authority --- .../platform-protocol-specs/command-key-legacy-v1-vectors.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/resources/platform-protocol-specs/command-key-legacy-v1-vectors.json b/resources/platform-protocol-specs/command-key-legacy-v1-vectors.json index e71f57442..892b5d049 100644 --- a/resources/platform-protocol-specs/command-key-legacy-v1-vectors.json +++ b/resources/platform-protocol-specs/command-key-legacy-v1-vectors.json @@ -129,7 +129,7 @@ "api_seed_driver": "scripts/ci/command_key_upgrade_seed.py", "baseline_key_authority": { "update": "Published 3.8 API admissions retain the raw caller request_id. These are physical legacy-key upgrade seeds.", - "signal": "Published 3.8 ordinary signal rows have request_id NULL; only bridge identities are stored. They are unkeyed controls, not physical retained caller-key proof. Do not manufacture a raw signal key or infer equality." + "signal": "Published 3.8 ordinary signal rows have request_id NULL; only bridge identities are stored. After upgrade preserve exact rows, identity and business outcome as compatibility only, not retained caller-key recovery. Do not manufacture a raw signal key, retained/intent metadata or equality." }, "scope": "Separate actual old-image admission and schema-upgrade cell. Do not substitute the synthetic transform or minimal orphan command-row schema tests.", "baseline": { From c7245e922a50e9d08466e40be32e1e014c032976 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 15:03:06 +0000 Subject: [PATCH 3/5] Bind physical seed transport codec and observed provenance --- .github/workflows/phpunit-feature.yml | 4 +- scripts/ci/command_key_upgrade_seed.py | 70 +++++-- scripts/ci/test_command_key_upgrade_seed.py | 195 ++++++++++++++++++++ 3 files changed, 256 insertions(+), 13 deletions(-) create mode 100644 scripts/ci/test_command_key_upgrade_seed.py diff --git a/.github/workflows/phpunit-feature.yml b/.github/workflows/phpunit-feature.yml index 8b1e7181b..4a9182d41 100644 --- a/.github/workflows/phpunit-feature.yml +++ b/.github/workflows/phpunit-feature.yml @@ -187,7 +187,9 @@ jobs: python3 scripts/ci/verify-php-apt-source.py - name: Check the shared legacy command-key seed and authority scopes - run: python3 scripts/ci/test_command_key_legacy_fixture.py + run: | + python3 scripts/ci/test_command_key_legacy_fixture.py + python3 -I scripts/ci/test_command_key_upgrade_seed.py - name: Check published failover image identity and recovery result gates run: python3 -m unittest discover -s tests/Unit/Support -p 'single_region_failover*test.py' diff --git a/scripts/ci/command_key_upgrade_seed.py b/scripts/ci/command_key_upgrade_seed.py index b0bdcd096..96f3950e5 100644 --- a/scripts/ci/command_key_upgrade_seed.py +++ b/scripts/ci/command_key_upgrade_seed.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """Write real 3.8 API receipts for the separately qualified legacy upgrade cell. -Run with the published Python SDK installed for its official Avro codec. This +Run with ``python -I`` and the published Python SDK installed for its Avro codec. This does not run an SDK Worker or transform database rows. Stop the disposable baseline Server and back up its database before the manifest's upgrade step. Published 3.8 retains ordinary update keys; its ordinary signal rows have no @@ -9,22 +9,68 @@ """ import argparse +import base64 import hashlib +import importlib +import importlib.machinery import importlib.metadata import json from pathlib import Path import re +import sys from urllib.error import HTTPError -from urllib.parse import urlparse -from urllib.request import Request, urlopen +from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener + + +CODEC_VERSION = "2.17.1" + + +class NoRedirect(HTTPRedirectHandler): + def redirect_request(self, request, response, code, message, headers, url): + return None + + +def installed_codec(): + """Check package resolution and RECORD hashes before importing SDK code.""" + if not sys.flags.isolated: + raise ValueError("run the seed driver with python -I, without cwd/PYTHONPATH imports") + distribution = importlib.metadata.distribution("durable-workflow") + if distribution.version != CODEC_VERSION: + raise ValueError("seed codec must be installed durable-workflow " + CODEC_VERSION) + expected = {} + for relative in ["durable_workflow/__init__.py", "durable_workflow/serializer.py"]: + files = [item for item in distribution.files or [] if str(item) == relative] + if len(files) != 1 or files[0].hash is None or files[0].hash.mode != "sha256": + raise ValueError("installed codec has no unique SHA256 RECORD entry") + path = Path(distribution.locate_file(files[0])).resolve(strict=True) + actual = hashlib.sha256(path.read_bytes()).digest() + if base64.urlsafe_b64encode(actual).decode().rstrip("=") != files[0].hash.value: + raise ValueError("installed codec bytes differ from their distribution RECORD") + expected[relative] = (path, actual.hex()) + package = importlib.machinery.PathFinder.find_spec("durable_workflow", sys.path) + if package is None or package.origin is None or Path(package.origin).resolve() != expected["durable_workflow/__init__.py"][0]: + raise ValueError("codec package is shadowed outside its installed distribution") + spec = importlib.machinery.PathFinder.find_spec("durable_workflow.serializer", package.submodule_search_locations) + if spec is None or spec.origin is None or Path(spec.origin).resolve() != expected["durable_workflow/serializer.py"][0]: + raise ValueError("serializer is shadowed outside its installed distribution") + for name, relative in [("durable_workflow", "durable_workflow/__init__.py"), + ("durable_workflow.serializer", "durable_workflow/serializer.py")]: + if name in sys.modules and Path(getattr(sys.modules[name], "__file__", "")).resolve() != expected[relative][0]: + raise ValueError("preloaded codec module is outside its installed distribution") + serializer = importlib.import_module("durable_workflow.serializer") + if Path(serializer.__file__).resolve() != expected["durable_workflow/serializer.py"][0]: + raise ValueError("imported serializer differs from the checked distribution") + return serializer, {"name": "durable-workflow", "version": distribution.version, + "serializer_sha256": expected["durable_workflow/serializer.py"][1]} class SeedClient: def __init__(self, url): - parsed = urlparse(url) - if parsed.scheme != "http" or parsed.hostname != "127.0.0.1" or not parsed.port or parsed.path: + match = re.fullmatch(r"http://127\.0\.0\.1:([1-9][0-9]{0,4})", url) + if match is None or int(match[1]) > 65535: raise ValueError("seed URL must be an explicit disposable http://127.0.0.1:port") - self.url = url + self.url = "http://127.0.0.1:" + str(int(match[1])) + self.opener = build_opener(ProxyHandler({}), NoRedirect()) self.protocol = None def request(self, method, path, body=None, expected=200): @@ -36,7 +82,7 @@ def request(self, method, path, body=None, expected=200): headers["X-Durable-Workflow-Control-Plane-Version"] = "2" data = None if body is None else json.dumps(body).encode() try: - response = urlopen(Request(self.url + path, data=data, headers=headers, method=method), timeout=30) + response = self.opener.open(Request(self.url + path, data=data, headers=headers, method=method), timeout=30) except HTTPError as error: response = error with response: @@ -63,13 +109,11 @@ def finish(self, task, commands): def seed(client, prefix, manifest): + serializer, codec_identity = installed_codec() info = client.request("GET", "/api/cluster/info") if info.get("version") != manifest["physical_upgrade"]["baseline"]["tag"]: raise ValueError("physical seed must run on the manifest's published baseline version") client.protocol = info["worker_protocol"]["version"] - # Import only for the codec. All seed admissions below use raw producer APIs. - from durable_workflow import serializer - arguments = serializer.envelope([manifest["legacy_readback"]["original_input"]], codec="avro") null = serializer.envelope(None, codec="avro") for value, envelope in [([1], arguments), (None, null)]: @@ -120,8 +164,10 @@ def seed(client, prefix, manifest): "path": path, "original_body": body, "receipt": receipt, "history": history}) return {"schema": "durable-workflow.command-key-physical-seed.v1", "scope": "Real baseline API admissions; not a synthetic transform, upgrade result or SDK Worker proof.", - "baseline": manifest["physical_upgrade"]["baseline"], "advertised_protocol": client.protocol, - "codec_package": {"name": "durable-workflow", "version": importlib.metadata.version("durable-workflow")}, + "expected_external_baseline": manifest["physical_upgrade"]["baseline"], + "external_identity_scope": "Image digest and source require verification by the outer runner; this driver does not observe them.", + "observed_cluster": {"version": info["version"], "worker_protocol": client.protocol}, + "codec_package": codec_identity, "cases": cases} diff --git a/scripts/ci/test_command_key_upgrade_seed.py b/scripts/ci/test_command_key_upgrade_seed.py new file mode 100644 index 000000000..052340fe1 --- /dev/null +++ b/scripts/ci/test_command_key_upgrade_seed.py @@ -0,0 +1,195 @@ +"""Boundary controls; run with python -I, no SDK install or database needed.""" +import base64 +from contextlib import contextmanager +import hashlib +from http.server import BaseHTTPRequestHandler, HTTPServer +import importlib.metadata +import importlib.util +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import threading +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +SCRIPT = Path(__file__).with_name("command_key_upgrade_seed.py") +SPEC = importlib.util.spec_from_file_location("command_key_seed", SCRIPT) +seed = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(seed) + + +@contextmanager +def server(redirect=False): + class Handler(BaseHTTPRequestHandler): + calls = [] + + def do_GET(self): + self.calls.append(self.path) + self.send_response(302 if redirect else 200) + if redirect: + self.send_header("Location", "http://127.0.0.1:1/outside") + self.end_headers() + self.wfile.write(json.dumps({"ok": True}).encode()) + + def log_message(self, *args): + pass + + with HTTPServer(("127.0.0.1", 0), Handler) as http: + thread = threading.Thread(target=http.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{http.server_port}", Handler.calls + finally: + http.shutdown() + thread.join(timeout=2) + + +class OriginTests(unittest.TestCase): + def test_only_canonical_numeric_loopback_origin_is_accepted(self): + for port in [1, 1234, 65535]: + self.assertEqual(seed.SeedClient(f"http://127.0.0.1:{port}").url, f"http://127.0.0.1:{port}") + for value in ["http://user:pass@127.0.0.1:1234", "http://127.0.0.1:1234?next=http://example.invalid", + "http://127.0.0.1:1234#fragment", "http://127.0.0.1:01234", "http://127.0.0.1:0", + "http://127.0.0.1:65536", "http://127.0.0.1:1234/", "http://127.0.0.1:1234/api", + "http://127.0.0.1", "http://localhost:1234", "https://127.0.0.1:1234", + "http://example.invalid:1234", " HTTP://127.0.0.1:1234", "http://127.0.0.1:1234\n"]: + with self.subTest(value=value), self.assertRaises(ValueError): + seed.SeedClient(value) + + def test_hostile_proxy_environment_does_not_route_the_request(self): + proxy = "http://127.0.0.1:1" + with server() as (origin, calls), patch.dict(os.environ, { + "http_proxy": proxy, "HTTP_PROXY": proxy, "HTTPS_PROXY": proxy, + "ALL_PROXY": proxy, "no_proxy": "", "NO_PROXY": ""}): + self.assertEqual(seed.SeedClient(origin).request("GET", "/api/cluster/info"), {"ok": True}) + self.assertEqual(calls, ["/api/cluster/info"]) + + def test_redirect_is_refused_without_contacting_its_target(self): + with server(redirect=True) as (origin, calls): + with self.assertRaisesRegex(ValueError, "got 302"): + seed.SeedClient(origin).request("GET", "/api/cluster/info") + self.assertEqual(calls, ["/api/cluster/info"]) + + +class CodecTests(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory() + self.addCleanup(self.directory.cleanup) + self.root = Path(self.directory.name) + package = self.root / "installed/durable_workflow" + package.mkdir(parents=True) + (package / "__init__.py").write_text("") + (package / "serializer.py").write_text("fixture_identity = 'recorded'\n" + "def envelope(value, **kwargs): return {'value': value}\n" + "def decode_envelope(value, **kwargs): return value['value']\n") + files = [] + for relative in ["durable_workflow/__init__.py", "durable_workflow/serializer.py"]: + entry = importlib.metadata.PackagePath(relative) + digest = hashlib.sha256((package.parent / relative).read_bytes()).digest() + entry.hash = importlib.metadata.FileHash("sha256=" + base64.urlsafe_b64encode(digest).decode().rstrip("=")) + files.append(entry) + self.distribution = SimpleNamespace(version=seed.CODEC_VERSION, files=files, + locate_file=lambda entry: package.parent / str(entry)) + self.package = package + self.addCleanup(patch.stopall) + patch.object(seed.importlib.metadata, "distribution", return_value=self.distribution).start() + patch.object(sys, "path", [str(package.parent)] + sys.path).start() + patch.dict(sys.modules).start() + sys.modules.pop("durable_workflow", None) + sys.modules.pop("durable_workflow.serializer", None) + importlib.invalidate_caches() + + def test_recorded_installed_module_and_hash_are_bound(self): + serializer, identity = seed.installed_codec() + self.assertEqual(serializer.fixture_identity, "recorded") + self.assertEqual(identity["version"], seed.CODEC_VERSION) + self.assertEqual(identity["serializer_sha256"], hashlib.sha256((self.package / "serializer.py").read_bytes()).hexdigest()) + + def test_mismatched_distribution_version_refuses_before_import(self): + self.distribution.version = "0.0.0" + with self.assertRaisesRegex(ValueError, "must be installed"): + seed.installed_codec() + self.assertNotIn("durable_workflow", sys.modules) + + def test_modified_serializer_bytes_refuse_before_import(self): + (self.package / "serializer.py").write_text("raise RuntimeError('must not execute')\n") + with self.assertRaisesRegex(ValueError, "RECORD"): + seed.installed_codec() + self.assertNotIn("durable_workflow", sys.modules) + + def test_cwd_or_pythonpath_package_shadow_never_executes(self): + shadow = self.root / "shadow/durable_workflow" + shadow.mkdir(parents=True) + marker = self.root / "executed" + (shadow / "__init__.py").write_text(f"from pathlib import Path; Path({str(marker)!r}).touch()\n") + sys.path.insert(0, str(shadow.parent)) + with self.assertRaisesRegex(ValueError, "shadowed"): + seed.installed_codec() + self.assertFalse(marker.exists()) + + def test_preloaded_serializer_from_another_origin_is_refused(self): + sys.modules["durable_workflow.serializer"] = SimpleNamespace(__file__=str(self.root / "shadow.py")) + with self.assertRaisesRegex(ValueError, "preloaded"): + seed.installed_codec() + + def test_nonisolated_process_refuses_ambient_pythonpath(self): + code = f"import importlib.util; s=importlib.util.spec_from_file_location('seed',{str(SCRIPT)!r}); m=importlib.util.module_from_spec(s); s.loader.exec_module(m); m.installed_codec()" + result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, timeout=5) + self.assertNotEqual(result.returncode, 0) + self.assertIn("python -I", result.stderr) + + def test_seed_rejects_wrong_codec_version_before_any_http(self): + self.distribution.version = "0.0.0" + client = SimpleNamespace(request=unittest.mock.Mock(return_value={"version": "3.8.0", + "worker_protocol": {"version": "1.20"}})) + manifest = {"physical_upgrade": {"baseline": {"tag": "3.8.0"}}, + "legacy_readback": {"original_input": 1, "types": []}} + with self.assertRaisesRegex(ValueError, "must be installed"): + seed.seed(client, "fixture-", manifest) + client.request.assert_not_called() + + def test_seed_rejects_executable_shadow_before_import_or_http(self): + shadow = self.root / "shadow/durable_workflow" + shadow.mkdir(parents=True) + marker = self.root / "executed" + (shadow / "__init__.py").write_text(f"from pathlib import Path; Path({str(marker)!r}).touch()\n") + (shadow / "serializer.py").write_text((self.package / "serializer.py").read_text()) + sys.path.insert(0, str(shadow.parent)) + client = SimpleNamespace(request=unittest.mock.Mock(return_value={"version": "3.8.0", + "worker_protocol": {"version": "1.20"}})) + manifest = {"physical_upgrade": {"baseline": {"tag": "3.8.0"}}, + "legacy_readback": {"original_input": 1, "types": []}} + with self.assertRaisesRegex(ValueError, "shadowed"): + seed.seed(client, "fixture-", manifest) + self.assertFalse(marker.exists()) + client.request.assert_not_called() + + +class ProducerTests(unittest.TestCase): + def test_nonbaseline_cluster_refuses_before_admission(self): + client = SimpleNamespace(request=unittest.mock.Mock(return_value={"version": "3.9.1"})) + with patch.object(seed, "installed_codec", return_value=(None, {})), self.assertRaisesRegex(ValueError, "baseline"): + seed.seed(client, "fixture-", {"physical_upgrade": {"baseline": {"tag": "3.8.0"}}}) + client.request.assert_called_once_with("GET", "/api/cluster/info") + + def test_receipt_labels_expected_external_and_observed_cluster_separately(self): + expected = {"tag": "3.8.0", "source_commit": "a" * 40, "index_digest": "sha256:" + "b" * 64} + manifest = {"physical_upgrade": {"baseline": expected}, "legacy_readback": {"original_input": 1, "types": []}} + codec = SimpleNamespace(envelope=lambda value, **kwargs: {"value": value}, + decode_envelope=lambda envelope, **kwargs: envelope["value"]) + client = SimpleNamespace(request=unittest.mock.Mock(return_value={"version": "3.8.0", "worker_protocol": {"version": "1.20"}})) + with patch.object(seed, "installed_codec", return_value=(codec, {"version": seed.CODEC_VERSION})): + receipt = seed.seed(client, "fixture-", manifest) + self.assertEqual(receipt["expected_external_baseline"], expected) + self.assertEqual(receipt["observed_cluster"], {"version": "3.8.0", "worker_protocol": "1.20"}) + self.assertNotIn("baseline", receipt) + self.assertIn("does not observe", receipt["external_identity_scope"]) + client.request.assert_called_once_with("GET", "/api/cluster/info") + + +if __name__ == "__main__": + unittest.main() From 09a4ccda3da11853606af031a8970bbde4b2d6ec Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 15:04:13 +0000 Subject: [PATCH 4/5] Run seed predecessor controls without successor-only setup --- scripts/ci/test_command_key_upgrade_seed.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/ci/test_command_key_upgrade_seed.py b/scripts/ci/test_command_key_upgrade_seed.py index 052340fe1..46384a700 100644 --- a/scripts/ci/test_command_key_upgrade_seed.py +++ b/scripts/ci/test_command_key_upgrade_seed.py @@ -92,7 +92,7 @@ def setUp(self): digest = hashlib.sha256((package.parent / relative).read_bytes()).digest() entry.hash = importlib.metadata.FileHash("sha256=" + base64.urlsafe_b64encode(digest).decode().rstrip("=")) files.append(entry) - self.distribution = SimpleNamespace(version=seed.CODEC_VERSION, files=files, + self.distribution = SimpleNamespace(version="2.17.1", files=files, locate_file=lambda entry: package.parent / str(entry)) self.package = package self.addCleanup(patch.stopall) From 6aaf6382c257928b957037f7d991ab6de615ce98 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 15:20:04 +0000 Subject: [PATCH 5/5] Simplify seed codec checks and scenario execution for quality gates --- scripts/ci/command_key_upgrade_seed.py | 135 +++++++++++--------- scripts/ci/test_command_key_upgrade_seed.py | 5 +- 2 files changed, 75 insertions(+), 65 deletions(-) diff --git a/scripts/ci/command_key_upgrade_seed.py b/scripts/ci/command_key_upgrade_seed.py index 96f3950e5..9838c3ab0 100644 --- a/scripts/ci/command_key_upgrade_seed.py +++ b/scripts/ci/command_key_upgrade_seed.py @@ -23,6 +23,10 @@ CODEC_VERSION = "2.17.1" +CODEC_PACKAGE = "durable_workflow" +CODEC_MODULE = CODEC_PACKAGE + ".serializer" +CODEC_INIT = CODEC_PACKAGE + "/__init__.py" +CODEC_SOURCE = CODEC_PACKAGE + "/serializer.py" class NoRedirect(HTTPRedirectHandler): @@ -30,6 +34,17 @@ def redirect_request(self, request, response, code, message, headers, url): return None +def recorded_codec_file(distribution, relative): + files = [item for item in distribution.files or [] if str(item) == relative] + if len(files) != 1 or files[0].hash is None or files[0].hash.mode != "sha256": + raise ValueError("installed codec has no unique SHA256 RECORD entry") + path = Path(distribution.locate_file(files[0])).resolve(strict=True) + actual = hashlib.sha256(path.read_bytes()).digest() + if base64.urlsafe_b64encode(actual).decode().rstrip("=") != files[0].hash.value: + raise ValueError("installed codec bytes differ from their distribution RECORD") + return path, actual.hex() + + def installed_codec(): """Check package resolution and RECORD hashes before importing SDK code.""" if not sys.flags.isolated: @@ -37,36 +52,27 @@ def installed_codec(): distribution = importlib.metadata.distribution("durable-workflow") if distribution.version != CODEC_VERSION: raise ValueError("seed codec must be installed durable-workflow " + CODEC_VERSION) - expected = {} - for relative in ["durable_workflow/__init__.py", "durable_workflow/serializer.py"]: - files = [item for item in distribution.files or [] if str(item) == relative] - if len(files) != 1 or files[0].hash is None or files[0].hash.mode != "sha256": - raise ValueError("installed codec has no unique SHA256 RECORD entry") - path = Path(distribution.locate_file(files[0])).resolve(strict=True) - actual = hashlib.sha256(path.read_bytes()).digest() - if base64.urlsafe_b64encode(actual).decode().rstrip("=") != files[0].hash.value: - raise ValueError("installed codec bytes differ from their distribution RECORD") - expected[relative] = (path, actual.hex()) - package = importlib.machinery.PathFinder.find_spec("durable_workflow", sys.path) - if package is None or package.origin is None or Path(package.origin).resolve() != expected["durable_workflow/__init__.py"][0]: + expected = {relative: recorded_codec_file(distribution, relative) for relative in [CODEC_INIT, CODEC_SOURCE]} + package = importlib.machinery.PathFinder.find_spec(CODEC_PACKAGE, sys.path) + if package is None or package.origin is None or Path(package.origin).resolve() != expected[CODEC_INIT][0]: raise ValueError("codec package is shadowed outside its installed distribution") - spec = importlib.machinery.PathFinder.find_spec("durable_workflow.serializer", package.submodule_search_locations) - if spec is None or spec.origin is None or Path(spec.origin).resolve() != expected["durable_workflow/serializer.py"][0]: + spec = importlib.machinery.PathFinder.find_spec(CODEC_MODULE, package.submodule_search_locations) + if spec is None or spec.origin is None or Path(spec.origin).resolve() != expected[CODEC_SOURCE][0]: raise ValueError("serializer is shadowed outside its installed distribution") - for name, relative in [("durable_workflow", "durable_workflow/__init__.py"), - ("durable_workflow.serializer", "durable_workflow/serializer.py")]: + for name, relative in [(CODEC_PACKAGE, CODEC_INIT), (CODEC_MODULE, CODEC_SOURCE)]: if name in sys.modules and Path(getattr(sys.modules[name], "__file__", "")).resolve() != expected[relative][0]: raise ValueError("preloaded codec module is outside its installed distribution") - serializer = importlib.import_module("durable_workflow.serializer") - if Path(serializer.__file__).resolve() != expected["durable_workflow/serializer.py"][0]: + serializer = importlib.import_module(CODEC_MODULE) + if Path(serializer.__file__).resolve() != expected[CODEC_SOURCE][0]: raise ValueError("imported serializer differs from the checked distribution") return serializer, {"name": "durable-workflow", "version": distribution.version, - "serializer_sha256": expected["durable_workflow/serializer.py"][1]} + "serializer_sha256": expected[CODEC_SOURCE][1]} class SeedClient: def __init__(self, url): - match = re.fullmatch(r"http://127\.0\.0\.1:([1-9][0-9]{0,4})", url) + # The published local fixture serves HTTP. ASCII numeric loopback is the only origin. + match = re.fullmatch(r"http://127\.0\.0\.1:([1-9]\d{0,4})", url, flags=re.ASCII) if match is None or int(match[1]) > 65535: raise ValueError("seed URL must be an explicit disposable http://127.0.0.1:port") self.url = "http://127.0.0.1:" + str(int(match[1])) @@ -108,6 +114,49 @@ def finish(self, task, commands): "workflow_task_attempt": task["workflow_task_attempt"], "commands": commands}) +def seed_case(client, prefix, kind, state, arguments, null, key): + parameters = [{"name": "value", "type": "int", "required": True, "variadic": False, + "default_available": False, "allows_null": False}] + workflow = f"{prefix}{kind}-{state}" + worker, queue = workflow + "-worker", workflow + "-queue" + client.request("POST", "/api/worker/register", { + "worker_id": worker, "task_queue": queue, "runtime": "php", + "supported_workflow_types": ["echo"], "supported_activity_types": [], + "capabilities": ["workflow_updates"], + "capability_manifest": {name: {"supported": False, "minimum_protocol_version": "1.18"} + for name in ["local_activities", "worker_sessions", "sticky_execution"]}, + "workflow_command_contracts": {"echo": { + "signals": ["finish"], "signal_contracts": [{"name": "finish", "parameters": parameters}], + "updates": ["set"], "update_validators": [], + "update_contracts": [{"name": "set", "parameters": parameters}]}}}, expected=201) + started = client.request("POST", "/api/workflows", { + "workflow_id": workflow, "workflow_type": "echo", "task_queue": queue, + "input": arguments}, expected=201) + task = client.poll(worker, queue) + if task["run_id"] != started["run_id"]: + raise ValueError("seed claimed another workflow's task") + client.finish(task, [{"type": "open_signal_wait", "signal_name": "finish"}]) + name = "finish" if kind == "signal" else "set" + path = f"/api/workflows/{workflow}/{kind}/{name}" + body = {"input": arguments, "request_id": key} + if kind == "update": + body["wait_for"] = "accepted" + receipt = client.request("POST", path, body, expected=202) + if state != "open": + task = client.poll(worker, queue) + if task["run_id"] != started["run_id"]: + raise ValueError("seed close claimed another workflow's task") + commands = [] + if kind == "update": + commands.append({"type": "complete_update", "update_id": receipt["update_id"], "result": null}) + commands.append({"type": "complete_workflow" if state == "completed" else "continue_as_new", + "result": null}) + client.finish(task, commands) + history = client.request("GET", f"/api/workflows/{workflow}/runs/{started['run_id']}/history") + return {"kind": kind, "state": state, "workflow_id": workflow, + "path": path, "original_body": body, "receipt": receipt, "history": history} + + def seed(client, prefix, manifest): serializer, codec_identity = installed_codec() info = client.request("GET", "/api/cluster/info") @@ -119,49 +168,9 @@ def seed(client, prefix, manifest): for value, envelope in [([1], arguments), (None, null)]: if serializer.decode_envelope(envelope, codec="avro") != value: raise ValueError("official codec round-trip failed") - parameters = [{"name": "value", "type": "int", "required": True, "variadic": False, - "default_available": False, "allows_null": False}] - cases = [] - for kind in manifest["legacy_readback"]["types"]: - for state in ["open", "completed", "continued"]: - workflow = f"{prefix}{kind}-{state}" - worker, queue = workflow + "-worker", workflow + "-queue" - client.request("POST", "/api/worker/register", { - "worker_id": worker, "task_queue": queue, "runtime": "php", - "supported_workflow_types": ["echo"], "supported_activity_types": [], - "capabilities": ["workflow_updates"], - "capability_manifest": {name: {"supported": False, "minimum_protocol_version": "1.18"} - for name in ["local_activities", "worker_sessions", "sticky_execution"]}, - "workflow_command_contracts": {"echo": { - "signals": ["finish"], "signal_contracts": [{"name": "finish", "parameters": parameters}], - "updates": ["set"], "update_validators": [], - "update_contracts": [{"name": "set", "parameters": parameters}]}}}, expected=201) - started = client.request("POST", "/api/workflows", { - "workflow_id": workflow, "workflow_type": "echo", "task_queue": queue, - "input": arguments}, expected=201) - task = client.poll(worker, queue) - if task["run_id"] != started["run_id"]: - raise ValueError("seed claimed another workflow's task") - client.finish(task, [{"type": "open_signal_wait", "signal_name": "finish"}]) - name = "finish" if kind == "signal" else "set" - path = f"/api/workflows/{workflow}/{kind}/{name}" - body = {"input": arguments, "request_id": manifest["keys"]["original"]} - if kind == "update": - body["wait_for"] = "accepted" - receipt = client.request("POST", path, body, expected=202) - if state != "open": - task = client.poll(worker, queue) - if task["run_id"] != started["run_id"]: - raise ValueError("seed close claimed another workflow's task") - commands = [] - if kind == "update": - commands.append({"type": "complete_update", "update_id": receipt["update_id"], "result": null}) - commands.append({"type": "complete_workflow" if state == "completed" else "continue_as_new", - "result": null}) - client.finish(task, commands) - history = client.request("GET", f"/api/workflows/{workflow}/runs/{started['run_id']}/history") - cases.append({"kind": kind, "state": state, "workflow_id": workflow, - "path": path, "original_body": body, "receipt": receipt, "history": history}) + cases = [seed_case(client, prefix, kind, state, arguments, null, manifest["keys"]["original"]) + for kind in manifest["legacy_readback"]["types"] + for state in ["open", "completed", "continued"]] return {"schema": "durable-workflow.command-key-physical-seed.v1", "scope": "Real baseline API admissions; not a synthetic transform, upgrade result or SDK Worker proof.", "expected_external_baseline": manifest["physical_upgrade"]["baseline"], diff --git a/scripts/ci/test_command_key_upgrade_seed.py b/scripts/ci/test_command_key_upgrade_seed.py index 46384a700..37ade0bbd 100644 --- a/scripts/ci/test_command_key_upgrade_seed.py +++ b/scripts/ci/test_command_key_upgrade_seed.py @@ -53,7 +53,7 @@ def test_only_canonical_numeric_loopback_origin_is_accepted(self): for port in [1, 1234, 65535]: self.assertEqual(seed.SeedClient(f"http://127.0.0.1:{port}").url, f"http://127.0.0.1:{port}") for value in ["http://user:pass@127.0.0.1:1234", "http://127.0.0.1:1234?next=http://example.invalid", - "http://127.0.0.1:1234#fragment", "http://127.0.0.1:01234", "http://127.0.0.1:0", + "http://127.0.0.1:1234#fragment", "http://127.0.0.1:01234", "http://127.0.0.1:1١", "http://127.0.0.1:0", "http://127.0.0.1:65536", "http://127.0.0.1:1234/", "http://127.0.0.1:1234/api", "http://127.0.0.1", "http://localhost:1234", "https://127.0.0.1:1234", "http://example.invalid:1234", " HTTP://127.0.0.1:1234", "http://127.0.0.1:1234\n"]: @@ -70,8 +70,9 @@ def test_hostile_proxy_environment_does_not_route_the_request(self): def test_redirect_is_refused_without_contacting_its_target(self): with server(redirect=True) as (origin, calls): + client = seed.SeedClient(origin) with self.assertRaisesRegex(ValueError, "got 302"): - seed.SeedClient(origin).request("GET", "/api/cluster/info") + client.request("GET", "/api/cluster/info") self.assertEqual(calls, ["/api/cluster/info"])