From 3fa4af69d074690f4258b4e6839d2b5fe9f3883a Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 29 Aug 2026 02:30:45 +0800 Subject: [PATCH] ci: gate Firstrade rollouts on target admission Co-Authored-By: Codex --- .github/workflows/sync-cloud-run-env.yml | 34 +++-- ...erify_deployed_runtime_target_admission.py | 116 ++++++++++++++++++ .../test_deployed_runtime_target_admission.py | 47 +++++++ 3 files changed, 185 insertions(+), 12 deletions(-) create mode 100644 scripts/verify_deployed_runtime_target_admission.py create mode 100644 tests/test_deployed_runtime_target_admission.py diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml index 81ddee8..0c60ec3 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -215,6 +215,28 @@ jobs: project_id: ${{ env.GCP_PROJECT_ID }} version: ">= 416.0.0" + - name: Set up Python for strategy requirement resolution + if: steps.deploy_config.outputs.enabled == 'true' + uses: actions/setup-python@v6 + with: + python-version: "3.12" + + - name: Install strategy status dependencies + if: steps.deploy_config.outputs.enabled == 'true' + run: | + set -euo pipefail + python -m pip install --upgrade pip uv + uv sync --frozen --no-dev + + - name: Verify deployed runtime target admission before traffic shift + if: steps.deploy_config.outputs.enabled == 'true' + run: | + set -euo pipefail + uv run --no-sync python scripts/verify_deployed_runtime_target_admission.py \ + --project="${GCP_PROJECT_ID}" \ + --region="${CLOUD_RUN_REGION}" \ + --service="${CLOUD_RUN_SERVICE}" + - name: Build, push, and deploy Cloud Run image if: steps.deploy_config.outputs.enabled == 'true' @@ -257,18 +279,6 @@ jobs: echo "enabled=true" >> "$GITHUB_OUTPUT" - - name: Set up Python for strategy requirement resolution - if: steps.env_sync_config.outputs.enabled == 'true' - uses: actions/setup-python@v6 - with: - python-version: "3.12" - - - name: Install strategy status dependencies - if: steps.env_sync_config.outputs.enabled == 'true' - run: | - set -euo pipefail - python -m pip install --upgrade pip uv - uv sync --frozen --no-dev - name: Resolve Cloud Run sync targets id: strategy_requirements if: steps.env_sync_config.outputs.enabled == 'true' diff --git a/scripts/verify_deployed_runtime_target_admission.py b/scripts/verify_deployed_runtime_target_admission.py new file mode 100644 index 0000000..dee51a0 --- /dev/null +++ b/scripts/verify_deployed_runtime_target_admission.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +"""Fail closed before a Cloud Run rollout reaches an unadmitted target. + +Only non-sensitive target identity fields are read from Cloud Run. This checker +never reads Secret Manager values and never mutates a service. +""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +from collections.abc import Mapping, Sequence +from typing import Any + +from strategy_registry import FIRSTRADE_PLATFORM, resolve_strategy_definition + + +class AdmissionError(ValueError): + """A deployed runtime target is not safe to receive a new image.""" + + +def _run(command: Sequence[str]) -> str: + result = subprocess.run(command, text=True, capture_output=True, check=False) + if result.returncode: + detail = (result.stderr or result.stdout).strip() + raise AdmissionError(detail or f"Command failed: {' '.join(command)}") + return result.stdout + + +def _describe_service(*, service: str, project: str, region: str) -> Mapping[str, Any]: + payload = _run(["gcloud", "run", "services", "describe", service, f"--project={project}", f"--region={region}", "--format=json"]) + loaded = json.loads(payload) + if not isinstance(loaded, Mapping): + raise AdmissionError(f"{service}: Cloud Run describe returned a non-object payload") + return loaded + + +def _container_env(service_json: Mapping[str, Any]) -> dict[str, str]: + containers = service_json.get("spec", {}).get("template", {}).get("spec", {}).get("containers", []) + if not isinstance(containers, list) or not containers: + raise AdmissionError("Cloud Run service has no container configuration") + entries = containers[0].get("env", []) + if not isinstance(entries, list): + raise AdmissionError("Cloud Run container environment is malformed") + return {str(item.get("name") or "").strip(): str(item.get("value") or "").strip() for item in entries if isinstance(item, Mapping) and str(item.get("name") or "").strip() and "value" in item} + + +def _parse_bool(value: object, *, field: str, service: str) -> bool: + if isinstance(value, bool): + return value + normalized = str(value).strip().lower() + if normalized in {"1", "true", "yes", "on"}: + return True + if normalized in {"0", "false", "no", "off"}: + return False + raise AdmissionError(f"{service}: {field} must be a boolean") + + +def verify_service(*, service: str, service_json: Mapping[str, Any]) -> dict[str, object]: + """Validate one deployed service without printing account or secret data.""" + + env = _container_env(service_json) + raw_target = env.get("RUNTIME_TARGET_JSON") or env.get("QSL_RUNTIME_TARGET_JSON") + if not raw_target: + raise AdmissionError(f"{service}: RUNTIME_TARGET_JSON is required for image admission") + try: + target = json.loads(raw_target) + except json.JSONDecodeError as exc: + raise AdmissionError(f"{service}: RUNTIME_TARGET_JSON is invalid JSON") from exc + if not isinstance(target, Mapping): + raise AdmissionError(f"{service}: RUNTIME_TARGET_JSON must be an object") + if (target_service := str(target.get("service_name") or "").strip()) and target_service != service: + raise AdmissionError(f"{service}: runtime target service_name does not match the deployed service") + raw_profile = str(target.get("strategy_profile") or "").strip() + if not raw_profile: + raise AdmissionError(f"{service}: runtime target strategy_profile is required") + try: + definition = resolve_strategy_definition(raw_profile, platform_id=FIRSTRADE_PLATFORM) + except (TypeError, ValueError) as exc: + raise AdmissionError(f"{service}: strategy profile is not admitted") from exc + canonical_profile = definition.profile + if str(env.get("STRATEGY_PROFILE") or "").strip() != canonical_profile: + raise AdmissionError(f"{service}: STRATEGY_PROFILE does not match the admitted runtime target profile") + execution_mode = str(target.get("execution_mode") or "").strip().lower() + if execution_mode not in {"paper", "live"}: + raise AdmissionError(f"{service}: execution_mode must be paper or live") + if "dry_run_only" not in target: + raise AdmissionError(f"{service}: runtime target dry_run_only is required") + target_dry_run = _parse_bool(target["dry_run_only"], field="runtime target dry_run_only", service=service) + configured_dry_run = env.get("FIRSTRADE_DRY_RUN_ONLY") + if configured_dry_run is not None and _parse_bool(configured_dry_run, field="FIRSTRADE_DRY_RUN_ONLY", service=service) != target_dry_run: + raise AdmissionError(f"{service}: FIRSTRADE_DRY_RUN_ONLY does not match runtime target dry_run_only") + if target_dry_run and execution_mode != "paper": + raise AdmissionError(f"{service}: a dry-run/shadow target must declare execution_mode=paper") + return {"service": service, "profile": canonical_profile, "execution_mode": execution_mode, "dry_run_only": target_dry_run, "enabled": _parse_bool(env.get("RUNTIME_TARGET_ENABLED", "true"), field="RUNTIME_TARGET_ENABLED", service=service)} + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--project", required=True) + parser.add_argument("--region", required=True) + parser.add_argument("--service", required=True) + args = parser.parse_args() + try: + result = verify_service(service=args.service, service_json=_describe_service(service=args.service, project=args.project, region=args.region)) + except AdmissionError as exc: + print(f"Deployed runtime target admission failed: {exc}", file=sys.stderr) + return 1 + print("Verified deployed runtime target admission: " f"service={result['service']}, profile={result['profile']}, " f"mode={result['execution_mode']}, dry_run_only={result['dry_run_only']}, enabled={result['enabled']}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_deployed_runtime_target_admission.py b/tests/test_deployed_runtime_target_admission.py new file mode 100644 index 0000000..796054f --- /dev/null +++ b/tests/test_deployed_runtime_target_admission.py @@ -0,0 +1,47 @@ +import importlib.util +import json +from pathlib import Path + +import pytest + + +path = Path(__file__).resolve().parents[1] / "scripts" / "verify_deployed_runtime_target_admission.py" +spec = importlib.util.spec_from_file_location("deployed_target_admission", path) +assert spec is not None and spec.loader is not None +admission = importlib.util.module_from_spec(spec) +spec.loader.exec_module(admission) + + +def payload(target, profile, dry_run="true"): + return {"spec": {"template": {"spec": {"containers": [{"env": [ + {"name": "RUNTIME_TARGET_JSON", "value": json.dumps(target)}, + {"name": "STRATEGY_PROFILE", "value": profile}, + {"name": "FIRSTRADE_DRY_RUN_ONLY", "value": dry_run}, + {"name": "RUNTIME_TARGET_ENABLED", "value": "true"}, + ]}]}}}} + + +def target(profile="ibit_smart_dca", dry_run=True): + return {"platform_id": "firstrade", "service_name": "live-service", "strategy_profile": profile, "execution_mode": "paper" if dry_run else "live", "dry_run_only": dry_run} + + +def test_admitted_shadow_target_passes(): + assert admission.verify_service(service="live-service", service_json=payload(target(), "ibit_smart_dca"))["profile"] == "ibit_smart_dca" + + +def test_paper_broker_submission_target_passes(): + configured = target(dry_run=False) | {"execution_mode": "paper"} + assert admission.verify_service(service="live-service", service_json=payload(configured, "ibit_smart_dca", "false"))["dry_run_only"] is False + + +@pytest.mark.parametrize( + ("configured", "profile", "message"), + [ + (target(), "different_profile", "STRATEGY_PROFILE does not match"), + (target() | {"execution_mode": "live"}, "ibit_smart_dca", "dry-run/shadow target"), + (target("retired_profile"), "retired_profile", "not admitted"), + ], +) +def test_target_drift_fails_closed(configured, profile, message): + with pytest.raises(admission.AdmissionError, match=message): + admission.verify_service(service="live-service", service_json=payload(configured, profile))