Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion .github/workflows/sync-cloud-run-env-protected.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down
42 changes: 42 additions & 0 deletions docs/ues-binding-coverage.md
Original file line number Diff line number Diff line change
@@ -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.
112 changes: 94 additions & 18 deletions scripts/verify_image_ues_release_binding.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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:
Expand All @@ -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


Expand Down
125 changes: 124 additions & 1 deletion tests/test_verify_image_ues_release_binding.py
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand Down Expand Up @@ -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
Loading