From 10158a880d7325bd59e49ea1d367147c747ad152 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 11 Oct 2026 22:50:43 +0900 Subject: [PATCH] Report sanitized build-time UES guard coverage without changing enforcement --- .../sync-cloud-run-env-protected.yml | 8 +- docs/ues-binding-coverage.md | 42 ++++++ scripts/verify_image_ues_release_binding.py | 112 +++++++++++++--- .../test_verify_image_ues_release_binding.py | 125 +++++++++++++++++- 4 files changed, 267 insertions(+), 20 deletions(-) create mode 100644 docs/ues-binding-coverage.md diff --git a/.github/workflows/sync-cloud-run-env-protected.yml b/.github/workflows/sync-cloud-run-env-protected.yml index 3c8e8d02..88e71d6f 100644 --- a/.github/workflows/sync-cloud-run-env-protected.yml +++ b/.github/workflows/sync-cloud-run-env-protected.yml @@ -662,7 +662,7 @@ jobs: docker build --pull -t "${image}" source # RS-08 guard: block the deploy if the image's installed UES revision differs from # strategy_release / runtime-risk binding (live would reject every SOXL rebalance). - python3 scripts/verify_image_ues_release_binding.py --image "${image}" --lockfile source/uv.lock + python3 scripts/verify_image_ues_release_binding.py --image "${image}" --lockfile source/uv.lock --report-json "${RUNNER_TEMP}/ues-binding-coverage.json" docker push "${image}" image_digest="$(gcloud artifacts docker images describe "${image}" --project="${GCP_PROJECT_ID}" --format='value(image_summary.digest)')" if [[ ! "${image_digest}" =~ ^sha256:[0-9a-f]{64}$ ]]; then @@ -694,6 +694,12 @@ jobs: --quiet done + - name: Summarize build-time UES binding coverage + continue-on-error: true + if: always() && steps.config.outputs.deploy_enabled == 'true' + run: | + python3 scripts/verify_image_ues_release_binding.py --summarize-report "${RUNNER_TEMP}/ues-binding-coverage.json" >> "${GITHUB_STEP_SUMMARY}" + - name: Verify no-traffic deployment readback if: steps.config.outputs.deploy_enabled == 'true' env: diff --git a/docs/ues-binding-coverage.md b/docs/ues-binding-coverage.md new file mode 100644 index 00000000..e9dde4bf --- /dev/null +++ b/docs/ues-binding-coverage.md @@ -0,0 +1,42 @@ +# Build-time UES binding coverage + +The existing deployment guard compares the installed UES revision in the built +image against `uv.lock` and the revisions present in planned runtime targets. +It still runs after build and before push/deploy, with the same blocking rules. +This is build-time evidence, **not proof of the live deployed revision**. + +`--report-json PATH` optionally writes a schema-version-1 JSON report. The report +is an observation of the existing validation, not a new configuration source or +validation policy. Failure to write this optional report warns without changing +the guard exit code. The workflow renders the report into its job summary even +when the guard blocks deployment; a missing report is reported as unavailable. + +Statuses: +- `verified`: every resolved target had at least one expected revision, and all + present expected revisions matched the installed revision. This does not mean + both binding fields were present or that other runtime requirements passed. +- `not_covered`: at least one target had neither revision binding. Such targets + remain non-blocking under the existing rules. Mixed verified/unbound coverage + is never reported as wholly verified. +- `mismatch`: a present expected revision mismatched, or installed revision + sources disagreed. Deployment remains blocked. +- `uncomputable`: no runtime target could be resolved. Deployment remains blocked. + Target counts also use this state when installed sources disagree, so target + comparisons could not be performed. +- `error`: revision extraction or input parsing failed, including the existing + 300-second image-probe timeout. Deployment remains blocked. + +Reports have fixed fields: schema version, evidence scope, status, aggregate +counts by state, and installed revisions from the three allowlisted sources +(`image`, `lockfile`, `explicit`). Only full hexadecimal commit SHAs are retained; +other revision strings become null. Reports contain no target names, release IDs, +accounts, assets, raw runtime configuration, environment values or probe stderr. +Console validation uses ordinal target aliases and similarly sanitized values. + +`--summarize-report PATH` only renders allowlisted enum/count values from an +existing report. It does **not** execute validation, Docker, or deployment and its +exit code must not be used as a validation result. The deploy step still uses the +normal guard exit code; report generation and rendering add no enforcement. + +All tests use synthetic targets and mocked Docker subprocess results. No live +broker, Cloud Run, credentials or deployment workflows are exercised. diff --git a/scripts/verify_image_ues_release_binding.py b/scripts/verify_image_ues_release_binding.py index b2b63b46..5848e7ba 100755 --- a/scripts/verify_image_ues_release_binding.py +++ b/scripts/verify_image_ues_release_binding.py @@ -18,8 +18,8 @@ Expected revisions come from each planned target's RUNTIME_TARGET_JSON (SYNC_PLAN_JSON targets[].env), else CLOUD_RUN_SERVICE_TARGETS_JSON runtime targets, else RUNTIME_TARGET_JSON. A target without a strategy_release and -without a binding ues_revision is reported and not blocked. Prints only service -names, release ids and commit SHAs. Read-only; does not touch Cloud Run. +without a binding ues_revision is reported and not blocked. Prints only +ordinal target aliases and validated commit SHAs. Read-only; does not touch Cloud Run. """ from __future__ import annotations @@ -71,7 +71,7 @@ def revision_from_image(image: str) -> str: values = dict(line.split("=", 1) for line in result.stdout.splitlines() if "=" in line) stamp, direct = values.get("UES_STAMP", "").strip(), values.get("UES_DIRECT_URL", "").strip() if stamp and direct and stamp != direct: - raise GuardError(f"image UES stamp {stamp} != installed direct_url {direct}") + raise GuardError("image UES stamp differs from installed direct_url") revision = stamp or direct if not SHA_RE.match(revision): raise GuardError("image has no installed UES commit") @@ -123,40 +123,112 @@ def expected_revisions(runtime_target: Mapping[str, Any]) -> dict[str, str]: return expected -def check(installed: Mapping[str, str], targets: Iterable[tuple[str, Mapping[str, Any]]]) -> list[str]: - """Return failure lines; prints a non-sensitive report.""" +STATES = ("verified", "mismatch", "not_covered", "uncomputable", "error") + + +def new_report() -> dict[str, Any]: + """Bounded, allowlisted build-time evidence; never a live deployment attestation.""" + return { + "schema_version": 1, + "evidence_scope": "built_image_and_planned_targets", + "status": "uncomputable", + "target_counts": {state: 0 for state in STATES}, + "installed_revisions": {}, + } + + +def safe_revision(value: str) -> str | None: + return value if SHA_RE.fullmatch(value) else None + + +def check( + installed: Mapping[str, str], targets: Iterable[tuple[str, Mapping[str, Any]]], + *, report: dict[str, Any] | None = None, +) -> list[str]: + """Keep validation decisions unchanged; optionally collect sanitized coverage.""" + report = report if report is not None else new_report() failures: list[str] = [] values = set(installed.values()) - for label, rev in installed.items(): - print(f"installed_ues_revision[{label}]={rev}") + for label in ("lockfile", "image", "explicit"): + if label in installed: + revision = safe_revision(installed[label]) + report["installed_revisions"][label] = revision + print(f"installed_ues_revision[{label}]={revision or 'unavailable'}") + targets = list(targets) if len(values) != 1: - failures.append("installed UES revision sources disagree: " - + ", ".join(f"{k}={v}" for k, v in sorted(installed.items()))) + report["status"] = "mismatch" + report["target_counts"]["uncomputable"] = len(targets) + failures.append("installed UES revision sources disagree") return failures (revision,) = values - targets = list(targets) if not targets: failures.append("no runtime target resolved; cannot verify the strategy release binding") - for service, runtime_target in targets: + for ordinal, (_, runtime_target) in enumerate(targets, 1): + alias = f"target-{ordinal}" expected = expected_revisions(runtime_target or {}) - release_id = ((runtime_target or {}).get("strategy_release") or {}).get("release_id") if not expected: - print(f"service={service or '?'} no strategy_release/binding ues_revision: not blocked") + report["target_counts"]["not_covered"] += 1 + print(f"{alias}: not_covered (no revision binding; not blocked)") continue + state = "verified" for field, value in sorted(expected.items()): ok = value == revision - print(f"service={service or '?'} release_id={release_id} {field}={value} match={ok}") + print(f"{alias} {field}={safe_revision(value) or 'unavailable'} match={ok}") if not ok: - failures.append(f"{service or '?'}: {field}={value} != installed {revision}") + state = "mismatch" + failures.append(f"{alias}: {field} differs from installed revision") + report["target_counts"][state] += 1 + counts = report["target_counts"] + report["status"] = ( + "mismatch" if failures and targets else + "uncomputable" if not targets else + "not_covered" if counts["not_covered"] else "verified" + ) return failures +def write_report(path: Path | None, report: dict[str, Any]) -> None: + # Reporting is optional and cannot change validation or deployment decisions. + if path is not None: + try: + path.write_text(json.dumps(report, sort_keys=True) + "\n", encoding="utf-8") + except OSError: + print("::warning::UES coverage report could not be written", file=sys.stderr) + + +def summarize_report(path: Path) -> None: + """Render only validated enum/count fields, never arbitrary report strings.""" + try: + report = json.loads(path.read_text(encoding="utf-8")) + if report.get("schema_version") != 1: + raise ValueError("schema") + status = report["status"] + counts = report["target_counts"] + if status not in STATES or any( + type(counts[state]) is not int or not 0 <= counts[state] <= 1_000_000 + for state in STATES + ): + raise ValueError("invalid summary") + except (OSError, ValueError, KeyError, TypeError, AttributeError): + print("UES build-time binding coverage: report unavailable.") + return + print(f"UES build-time binding coverage: {status}.") + print("Planned targets: " + ", ".join(f"{state}={counts[state]}" for state in STATES) + ".") + print("Build-time evidence only; this does not verify a live deployed revision.") + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--image") parser.add_argument("--lockfile", type=Path) parser.add_argument("--installed-revision") + parser.add_argument("--report-json", type=Path, help="Optional sanitized coverage output; does not change validation") + parser.add_argument("--summarize-report", type=Path, help="Render existing coverage output without running validation") args = parser.parse_args(argv) + if args.summarize_report: + summarize_report(args.summarize_report) + return 0 + report = new_report() try: installed: dict[str, str] = {} if args.lockfile: @@ -167,17 +239,21 @@ def main(argv: list[str] | None = None) -> int: installed["explicit"] = args.installed_revision.strip() if not installed: raise GuardError("give --image, --lockfile or --installed-revision") - failures = check(installed, runtime_targets_from_env(os.environ)) + failures = check(installed, runtime_targets_from_env(os.environ), report=report) except (GuardError, OSError, ValueError, subprocess.TimeoutExpired) as exc: - print(f"::error::UES release binding guard could not verify: {exc}", file=sys.stderr) + report["status"] = "error" + # Exceptions may contain image names, paths, probe stderr, or runtime JSON. + print(f"::error::UES release binding guard could not verify ({type(exc).__name__})", file=sys.stderr) + write_report(args.report_json, report) return 1 + write_report(args.report_json, report) if failures: for line in failures: print(f"::error::UES release binding mismatch: {line}", file=sys.stderr) print("Deploy blocked: update strategy_release.strategy_revision and runtime_risk_limits.binding." "ues_revision together with the UES pin (or keep the pin) before deploying.", file=sys.stderr) return 1 - print("UES release binding guard: OK") + print(f"UES release binding guard: not blocked; coverage={report['status']}") return 0 diff --git a/tests/test_verify_image_ues_release_binding.py b/tests/test_verify_image_ues_release_binding.py index 7eef7ba1..9081363a 100644 --- a/tests/test_verify_image_ues_release_binding.py +++ b/tests/test_verify_image_ues_release_binding.py @@ -37,7 +37,7 @@ def test_any_mismatch_blocks(release, binding): env = {"SYNC_PLAN_JSON": _plan(_target("soxl", release=release, binding=binding))} installed = NEW if (release, binding) == (OLD, OLD) else OLD failures = guard.check({"image": installed}, guard.runtime_targets_from_env(env)) - assert failures and all("soxl" in line for line in failures) + assert failures and all("target-1" in line for line in failures) def test_rebuild_from_newer_pin_without_release_update_is_blocked(): @@ -83,3 +83,126 @@ def test_deploy_workflow_runs_guard_before_push_and_deploy(): guard_at = workflow.index("scripts/verify_image_ues_release_binding.py --image") assert workflow.index("docker build --pull") < guard_at < workflow.index('docker push "${image}"') assert guard_at < workflow.index("gcloud run deploy") + + +@pytest.mark.parametrize("stamp,direct,expected", [ + (OLD, OLD, OLD), (OLD, "", OLD), ("", OLD, OLD), + (OLD, NEW, None), ("", "", None), ("private-account", "", None), +]) +def test_actual_image_probe_branch(monkeypatch, stamp, direct, expected): + import subprocess + + def run(command, **kwargs): + assert command[:7] == ["docker", "run", "--rm", "--network", "none", "--entrypoint", "python"] + assert command[7:] == ["synthetic-image", "-c", guard._IMAGE_PROBE] + assert kwargs == {"capture_output": True, "text": True, "timeout": 300, "check": False} + return subprocess.CompletedProcess(command, 0, f"UES_STAMP={stamp}\nUES_DIRECT_URL={direct}\n", "secret-stderr") + + monkeypatch.setattr(guard.subprocess, "run", run) + if expected: + assert guard.revision_from_image("synthetic-image") == expected + else: + with pytest.raises(guard.GuardError): + guard.revision_from_image("synthetic-image") + + +@pytest.mark.parametrize("failure", ["nonzero", "timeout", "missing_docker"]) +def test_image_failures_keep_exit_code_and_redact(monkeypatch, tmp_path, capsys, failure): + import subprocess + + secret = "DO-NOT-PRINT-account-service-runtime-stderr" + + def run(command, **kwargs): + if failure == "timeout": + raise subprocess.TimeoutExpired(command, 300, output=secret, stderr=secret) + if failure == "missing_docker": + raise OSError(secret) + return subprocess.CompletedProcess(command, 1, secret, secret) + + monkeypatch.setattr(guard.subprocess, "run", run) + path = tmp_path / "report.json" + assert guard.main(["--image", secret, "--report-json", str(path)]) == 1 + report = json.loads(path.read_text()) + assert report["status"] == "error" + captured = capsys.readouterr() + assert secret not in captured.out + captured.err + path.read_text() + + +@pytest.mark.parametrize("targets,status,code,counts", [ + ([_target("secret-service", release=OLD)], "verified", 0, (1, 0, 0)), + ([_target("secret-service")], "not_covered", 0, (0, 0, 1)), + ([_target("secret-service", release=OLD), _target("secret-unbound")], "not_covered", 0, (1, 0, 1)), + ([_target("secret-service", release=NEW)], "mismatch", 1, (0, 1, 0)), +]) +def test_structured_coverage_is_sanitized(monkeypatch, tmp_path, capsys, targets, status, code, counts): + monkeypatch.setenv("SYNC_PLAN_JSON", _plan(*targets)) + path = tmp_path / "report.json" + assert guard.main(["--installed-revision", OLD, "--report-json", str(path)]) == code + report = json.loads(path.read_text()) + assert report["status"] == status + assert tuple(report["target_counts"][key] for key in ("verified", "mismatch", "not_covered")) == counts + assert report["evidence_scope"] == "built_image_and_planned_targets" + assert set(report) == {"schema_version", "evidence_scope", "status", "target_counts", "installed_revisions"} + output = capsys.readouterr() + assert "secret-" not in path.read_text() + output.out + output.err + assert "guard: OK" not in output.out + + +def test_no_target_and_disagreeing_sources_reports(): + report = guard.new_report() + assert guard.check({"image": OLD}, [], report=report) + assert report["status"] == "uncomputable" + report = guard.new_report() + assert guard.check({"image": OLD, "lockfile": NEW}, [("private", {})], report=report) + assert report["status"] == "mismatch" + assert report["target_counts"]["uncomputable"] == 1 + + +def test_untrusted_values_are_not_printed(monkeypatch, tmp_path, capsys): + private = "account-secret-env-assets" + monkeypatch.setenv("SYNC_PLAN_JSON", _plan(_target(private, release=private))) + path = tmp_path / "report.json" + assert guard.main(["--installed-revision", OLD, "--report-json", str(path)]) == 1 + output = capsys.readouterr() + assert private not in output.out + output.err + path.read_text() + monkeypatch.setenv("SYNC_PLAN_JSON", "{invalid-" + private) + assert guard.main(["--installed-revision", OLD, "--report-json", str(path)]) == 1 + output = capsys.readouterr() + assert private not in output.out + output.err + path.read_text() + + +def test_report_write_failure_does_not_change_validation(monkeypatch, tmp_path): + monkeypatch.setenv("SYNC_PLAN_JSON", _plan(_target("private", release=OLD))) + assert guard.main(["--installed-revision", OLD, "--report-json", str(tmp_path)]) == 0 + assert guard.main(["--installed-revision", NEW, "--report-json", str(tmp_path)]) == 1 + + +def test_summary_reads_report_without_validation(tmp_path, monkeypatch, capsys): + path = tmp_path / "report.json" + report = guard.new_report() + report["status"] = "not_covered" + report["target_counts"]["verified"] = 1 + report["target_counts"]["not_covered"] = 1 + report["unexpected"] = "private-account" + path.write_text(json.dumps(report)) + monkeypatch.setattr(guard, "check", lambda *a, **kw: pytest.fail("summary must not validate")) + assert guard.main(["--summarize-report", str(path)]) == 0 + output = capsys.readouterr().out + assert "not_covered=1" in output + assert "does not verify a live deployed revision" in output + assert "private-account" not in output + report["target_counts"]["verified"] = "private-account" + path.write_text(json.dumps(report)) + assert guard.main(["--summarize-report", str(path)]) == 0 + assert capsys.readouterr().out == "UES build-time binding coverage: report unavailable.\n" + + +def test_workflow_summary_only_consumes_report(): + workflow = Path(".github/workflows/sync-cloud-run-env-protected.yml").read_text() + assert '--report-json "${RUNNER_TEMP}/ues-binding-coverage.json"' in workflow + summary = workflow.split("- name: Summarize build-time UES binding coverage", 1)[1].split("- name:", 1)[0] + assert "if: always() && steps.config.outputs.deploy_enabled == 'true'" in summary + assert '--summarize-report "${RUNNER_TEMP}/ues-binding-coverage.json"' in summary + assert "continue-on-error: true" in summary + assert "GITHUB_STEP_SUMMARY" in summary + assert "SYNC_PLAN_JSON" not in summary