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/resources/platform-protocol-specs/command-key-legacy-v1-vectors.json b/resources/platform-protocol-specs/command-key-legacy-v1-vectors.json index 8a3619209..892b5d049 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. 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": { "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..9838c3ab0 --- /dev/null +++ b/scripts/ci/command_key_upgrade_seed.py @@ -0,0 +1,203 @@ +#!/usr/bin/env python3 +"""Write real 3.8 API receipts for the separately qualified legacy upgrade cell. + +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 +stored caller key and are controls, not physical legacy-key readback proof. +""" + +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.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener + + +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): + 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: + 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 = {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(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 [(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(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[CODEC_SOURCE][1]} + + +class SeedClient: + def __init__(self, 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])) + self.opener = build_opener(ProxyHandler({}), NoRedirect()) + 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 = self.opener.open(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_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") + 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"] + 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") + 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"], + "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} + + +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() 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..37ade0bbd --- /dev/null +++ b/scripts/ci/test_command_key_upgrade_seed.py @@ -0,0 +1,196 @@ +"""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: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"]: + 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): + client = seed.SeedClient(origin) + with self.assertRaisesRegex(ValueError, "got 302"): + client.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="2.17.1", 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()