From 31945aafc0c06c33bbfa8e5c4b2ad6896c79b860 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 29 Aug 2026 01:39:50 +0800 Subject: [PATCH] feat: gate Cloud Run rollouts on target admission Co-Authored-By: Codex --- .github/workflows/sync-cloud-run-env.yml | 28 ++- docs/ibkr_runtime_rollout.md | 6 +- ...erify_deployed_runtime_target_admission.py | 218 ++++++++++++++++++ .../test_deployed_runtime_target_admission.py | 108 +++++++++ 4 files changed, 351 insertions(+), 9 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 89b5890..d9e1de0 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -68,6 +68,7 @@ jobs: ENABLE_GITHUB_ENV_SYNC: ${{ vars.ENABLE_GITHUB_ENV_SYNC }} ENABLE_MAIN_PUSH_CLOUD_RUN_AUTOMATION: ${{ vars.ENABLE_MAIN_PUSH_CLOUD_RUN_AUTOMATION }} QSL_ENABLE_CLOUD_RUN_AUTOMATION: ${{ vars.QSL_ENABLE_CLOUD_RUN_AUTOMATION }} + CLOUD_RUN_CLEANUP_ENABLED: ${{ vars.CLOUD_RUN_CLEANUP_ENABLED }} WORKFLOW_TARGET: ${{ inputs.target || 'configured' }} INPUT_CLOUD_RUN_REGION: ${{ inputs.cloud_run_region }} INPUT_CLOUD_RUN_SERVICE: ${{ inputs.cloud_run_service }} @@ -283,21 +284,21 @@ jobs: echo "CLOUD_RUN_ENV_SYNC_WAIT_FOR_COMMIT=false" >> "$GITHUB_ENV" fi - - name: Set up Python for strategy requirement resolution - if: steps.config.outputs.env_sync_enabled == 'true' + - name: Set up Python for runtime target admission + 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 uv sync --frozen --no-dev - - name: Resolve Cloud Run sync targets + - name: Resolve admissible Cloud Run targets id: strategy_requirements - if: steps.config.outputs.env_sync_enabled == 'true' + if: steps.config.outputs.enabled == 'true' run: | set -euo pipefail sync_plan_json="$(uv run --no-sync python scripts/build_cloud_run_env_sync_plan.py --json)" @@ -351,6 +352,7 @@ jobs: fi - name: Validate deploy inputs + if: steps.config.outputs.deploy_enabled == 'true' run: | set -euo pipefail @@ -427,7 +429,18 @@ jobs: exit 1 fi + - name: Verify deployed runtime target admission before traffic shift + if: steps.config.outputs.deploy_enabled == 'true' + env: + SYNC_PLAN_JSON: ${{ steps.strategy_requirements.outputs.sync_plan_json }} + run: | + set -euo pipefail + uv run --no-sync python scripts/verify_deployed_runtime_target_admission.py \ + --project="${GCP_PROJECT_ID}" \ + --region="${CLOUD_RUN_REGION}" + - name: Build, push, and deploy Cloud Run image + if: steps.config.outputs.deploy_enabled == 'true' run: | set -euo pipefail @@ -492,7 +505,7 @@ jobs: done - name: Wait for Cloud Run deployment of current commit - if: steps.config.outputs.env_sync_enabled == 'true' + if: steps.config.outputs.deploy_enabled == 'true' && steps.config.outputs.env_sync_enabled == 'true' env: SYNC_PLAN_JSON: ${{ steps.strategy_requirements.outputs.sync_plan_json }} run: | @@ -1267,7 +1280,7 @@ jobs: done - name: Prune old Cloud Run revisions - if: steps.config.outputs.enabled == 'true' + if: steps.config.outputs.deploy_enabled == 'true' && env.CLOUD_RUN_CLEANUP_ENABLED == 'true' env: SYNC_PLAN_JSON: ${{ steps.strategy_requirements.outputs.sync_plan_json }} run: | @@ -1356,6 +1369,7 @@ jobs: done - name: Clean up old Cloud Run images + if: steps.config.outputs.deploy_enabled == 'true' && env.CLOUD_RUN_CLEANUP_ENABLED == 'true' run: | set -euo pipefail diff --git a/docs/ibkr_runtime_rollout.md b/docs/ibkr_runtime_rollout.md index c426182..351a7a1 100644 --- a/docs/ibkr_runtime_rollout.md +++ b/docs/ibkr_runtime_rollout.md @@ -285,8 +285,10 @@ gcloud storage buckets add-iam-policy-binding "gs://run-sources-${PROJECT_ID}-${ - `IB_ACCOUNT_GROUP_CONFIG_SECRET_NAME=ibkr-account-groups` - 其他服务级变量按上面补齐 -5. **触发 env sync** - - push 到 `main`，或手动跑一次同等的 `gcloud run services update` +5. **通过受保护的部署 workflow 触发同步或镜像发布** + - 标准路径是 `Deploy Cloud Run` workflow；它会在流量切换前验证已部署目标与待发布策略均仍在准入目录。 + - 不要把临时的 `gcloud run services update --image ...` 当作常规发布方式：它会绕过策略准入、运行身份与 Paper/Shadow/Live 语义校验。 + - 历史 revision 与镜像默认保留；只有显式设置 `CLOUD_RUN_CLEANUP_ENABLED=true` 才允许自动清理，以保证有可回滚版本。 6. **检查 Cloud Run 当前 env** diff --git a/scripts/verify_deployed_runtime_target_admission.py b/scripts/verify_deployed_runtime_target_admission.py new file mode 100644 index 0000000..bc56992 --- /dev/null +++ b/scripts/verify_deployed_runtime_target_admission.py @@ -0,0 +1,218 @@ +#!/usr/bin/env python3 +"""Fail closed before a Cloud Run image rollout reaches an unadmitted target. + +The deployment plan validates the desired configuration. This checker protects +the other half of the boundary: a service which is already configured with a +retired or inconsistent runtime target must not receive a new image and become +an accidental compatibility migration. + +Only non-sensitive target identity fields are read from Cloud Run. The script +does not read Secret Manager values and never mutates a service or scheduler. +""" + +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +from collections.abc import Mapping, Sequence +from typing import Any + +from strategy_registry import IBKR_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 != 0: + detail = (result.stderr or result.stdout or "").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") + env_entries = containers[0].get("env", []) + if not isinstance(env_entries, list): + raise AdmissionError("Cloud Run container environment is malformed") + env: dict[str, str] = {} + for entry in env_entries: + if not isinstance(entry, Mapping): + continue + name = str(entry.get("name") or "").strip() + if name and "value" in entry: + env[name] = str(entry.get("value") or "").strip() + return env + + +def _parse_bool(value: object, *, field: str, service: str) -> bool: + if isinstance(value, bool): + return value + normalized = str(value or "").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 _runtime_target(env: Mapping[str, str], *, service: str) -> Mapping[str, Any]: + raw = env.get("RUNTIME_TARGET_JSON") or env.get("QSL_RUNTIME_TARGET_JSON") + if not raw: + raise AdmissionError(f"{service}: RUNTIME_TARGET_JSON is required for image admission") + try: + target = json.loads(raw) + 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") + return target + + +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) + target = _runtime_target(env, service=service) + 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=IBKR_PLATFORM) + except (TypeError, ValueError) as exc: + raise AdmissionError(f"{service}: strategy profile is not admitted") from exc + canonical_profile = definition.profile + configured_profile = str(env.get("STRATEGY_PROFILE") or "").strip() + if configured_profile != 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("IBKR_DRY_RUN_ONLY") + if configured_dry_run is not None and _parse_bool( + configured_dry_run, field="IBKR_DRY_RUN_ONLY", service=service + ) != target_dry_run: + raise AdmissionError( + f"{service}: IBKR_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 _services_from_plan(raw_plan: str) -> list[str]: + try: + plan = json.loads(raw_plan) + except json.JSONDecodeError as exc: + raise AdmissionError("SYNC_PLAN_JSON is invalid JSON") from exc + targets = plan.get("targets") if isinstance(plan, Mapping) else None + if not isinstance(targets, list): + raise AdmissionError("SYNC_PLAN_JSON.targets must be a list") + services: list[str] = [] + for target in targets: + if not isinstance(target, Mapping): + raise AdmissionError("SYNC_PLAN_JSON targets must be objects") + service = str(target.get("service_name") or "").strip() + if not service: + raise AdmissionError("SYNC_PLAN_JSON target is missing service_name") + services.append(service) + return list(dict.fromkeys(services)) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--project", required=True) + parser.add_argument("--region", required=True) + parser.add_argument("--service", action="append", default=[]) + args = parser.parse_args() + + services = [str(service).strip() for service in args.service if str(service).strip()] + if not services: + raw_plan = (os.environ.get("SYNC_PLAN_JSON") or "").strip() + if not raw_plan: + parser.error("--service or SYNC_PLAN_JSON is required") + services = _services_from_plan(raw_plan) + + try: + for service in services: + result = verify_service( + service=service, + service_json=_describe_service( + service=service, + project=args.project, + region=args.region, + ), + ) + 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']}" + ) + except AdmissionError as exc: + print(f"Deployed runtime target admission failed: {exc}", file=sys.stderr) + return 1 + 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..8812193 --- /dev/null +++ b/tests/test_deployed_runtime_target_admission.py @@ -0,0 +1,108 @@ +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": "IBKR_DRY_RUN_ONLY", "value": dry_run_only}, + {"name": "RUNTIME_TARGET_ENABLED", "value": "true"}, + ] + } + ] + } + } + } + } + + +def admitted_target(*, profile: str = "tqqq_growth_income", dry_run_only: bool = True) -> dict: + return { + "platform_id": "ibkr", + "service_name": "shadow-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="shadow-service", + service_json=service_payload(runtime_target=admitted_target(), profile="tqqq_growth_income"), + ) + + assert result == { + "service": "shadow-service", + "profile": "tqqq_growth_income", + "execution_mode": "paper", + "dry_run_only": True, + "enabled": True, + } + + +def test_verify_service_accepts_admitted_paper_broker_target(): + target = admitted_target(dry_run_only=False) + target["execution_mode"] = "paper" + + result = admission.verify_service( + service="shadow-service", + service_json=service_payload( + runtime_target=target, + profile="tqqq_growth_income", + dry_run_only="false", + ), + ) + + assert result["execution_mode"] == "paper" + assert result["dry_run_only"] is False + + +def test_verify_service_rejects_profile_drift(): + with pytest.raises(admission.AdmissionError, match="STRATEGY_PROFILE does not match"): + admission.verify_service( + service="shadow-service", + service_json=service_payload(runtime_target=admitted_target(), profile="soxl_soxx_trend_income"), + ) + + +def test_verify_service_rejects_shadow_declared_as_live(): + target = admitted_target() + target["execution_mode"] = "live" + + with pytest.raises(admission.AdmissionError, match="dry-run/shadow target"): + admission.verify_service( + service="shadow-service", + service_json=service_payload(runtime_target=target, profile="tqqq_growth_income"), + ) + + +def test_verify_service_rejects_retired_profile(): + target = admitted_target(profile="retired_profile") + + with pytest.raises(admission.AdmissionError, match="not admitted"): + admission.verify_service( + service="shadow-service", + service_json=service_payload(runtime_target=target, profile="retired_profile"), + )