From 27628475034a6b78806de156d8a96c297371237d Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 17 Oct 2026 21:57:32 +0800 Subject: [PATCH] fix: preserve exact Schwab 00781 TCP probe in local revision plans Co-Authored-By: Codex --- ...build_schwab_notification_revision_plan.py | 90 ++++- scripts/verify_cloud_run_no_traffic_deploy.py | 31 +- ...build_schwab_notification_revision_plan.py | 363 ++++++++++++++++++ ...test_verify_cloud_run_no_traffic_deploy.py | 36 ++ 4 files changed, 493 insertions(+), 27 deletions(-) diff --git a/scripts/build_schwab_notification_revision_plan.py b/scripts/build_schwab_notification_revision_plan.py index 78a6b94..b7d01c1 100644 --- a/scripts/build_schwab_notification_revision_plan.py +++ b/scripts/build_schwab_notification_revision_plan.py @@ -43,6 +43,21 @@ "candidate_image", "resource_version", ), str) _ALIASES = {"latest", "latest-created", "latest-ready"} +HTTP_STARTUP_POLICY = "http-health-8080" +TCP_00781_STARTUP_POLICY = "schwab-00781-tcp-preserve" +_TCP_00781_IDENTITY = { + "project": "charlesschwabquant", "region": "us-central1", "service": "charles-schwab-quant-service", + "baseline_revision": "charles-schwab-quant-service-00781-lel", +} +_TCP_00781_IMAGE = ( + "us-central1-docker.pkg.dev/charlesschwabquant/cloud-run-source-deploy/" + "charlesschwabplatform/charles-schwab-quant-service@sha256:" + "cd4779376abf0ec4a54c0b04c690adc953756cda782ecd8adcf773e659a1733a" +) +_TCP_00781_SOURCE = "6c890e3fdfbc0f8dcd7f96b5a634cc482ba3606c" +_TCP_00781_PROBE = { + "failureThreshold": 1, "periodSeconds": 240, "tcpSocket": {"port": 8080}, "timeoutSeconds": 240, +} class PlanError(ValueError): @@ -85,7 +100,12 @@ def _image(value: object) -> bool: r"[A-Za-z0-9][A-Za-z0-9./_-]*@sha256:[0-9a-f]{64}", value) is not None) -def _expected(expected: object) -> None: +def _startup_policy(policy: object) -> None: + if type(policy) is not str or policy not in {HTTP_STARTUP_POLICY, TCP_00781_STARTUP_POLICY}: + _stop("INVALID_STARTUP_POLICY") + + +def _expected(expected: object, startup_policy: str = HTTP_STARTUP_POLICY) -> None: _required(expected, _EXPECTED) a._shape(expected, _EXPECTED) if (not all(_name(expected[key]) for key in _IDENTITY) @@ -100,6 +120,9 @@ def _expected(expected: object) -> None: _stop("INVALID_IMMUTABLE_IMAGE") if not _nonempty(expected["resource_version"]): _stop("INVALID_RESOURCE_VERSION") + if startup_policy == TCP_00781_STARTUP_POLICY and any( + expected[key] != value for key, value in _TCP_00781_IDENTITY.items()): + _stop("TCP_BASELINE_IDENTITY_MISMATCH") def _projection(snapshot: dict) -> dict: @@ -108,7 +131,7 @@ def _projection(snapshot: dict) -> dict: "runtime": revision["runtime"], "traffic": snapshot["service"]["traffic"]} -def _inventory(snapshot: object) -> None: +def _inventory(snapshot: object, startup_policy: str = HTTP_STARTUP_POLICY) -> None: # A never reads literal values during inventory. Inventory BOTH snapshots # before any subsequent equality, hash, deepcopy or literal type check. a._shape(snapshot, _SNAPSHOT) @@ -119,7 +142,13 @@ def _inventory(snapshot: object) -> None: for field in ("identity", "metadata", "provenance"): _required(part[field], schema[field]) _required(part["metadata"]["labels"], _LABELS) - a._inventory(_projection(snapshot)) + projection = _projection(snapshot) + if startup_policy == HTTP_STARTUP_POLICY: + a._inventory(projection) + else: + a._structure_inventory(projection) + if projection["runtime"]["containers"][0].get("startupProbe") != _TCP_00781_PROBE: + _stop("TCP_BASELINE_PROBE_MISMATCH") def _traffic(rows: list, expected: dict) -> None: @@ -146,7 +175,7 @@ def _traffic(rows: list, expected: dict) -> None: _stop("BASELINE_NOT_SERVING") -def _snapshot_values(snapshot: dict, expected: dict, *, candidate: bool = False) -> None: +def _snapshot_values(snapshot: dict, expected: dict, startup_policy: str = HTTP_STARTUP_POLICY, *, candidate: bool = False) -> None: revision_name = expected["candidate_revision" if candidate else "baseline_revision"] if snapshot["revision"]["identity"]["revision"] != revision_name: _stop("BASELINE_IDENTITY_MISMATCH" if not candidate else "CANDIDATE_IDENTITY_MISMATCH") @@ -170,6 +199,15 @@ def _snapshot_values(snapshot: dict, expected: dict, *, candidate: bool = False) _stop("RESOURCE_VERSION_MISMATCH") _traffic(snapshot["service"]["traffic"], expected) runtime = snapshot["revision"]["runtime"] + if startup_policy == TCP_00781_STARTUP_POLICY: + # These are exact LOCAL declarations, never proof of cloud origin or + # installed source. UID/RV and full equality still bind both snapshots. + revision = snapshot["revision"] + if not candidate and runtime["containers"][0]["image"] != _TCP_00781_IMAGE: + _stop("TCP_BASELINE_IMAGE_MISMATCH") + if (revision["provenance"]["commit"] != _TCP_00781_SOURCE + or revision["metadata"]["labels"]["commit-sha"] != _TCP_00781_SOURCE): + _stop("TCP_BASELINE_SOURCE_MISMATCH") if (not re.fullmatch(r"[A-Za-z0-9._-]+@[A-Za-z0-9.-]+", runtime["serviceAccountName"]) or runtime["ingress"] not in {"all", "internal", "internal-and-cloud-load-balancing"} or runtime["executionEnvironment"] not in {"gen1", "gen2"} @@ -186,15 +224,16 @@ def _snapshot_values(snapshot: dict, expected: dict, *, candidate: bool = False) if "value" in entry and type(entry["value"]) is not str: _stop("INVALID_ENV_LITERAL") if 8080 not in [port["containerPort"] for port in runtime["containers"][0]["ports"]]: - _stop("HTTP_STARTUP_GUARD_REQUIRED") + _stop("HTTP_STARTUP_GUARD_REQUIRED" if startup_policy == HTTP_STARTUP_POLICY else "TCP_BASELINE_PROBE_MISMATCH") -def _review_candidate(before: dict, candidate: dict, expected: dict) -> dict: - _expected(expected) - _inventory(before) - _inventory(candidate) - _snapshot_values(before, expected) - _snapshot_values(candidate, expected, candidate=True) +def _review_candidate(before: dict, candidate: dict, expected: dict, startup_policy: str = HTTP_STARTUP_POLICY) -> dict: + _startup_policy(startup_policy) + _expected(expected, startup_policy) + _inventory(before, startup_policy) + _inventory(candidate, startup_policy) + _snapshot_values(before, expected, startup_policy) + _snapshot_values(candidate, expected, startup_policy, candidate=True) old_image = before["revision"]["runtime"]["containers"][0]["image"] new_image = candidate["revision"]["runtime"]["containers"][0]["image"] if new_image != expected["candidate_image"]: @@ -214,7 +253,7 @@ def _review_candidate(before: dict, candidate: dict, expected: dict) -> dict: ("revision.runtime.containers[0].image", old_image, new_image), ): changes.append({"path": path, "comparison": "changed", "before_sha256": _hash(old), "after_sha256": _hash(new)}) - return { + review = { "model": MODEL, "mode": "review_only", "deployment_ready": False, "old_installation": "UNKNOWN", "startup_side_effects": "UNVERIFIED", "source": "NOT_VERIFIED", "admission": "NOT_VERIFIED", "cloud_configuration": "NOT_VERIFIED", @@ -224,12 +263,19 @@ def _review_candidate(before: dict, candidate: dict, expected: dict) -> dict: {"path": "revision_except_image_and_names", "comparison": "equal", "sha256": _hash(restored["revision"])}], "coverage": "CLOSED_SYNTHETIC_ENVELOPE_ONLY_OPAQUE_ENV_STRINGS", } + if startup_policy == TCP_00781_STARTUP_POLICY: + review["startup_assurance"] = { + "policy": startup_policy, "readiness": "TCP_PORT_READINESS_ONLY", + "http_application_health": "NOT_VERIFIED", "runtime_startup": "NOT_EXECUTED", + "baseline_binding": "LOCAL_DECLARATION_ONLY_NOT_CLOUD_PROVENANCE", + } + return review -def review_candidate(before: dict, candidate: dict, expected: dict) -> dict: +def review_candidate(before: dict, candidate: dict, expected: dict, *, startup_policy: str = HTTP_STARTUP_POLICY) -> dict: """Review an in-memory candidate without weakening probe or preservation guards.""" try: - return _review_candidate(before, candidate, expected) + return _review_candidate(before, candidate, expected, startup_policy) except PlanError: raise except a.PreflightError as exc: @@ -242,18 +288,20 @@ def build_plan(request: dict) -> tuple[dict, dict]: """Return a detached candidate and redacted plan; never inspect desired data.""" try: fields = {"model", "expected", "before", "current", "excluded_desired_template"} - a._keys(request, fields, fields) + a._keys(request, fields | {"startup_policy"}, fields) if request["model"] != MODEL: _stop("INVALID_MODEL") + startup_policy = request["startup_policy"] if "startup_policy" in request else HTTP_STARTUP_POLICY + _startup_policy(startup_policy) expected = request["expected"] - _expected(expected) + _expected(expected, startup_policy) before, current = request["before"], request["current"] if before is current: _stop("INDEPENDENT_SNAPSHOTS_REQUIRED") - _inventory(before) - _inventory(current) - _snapshot_values(before, expected) - _snapshot_values(current, expected) + _inventory(before, startup_policy) + _inventory(current, startup_policy) + _snapshot_values(before, expected, startup_policy) + _snapshot_values(current, expected, startup_policy) for key in ("service", "revision"): if before[key]["metadata"]["uid"] != current[key]["metadata"]["uid"]: _stop("OBJECT_RECREATED") @@ -263,7 +311,7 @@ def build_plan(request: dict) -> tuple[dict, dict]: candidate["revision"]["identity"]["revision"] = expected["candidate_revision"] candidate["revision"]["metadata"]["name"] = expected["candidate_revision"] candidate["revision"]["runtime"]["containers"][0]["image"] = expected["candidate_image"] - return candidate, _review_candidate(before, candidate, expected) + return candidate, _review_candidate(before, candidate, expected, startup_policy) except PlanError: raise except a.PreflightError as exc: diff --git a/scripts/verify_cloud_run_no_traffic_deploy.py b/scripts/verify_cloud_run_no_traffic_deploy.py index 4629390..1025c83 100644 --- a/scripts/verify_cloud_run_no_traffic_deploy.py +++ b/scripts/verify_cloud_run_no_traffic_deploy.py @@ -391,7 +391,7 @@ def _probe_inventory(probe: Mapping) -> None: _stop("INVALID_PROBE") -def _inventory(projection: object) -> None: +def _inventory_core(projection: object) -> None: _shape(projection, _PROJECTION) _keys(projection, _PROJECTION, _PROJECTION) _keys(projection["identity"], _PROJECTION["identity"], _PROJECTION["identity"]) @@ -443,10 +443,9 @@ def _inventory(projection: object) -> None: _stop("REQUIRED_LITERAL_MISSING") if any(entry["name"] in targets for container in containers[1:] for entry in container["env"]): _stop("AMBIGUOUS_OR_MISSING_TARGET") - startup = containers[0].get("startupProbe", {}) - if (startup.get("httpGet") != {"path": "/health", "port": 8080} - or "tcpSocket" in startup or "grpc" in startup): - _stop("HTTP_STARTUP_GUARD_REQUIRED") + + +def _volume_inventory(runtime: Mapping) -> None: volumes = runtime["volumes"] volume_names = [] for volume in volumes: @@ -467,13 +466,33 @@ def _inventory(projection: object) -> None: _stop("AMBIGUOUS_VOLUME") if len(volume_names) != len(set(volume_names)): _stop("AMBIGUOUS_VOLUME") - for container in containers: + for container in runtime["containers"]: for mount in container["volumeMounts"]: _keys(mount, _CONTAINER["volumeMounts"][0], {"name", "mountPath"}) if mount["name"] not in volume_names or not mount["mountPath"].startswith("/"): _stop("INVALID_VOLUME_MOUNT") +def _structure_inventory(projection: object) -> None: + """Inventory the closed local schema without choosing a startup policy.""" + _inventory_core(projection) + _volume_inventory(projection["runtime"]) + + +def _http_startup_guard(projection: Mapping) -> None: + startup = projection["runtime"]["containers"][0].get("startupProbe", {}) + if (startup.get("httpGet") != {"path": "/health", "port": 8080} + or "tcpSocket" in startup or "grpc" in startup): + _stop("HTTP_STARTUP_GUARD_REQUIRED") + + +def _inventory(projection: object) -> None: + """Keep the original API and CLI strictly on HTTP /health:8080.""" + _inventory_core(projection) + _http_startup_guard(projection) + _volume_inventory(projection["runtime"]) + + def _identity(projection: dict, args: argparse.Namespace, *, candidate: bool) -> None: identity = projection["identity"] expected_revision = args.candidate_revision if candidate else args.baseline_revision diff --git a/tests/test_build_schwab_notification_revision_plan.py b/tests/test_build_schwab_notification_revision_plan.py index 166cabf..2612ccb 100644 --- a/tests/test_build_schwab_notification_revision_plan.py +++ b/tests/test_build_schwab_notification_revision_plan.py @@ -757,3 +757,366 @@ def test_non_string_python_mapping_key_has_fixed_error(case): case["before"]["revision"]["runtime"]["network"][17] = SENTINEL with pytest.raises(planner().PlanError, match="^UNKNOWN_FIELD$"): planner().build_plan(case) + + +# Only these public, explicitly authorized identity declarations are real. All +# UID/RV, env, targets, accounts, secret references and candidate data stay fake. +TCP_POLICY = "schwab-00781-tcp-preserve" +HTTP_POLICY = "http-health-8080" +TCP_SERVICE = "charles-schwab-quant-service" +TCP_BASE = TCP_SERVICE + "-00781-lel" +TCP_IMAGE = ("us-central1-docker.pkg.dev/charlesschwabquant/cloud-run-source-deploy/" + "charlesschwabplatform/charles-schwab-quant-service@sha256:" + "cd4779376abf0ec4a54c0b04c690adc953756cda782ecd8adcf773e659a1733a") +TCP_SOURCE = "6c890e3fdfbc0f8dcd7f96b5a634cc482ba3606c" +TCP_PROBE = {"failureThreshold": 1, "periodSeconds": 240, "tcpSocket": {"port": 8080}, "timeoutSeconds": 240} + + +@pytest.fixture +def tcp_case(case): + case["startup_policy"] = TCP_POLICY + case["expected"].update(project="charlesschwabquant", region="us-central1", service=TCP_SERVICE, + baseline_revision=TCP_BASE, candidate_revision=TCP_SERVICE + "-00782-review") + for side in ("before", "current"): + payload = case[side] + for part in ("service", "revision"): + payload[part]["identity"].update(project="charlesschwabquant", region="us-central1", service=TCP_SERVICE) + payload["service"]["metadata"]["name"] = TCP_SERVICE + for row in payload["service"]["traffic"]: + row["revisionName"] = row["revisionName"].replace("demo-service", TCP_SERVICE) + payload["service"]["traffic"][0]["revisionName"] = TCP_BASE + payload["revision"]["identity"]["revision"] = TCP_BASE + payload["revision"]["metadata"]["name"] = TCP_BASE + payload["revision"]["metadata"]["labels"]["commit-sha"] = TCP_SOURCE + payload["revision"]["provenance"]["commit"] = TCP_SOURCE + main_container(payload)["image"] = TCP_IMAGE + main_container(payload)["startupProbe"] = copy.deepcopy(TCP_PROBE) + return case + + +def tcp_candidate(case): + """Hand derivation lets negative review tests run independently of build_plan.""" + candidate = copy.deepcopy(case["before"]) + candidate["revision"]["identity"]["revision"] = case["expected"]["candidate_revision"] + candidate["revision"]["metadata"]["name"] = case["expected"]["candidate_revision"] + main_container(candidate)["image"] = case["expected"]["candidate_image"] + return candidate + + +def tcp_review(case, candidate): + return planner().review_candidate(case["before"], candidate, case["expected"], startup_policy=TCP_POLICY) + + +def test_tcp_exact_00781_preserves_complete_context_without_mutation(tcp_case): + frozen = copy.deepcopy(tcp_case) + candidate, review = planner().build_plan(tcp_case) + assert candidate == tcp_candidate(tcp_case) + assert tcp_case == frozen + assert candidate["service"] == tcp_case["current"]["service"] + assert main_container(candidate)["startupProbe"] == TCP_PROBE + assert main_container(candidate)["env"][0]["value"].encode() == RAW_TARGET.encode() + assert tcp_review(tcp_case, candidate) == review + assert review["mode"] == "review_only" and review["deployment_ready"] is False + assert review["startup_side_effects"] == "UNVERIFIED" + assert review["source"] == review["cloud_configuration"] == "NOT_VERIFIED" + assert review["concurrency"] == "MATCHED_LOCAL_SNAPSHOTS_NO_CAS" + assert review["startup_assurance"] == { + "policy": TCP_POLICY, "readiness": "TCP_PORT_READINESS_ONLY", + "http_application_health": "NOT_VERIFIED", "runtime_startup": "NOT_EXECUTED", + "baseline_binding": "LOCAL_DECLARATION_ONLY_NOT_CLOUD_PROVENANCE", + } + again, repeated = planner().build_plan(copy.deepcopy(tcp_case)) + assert again == candidate and repeated == review + main_container(candidate)["env"].reverse() + candidate["service"]["metadata"]["labels"]["purpose"] = "detached" + assert tcp_case == frozen + + +@pytest.mark.parametrize("policy", [None, True, False, [], {}, 1, "allow_tcp", "skip", TCP_POLICY + "-other"]) +@pytest.mark.parametrize("entry", ["build", "review"]) +def test_tcp_policy_is_a_closed_explicit_enum(tcp_case, policy, entry): + candidate = tcp_candidate(tcp_case) + with pytest.raises(planner().PlanError, match="^INVALID_STARTUP_POLICY$"): + if entry == "build": + tcp_case["startup_policy"] = policy + planner().build_plan(tcp_case) + else: + planner().review_candidate(tcp_case["before"], candidate, tcp_case["expected"], startup_policy=policy) + + +@pytest.mark.parametrize("entry", ["build", "review"]) +@pytest.mark.parametrize("explicit", [False, True]) +@pytest.mark.parametrize("handler", ["tcpSocket", "grpc"]) +def test_tcp_or_grpc_exact_identity_cannot_change_default_http_policy(tcp_case, entry, explicit, handler): + for side in ("before", "current"): + main_container(tcp_case[side])["startupProbe"] = {handler: {"port": 8080}} + candidate = tcp_candidate(tcp_case) + with pytest.raises(planner().PlanError, match="^HTTP_STARTUP_GUARD_REQUIRED$"): + if entry == "build": + if explicit: + tcp_case["startup_policy"] = HTTP_POLICY + else: + del tcp_case["startup_policy"] + planner().build_plan(tcp_case) + else: + kwargs = {"startup_policy": HTTP_POLICY} if explicit else {} + planner().review_candidate(tcp_case["before"], candidate, tcp_case["expected"], **kwargs) + + +def test_explicit_http_policy_is_identical_to_default(case): + default_candidate, default_review = planner().build_plan(case) + case["startup_policy"] = HTTP_POLICY + assert planner().build_plan(case) == (default_candidate, default_review) + assert planner().review_candidate(case["before"], default_candidate, case["expected"], + startup_policy=HTTP_POLICY) == default_review + assert "startup_assurance" not in default_review + + +@pytest.mark.parametrize("field,value", [ + ("project", "other-project"), ("region", "other-region"), ("service", "other-service"), + ("baseline_revision", TCP_SERVICE + "-00781-other"), + ("baseline_revision", TCP_SERVICE + "-00715-old"), +]) +@pytest.mark.parametrize("entry", ["build", "review"]) +def test_tcp_rejects_coherently_rebound_unknown_baselines(tcp_case, field, value, entry): + expected = tcp_case["expected"] + previous_service = expected["service"] + expected[field] = value + if field == "service": + expected["baseline_revision"] = value + "-00781-lel" + expected["candidate_revision"] = value + "-00782-review" + for side in ("before", "current"): + payload = tcp_case[side] + for part in ("service", "revision"): + payload[part]["identity"].update({key: expected[key] for key in ("project", "region", "service")}) + payload["service"]["metadata"]["name"] = expected["service"] + payload["revision"]["identity"]["revision"] = expected["baseline_revision"] + payload["revision"]["metadata"]["name"] = expected["baseline_revision"] + for row in payload["service"]["traffic"]: + row["revisionName"] = row["revisionName"].replace(previous_service, expected["service"]) + payload["service"]["traffic"][0]["revisionName"] = expected["baseline_revision"] + with pytest.raises(planner().PlanError, match="^TCP_BASELINE_IDENTITY_MISMATCH$"): + if entry == "build": planner().build_plan(tcp_case) + else: tcp_review(tcp_case, tcp_candidate(tcp_case)) + + +@pytest.mark.parametrize("side", ["before", "current"]) +@pytest.mark.parametrize("image", [OLD_IMAGE, TCP_IMAGE.replace("charlesschwabplatform/", "other/"), + TCP_IMAGE[:-1] + "0"]) +def test_tcp_old_image_binding_is_exact(tcp_case, side, image): + main_container(tcp_case[side])["image"] = image + stop(tcp_case, "TCP_BASELINE_IMAGE_MISMATCH") + + +@pytest.mark.parametrize("side", ["before", "current", "candidate"]) +@pytest.mark.parametrize("path", [("revision", "provenance", "commit"), + ("revision", "metadata", "labels", "commit-sha")]) +def test_tcp_source_declaration_must_match_on_every_snapshot(tcp_case, side, path): + candidate = tcp_candidate(tcp_case) + set_path(candidate if side == "candidate" else tcp_case[side], path, "d" * 40) + with pytest.raises(planner().PlanError, match="^TCP_BASELINE_SOURCE_MISMATCH$"): + if side == "candidate": tcp_review(tcp_case, candidate) + else: planner().build_plan(tcp_case) + + +TCP_BAD_PROBES = [ + (None, "TCP_BASELINE_PROBE_MISMATCH"), ({}, "INVALID_PROBE"), + *[({key: copy.deepcopy(value) for key, value in TCP_PROBE.items() if key != missing}, + "INVALID_PROBE" if missing == "tcpSocket" else "TCP_BASELINE_PROBE_MISMATCH") + for missing in TCP_PROBE], + (TCP_PROBE | {"initialDelaySeconds": 0}, "TCP_BASELINE_PROBE_MISMATCH"), + (TCP_PROBE | {"successThreshold": 1}, "TCP_BASELINE_PROBE_MISMATCH"), + (TCP_PROBE | {"unknown": SENTINEL}, "UNKNOWN_FIELD"), + *[(TCP_PROBE | {field: value}, "TCP_BASELINE_PROBE_MISMATCH") for field, value in + [("failureThreshold", 2), ("periodSeconds", 239), ("timeoutSeconds", 239), + ("tcpSocket", {"port": 8081}), ("tcpSocket", {"port": 1}), ("tcpSocket", {"port": 65535})]], + (TCP_PROBE | {"timeoutSeconds": 0}, "INVALID_PROBE"), + (TCP_PROBE | {"tcpSocket": {"port": True}}, "INVALID_STRUCTURE"), + (TCP_PROBE | {"tcpSocket": {"port": "8080"}}, "INVALID_STRUCTURE"), + (TCP_PROBE | {"tcpSocket": {}}, "MISSING_FIELD"), + (TCP_PROBE | {"tcpSocket": {"port": 8080, "unknown": SENTINEL}}, "UNKNOWN_FIELD"), + (TCP_PROBE | {"grpc": {"port": 8080}}, "INVALID_PROBE"), + (TCP_PROBE | {"httpGet": {"path": "/health", "port": 8080}}, "INVALID_PROBE"), + ({"grpc": {"port": 8080}}, "TCP_BASELINE_PROBE_MISMATCH"), + ({"httpGet": {"path": "/health", "port": 8080}}, "TCP_BASELINE_PROBE_MISMATCH"), +] + + +@pytest.mark.parametrize("side", ["before", "current", "candidate"]) +@pytest.mark.parametrize("probe,reason", TCP_BAD_PROBES) +def test_tcp_complete_probe_is_never_defaulted_or_repaired(tcp_case, side, probe, reason): + candidate = tcp_candidate(tcp_case) + payload = candidate if side == "candidate" else tcp_case[side] + if probe is None: del main_container(payload)["startupProbe"] + else: main_container(payload)["startupProbe"] = copy.deepcopy(probe) + with pytest.raises(planner().PlanError, match="^" + reason + "$"): + if side == "candidate": tcp_review(tcp_case, candidate) + else: planner().build_plan(tcp_case) + + +@pytest.mark.parametrize("side", ["current", "candidate"]) +@pytest.mark.parametrize("path,value", PROTECTED_CHANGES) +def test_tcp_protected_context_cannot_drift(tcp_case, side, path, value): + candidate = tcp_candidate(tcp_case) + set_path(candidate if side == "candidate" else tcp_case[side], path, value) + reason = "CONCURRENT_SNAPSHOT_CHANGED" if side == "current" else "PROTECTED_CONFIGURATION_CHANGED" + if path == ("revision", "provenance", "commit"): reason = "TCP_BASELINE_SOURCE_MISMATCH" + if "startupProbe" in path: reason = "TCP_BASELINE_PROBE_MISMATCH" + with pytest.raises(planner().PlanError, match="^" + reason + "$"): + if side == "candidate": tcp_review(tcp_case, candidate) + else: planner().build_plan(tcp_case) + + +@pytest.mark.parametrize("side", ["current", "candidate"]) +@pytest.mark.parametrize("mutation", ["traffic-order", "traffic-tag", "traffic-split", "env-order", + "container-count", "mount-order", "volume-order", "liveness"]) +def test_tcp_preserves_order_routing_and_other_probes(tcp_case, side, mutation): + candidate = tcp_candidate(tcp_case) + payload = candidate if side == "candidate" else tcp_case[side] + container = main_container(payload) + if mutation == "traffic-order": payload["service"]["traffic"].reverse() + elif mutation == "traffic-tag": payload["service"]["traffic"][-1]["tag"] = "other-tag" + elif mutation == "traffic-split": + payload["service"]["traffic"][0]["percent"] = 80 + payload["service"]["traffic"][1]["percent"] = 20 + elif mutation == "env-order": container["env"].reverse() + elif mutation == "container-count": payload["revision"]["runtime"]["containers"].pop() + elif mutation == "mount-order": container["volumeMounts"].reverse() + elif mutation == "volume-order": payload["revision"]["runtime"]["volumes"].reverse() + elif mutation == "liveness": container["livenessProbe"]["periodSeconds"] = 11 + reason = "CONCURRENT_SNAPSHOT_CHANGED" if side == "current" else "PROTECTED_CONFIGURATION_CHANGED" + with pytest.raises(planner().PlanError, match="^" + reason + "$"): + if side == "candidate": tcp_review(tcp_case, candidate) + else: planner().build_plan(tcp_case) + + +@pytest.mark.parametrize("mutation,reason", [ + ("alias", "INDEPENDENT_SNAPSHOTS_REQUIRED"), ("service-uid", "OBJECT_RECREATED"), + ("revision-uid", "OBJECT_RECREATED"), ("before-rv", "RESOURCE_VERSION_MISMATCH"), + ("current-rv", "RESOURCE_VERSION_MISMATCH"), ("expected-rv", "RESOURCE_VERSION_MISMATCH"), + ("missing-rv", "MISSING_FIELD"), ("baseline-zero", "BASELINE_NOT_SERVING"), + ("candidate-zero", "CANDIDATE_ROUTE_PRESENT"), ("candidate-tag", "CANDIDATE_ROUTE_PRESENT"), + ("duplicate-tag", "AMBIGUOUS_TRAFFIC"), ("missing-port", "TCP_BASELINE_PROBE_MISMATCH"), +]) +def test_tcp_existing_snapshot_and_traffic_guards_remain(tcp_case, mutation, reason): + if mutation == "alias": tcp_case["current"] = tcp_case["before"] + elif mutation.endswith("-uid"): + tcp_case["current"][mutation[:-4]]["metadata"]["uid"] = "replacement-uid" + elif mutation in {"before-rv", "current-rv"}: + tcp_case[mutation[:-3]]["service"]["metadata"]["resourceVersion"] = "other-rv" + elif mutation == "expected-rv": tcp_case["expected"]["resource_version"] = "other-rv" + elif mutation == "missing-rv": del tcp_case["before"]["service"]["metadata"]["resourceVersion"] + elif mutation == "missing-port": main_container(tcp_case["before"])["ports"] = [] + else: + for side in ("before", "current"): + rows = tcp_case[side]["service"]["traffic"] + if mutation == "baseline-zero": rows[0]["percent"], rows[1]["percent"] = 0, 100 + elif mutation == "duplicate-tag": rows[1]["tag"] = rows[0]["tag"] + else: + row = {"revisionName": tcp_case["expected"]["candidate_revision"], "percent": 0} + if mutation == "candidate-tag": row["tag"] = "forbidden-tag" + rows.append(row) + stop(tcp_case, reason) + + +@pytest.mark.parametrize("side", ["before", "current", "candidate"]) +@pytest.mark.parametrize("path", [("service", "identity"), ("service", "metadata"), + ("revision", "provenance"), ("revision", "runtime"), + ("revision", "runtime", "network"), + ("revision", "runtime", "containers", 0, "resources")]) +def test_tcp_context_unknown_fields_still_stop(tcp_case, side, path): + candidate = tcp_candidate(tcp_case) + payload = candidate if side == "candidate" else tcp_case[side] + for key in path: payload = payload[key] + payload["unknown"] = SENTINEL + with pytest.raises(planner().PlanError, match="^UNKNOWN_FIELD$"): + if side == "candidate": tcp_review(tcp_case, candidate) + else: planner().build_plan(tcp_case) + + +@pytest.mark.parametrize("side", ["before", "current", "candidate"]) +def test_tcp_unknown_literal_precedes_literal_access_or_copy(tcp_case, side): + candidate = tcp_candidate(tcp_case) + for payload in (tcp_case["before"], tcp_case["current"], candidate): + entries = main_container(payload)["env"] + entries[0] = LiteralTrap(entries[0]) + main_container(candidate if side == "candidate" else tcp_case[side])["env"].append( + LiteralTrap(name="UNKNOWN_LITERAL", value=SENTINEL)) + with pytest.raises(planner().PlanError, match="^UNKNOWN_ENV_LITERAL$"): + if side == "candidate": tcp_review(tcp_case, candidate) + else: planner().build_plan(tcp_case) + + +def test_tcp_desired_template_is_never_accessed(tcp_case): + candidate, review = planner().build_plan(DesiredTrap(tcp_case)) + assert main_container(candidate)["env"][0]["value"] == RAW_TARGET + assert review["excluded_desired_template"] == "NOT_ACCESSED" + + +def test_tcp_pure_api_reads_no_environment_files_or_full_a_preflight(tcp_case, monkeypatch): + module = planner() + calls = [] + + def denied(*args, **kwargs): + calls.append("impure-call") + raise AssertionError("Pure TCP API accessed external state") + + class EnvironmentTrap(dict): + __getitem__ = get = __iter__ = __contains__ = denied + + with monkeypatch.context() as scoped: + scoped.setattr(builtins, "open", denied) + scoped.setattr(io, "open", denied) + scoped.setattr(os, "getenv", denied) + scoped.setattr(os, "environ", EnvironmentTrap()) + for name in ("_preflight_local", "_run_json", "_dependency_projection", "_runtime_values"): + scoped.setattr(module.a, name, denied) + candidate, review = module.build_plan(tcp_case) + assert tcp_review(tcp_case, candidate) == review + assert calls == [] + + +def test_tcp_review_rechecks_old_binding_and_candidate_image(tcp_case): + candidate = tcp_candidate(tcp_case) + main_container(tcp_case["before"])["image"] = OLD_IMAGE + with pytest.raises(planner().PlanError, match="^TCP_BASELINE_IMAGE_MISMATCH$"): + tcp_review(tcp_case, candidate) + main_container(tcp_case["before"])["image"] = TCP_IMAGE + main_container(candidate)["image"] = OLD_IMAGE + with pytest.raises(planner().PlanError, match="^CANDIDATE_IMAGE_MISMATCH$"): + tcp_review(tcp_case, candidate) + + +def test_tcp_real_disk_cli_preserves_guard_and_redacted_assurance(tmp_path, tcp_case): + result = cli(tmp_path, tcp_case) + assert result.returncode == 0, result.stderr + assert result.stderr == "LOCAL_GUARD_EVENTS=[]\n" + review = json.loads(result.stdout) + assert review == planner().build_plan(tcp_case)[1] + assert result.stdout == json.dumps(review, sort_keys=True, separators=(",", ":")) + "\n" + for forbidden in (SENTINEL, RAW_TARGET, "fictional-env-secret", "fictional-volume-secret", TCP_IMAGE, + TCP_SOURCE, TCP_BASE, "fictional-service-uid", "opaque/RV+781==", '"containers"', + '"metadata"', '"traffic"', "runtime_risk_limits", "apiVersion", "kind: Service"): + assert forbidden not in result.stdout + result.stderr + + +@pytest.mark.parametrize("failure,reason", [ + ("policy", "INVALID_STARTUP_POLICY"), ("probe", "TCP_BASELINE_PROBE_MISMATCH"), + ("image", "TCP_BASELINE_IMAGE_MISMATCH"), ("source", "TCP_BASELINE_SOURCE_MISMATCH"), + ("uid", "OBJECT_RECREATED"), ("rv", "RESOURCE_VERSION_MISMATCH"), + ("unknown-literal", "UNKNOWN_ENV_LITERAL"), ("unknown-field", "UNKNOWN_FIELD"), +]) +def test_tcp_real_disk_cli_failures_are_fixed_and_redacted(tmp_path, tcp_case, failure, reason): + if failure == "policy": tcp_case["startup_policy"] = SENTINEL + elif failure == "probe": main_container(tcp_case["before"])["startupProbe"]["periodSeconds"] = 239 + elif failure == "image": main_container(tcp_case["before"])["image"] = OLD_IMAGE + elif failure == "source": tcp_case["current"]["revision"]["provenance"]["commit"] = "d" * 40 + elif failure == "uid": tcp_case["current"]["service"]["metadata"]["uid"] = SENTINEL + elif failure == "rv": tcp_case["expected"]["resource_version"] = SENTINEL + elif failure == "unknown-literal": main_container(tcp_case["current"])["env"].append({"name": "UNKNOWN", "value": SENTINEL}) + elif failure == "unknown-field": tcp_case["before"]["service"]["secret_field"] = SENTINEL + result = cli(tmp_path, tcp_case) + assert result.returncode == 1 + assert result.stdout == "" + assert result.stderr == "Review stopped: " + reason + "\nLOCAL_GUARD_EVENTS=[]\n" + assert SENTINEL not in result.stderr diff --git a/tests/test_verify_cloud_run_no_traffic_deploy.py b/tests/test_verify_cloud_run_no_traffic_deploy.py index de6c4b2..4287813 100644 --- a/tests/test_verify_cloud_run_no_traffic_deploy.py +++ b/tests/test_verify_cloud_run_no_traffic_deploy.py @@ -755,3 +755,39 @@ def safe_import(name, *args, **kwargs): assert not fake_log.exists() assert before_inputs == snapshot_files(tmp_path) assert non_target == {name: hashlib.sha256((ROOT / name).read_bytes()).hexdigest() for name in non_target} + + +@pytest.mark.parametrize("handler", ["tcpSocket", "grpc"]) +def test_structure_inventory_does_not_select_a_startup_policy(case, handler): + for payload in case[1:]: + payload["runtime"]["containers"][0]["startupProbe"] = {handler: {"port": 8080}} + trap_all_literals(case) + for payload in case[1:]: + check._structure_inventory(payload) + with pytest.raises(check.PreflightError, match="^HTTP_STARTUP_GUARD_REQUIRED$"): + check._inventory(payload) + + +@pytest.mark.parametrize("handler", ["tcpSocket", "grpc"]) +@pytest.mark.parametrize("exact_identity", [False, True]) +def test_default_preflight_rejects_non_http_even_with_exact_00781_identity(case, handler, exact_identity): + for payload in case[1:]: + payload["runtime"]["containers"][0]["startupProbe"] = { + "failureThreshold": 1, "periodSeconds": 240, handler: {"port": 8080}, "timeoutSeconds": 240, + } + if exact_identity: + payload["identity"].update(project="charlesschwabquant", region="us-central1", + service="charles-schwab-quant-service", + revision="charles-schwab-quant-service-00781-lel") + trap_all_literals(case) + with pytest.raises(check.PreflightError, match="^HTTP_STARTUP_GUARD_REQUIRED$"): + check._preflight_local(*case) + + +def test_default_http_failure_keeps_original_precedence_over_volume_semantics(case): + for payload in case[1:]: + payload["runtime"]["containers"][0]["startupProbe"] = {"tcpSocket": {"port": 8080}} + payload["runtime"]["volumes"] = [{"name": "cache", "emptyDir": {}}, {"name": "cache", "emptyDir": {}}] + trap_all_literals(case) + with pytest.raises(check.PreflightError, match="^HTTP_STARTUP_GUARD_REQUIRED$"): + check._preflight_local(*case)