From 076920d65b863727ad40fa2b6933a01cbab37bef Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 29 Aug 2026 02:27:50 +0800 Subject: [PATCH] ci: gate LongBridge rollouts on target admission Co-Authored-By: Codex --- .github/workflows/sync-cloud-run-env.yml | 13 +- ...erify_deployed_runtime_target_admission.py | 152 ++++++++++++++++++ .../test_deployed_runtime_target_admission.py | 70 ++++++++ 3 files changed, 233 insertions(+), 2 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 9fe317f..f59cace 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -477,13 +477,13 @@ jobs: fi - name: Set up Python for strategy requirement resolution - if: steps.config.outputs.env_sync_enabled == 'true' + if: steps.config.outputs.enabled == 'true' uses: actions/setup-python@v6 with: python-version: "3.12" - name: Install strategy status dependencies - if: steps.config.outputs.env_sync_enabled == 'true' + if: steps.config.outputs.enabled == 'true' run: | set -euo pipefail python -m pip install --upgrade pip uv @@ -592,6 +592,15 @@ jobs: project_id: ${{ env.GCP_PROJECT_ID }} version: ">= 416.0.0" + - name: Verify deployed runtime target admission before traffic shift + if: steps.config.outputs.deploy_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: Download constraints if: steps.config.outputs.deploy_enabled == 'true' run: curl -sL https://raw.githubusercontent.com/QuantStrategyLab/QuantPlatformKit/main/constraints.txt -o constraints.txt diff --git a/scripts/verify_deployed_runtime_target_admission.py b/scripts/verify_deployed_runtime_target_admission.py new file mode 100644 index 0000000..91de948 --- /dev/null +++ b/scripts/verify_deployed_runtime_target_admission.py @@ -0,0 +1,152 @@ +#!/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 LONGBRIDGE_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(entry.get("name") or "").strip(): str(entry.get("value") or "").strip() + for entry in entries + if isinstance(entry, Mapping) and str(entry.get("name") or "").strip() and "value" in entry + } + + +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") + target_service = str(target.get("service_name") or "").strip() + if target_service 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=LONGBRIDGE_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("LONGBRIDGE_DRY_RUN_ONLY") + if configured_dry_run is not None and _parse_bool( + configured_dry_run, field="LONGBRIDGE_DRY_RUN_ONLY", service=service + ) != target_dry_run: + raise AdmissionError(f"{service}: LONGBRIDGE_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") + + enabled = _parse_bool(env.get("RUNTIME_TARGET_ENABLED", "true"), field="RUNTIME_TARGET_ENABLED", service=service) + return { + "service": service, + "profile": canonical_profile, + "execution_mode": execution_mode, + "dry_run_only": target_dry_run, + "enabled": enabled, + } + + +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']}, " + f"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..2464530 --- /dev/null +++ b/tests/test_deployed_runtime_target_admission.py @@ -0,0 +1,70 @@ +import importlib.util +import json +from pathlib import Path + +import pytest + + +SCRIPT_PATH = Path(__file__).resolve().parents[1] / "scripts" / "verify_deployed_runtime_target_admission.py" +SPEC = importlib.util.spec_from_file_location("deployed_target_admission", SCRIPT_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 service_payload(*, runtime_target: dict, profile: str, dry_run_only: str = "true") -> dict: + return { + "spec": {"template": {"spec": {"containers": [{"env": [ + {"name": "RUNTIME_TARGET_JSON", "value": json.dumps(runtime_target)}, + {"name": "STRATEGY_PROFILE", "value": profile}, + {"name": "LONGBRIDGE_DRY_RUN_ONLY", "value": dry_run_only}, + {"name": "RUNTIME_TARGET_ENABLED", "value": "true"}, + ]}]}}} + } + + +def admitted_target(*, profile: str = "russell_top50_leader_rotation", dry_run_only: bool = True) -> dict: + return { + "platform_id": "longbridge", + "service_name": "paper-service", + "strategy_profile": profile, + "execution_mode": "paper" if dry_run_only else "live", + "dry_run_only": dry_run_only, + } + + +def test_verify_service_accepts_admitted_shadow_target(): + result = admission.verify_service( + service="paper-service", + service_json=service_payload(runtime_target=admitted_target(), profile="russell_top50_leader_rotation"), + ) + assert result["profile"] == "russell_top50_leader_rotation" + assert result["dry_run_only"] is True + + +def test_verify_service_accepts_paper_broker_submission_target(): + target = admitted_target(dry_run_only=False) + target["execution_mode"] = "paper" + result = admission.verify_service( + service="paper-service", + service_json=service_payload( + runtime_target=target, profile="russell_top50_leader_rotation", dry_run_only="false" + ), + ) + assert result["execution_mode"] == "paper" + assert result["dry_run_only"] is False + + +@pytest.mark.parametrize( + ("target", "profile", "message"), + [ + (admitted_target(), "different_profile", "STRATEGY_PROFILE does not match"), + ({**admitted_target(), "execution_mode": "live"}, "russell_top50_leader_rotation", "dry-run/shadow target"), + ({**admitted_target(), "strategy_profile": "retired_profile"}, "retired_profile", "not admitted"), + ], +) +def test_verify_service_rejects_target_drift(target, profile, message): + with pytest.raises(admission.AdmissionError, match=message): + admission.verify_service( + service="paper-service", service_json=service_payload(runtime_target=target, profile=profile) + )