From 16f616301b5eb63912f5de64021b051ab3f6766a Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 11 Nov 2026 19:37:31 +0800 Subject: [PATCH 1/2] fix(observation): enable the SG snapshot on its serving image --- .github/workflows/sync-cloud-run-env.yml | 34 +++++ config/hk_snapshot_verification.workflow.yaml | 9 +- docs/account-snapshot-history.md | 4 + docs/read-only-account-observation.zh-CN.md | 2 +- scripts/enable_sg_serving_snapshot.py | 118 ++++++++++++++++++ tests/test_enable_sg_serving_snapshot.py | 68 ++++++++++ tests/test_runtime_monitor_workflows.py | 7 ++ 7 files changed, 237 insertions(+), 5 deletions(-) create mode 100644 scripts/enable_sg_serving_snapshot.py create mode 100644 tests/test_enable_sg_serving_snapshot.py diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml index 280cc6e1..315c08dc 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -13,6 +13,7 @@ on: - legacy - image-only-no-traffic - image-only-notification-overlay + - sg-serving-snapshot approved_ref: description: "Exact workflow branch/tag name approved for image-only staging." required: false @@ -2307,3 +2308,36 @@ jobs: --location="${scheduler_location}" \ --quiet fi + + + sg-serving-snapshot: + name: Enable only the SG read-only snapshot + if: >- + github.event_name == 'workflow_dispatch' && github.ref == 'refs/heads/main' && + github.run_attempt == 1 && inputs.deployment_mode == 'sg-serving-snapshot' && + inputs.target == 'SG' && inputs.approved_ref == 'main' && + inputs.account_snapshot_enabled == 'true' && + inputs.source_commit == '2bace9b851495dc25f2b4772bcf602b09eb9e61e' && + inputs.account_history_recording_enabled == '' && inputs.account_history_gcs_prefix == '' && + inputs.account_history_target_id == '' && inputs.account_history_expected_scope == '' + runs-on: ubuntu-latest + timeout-minutes: 12 + environment: longbridge-sg + permissions: + contents: read + id-token: write + env: + GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT: ${{ vars.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }} + CLOUD_RUN_SERVICE: ${{ secrets.CLOUD_RUN_SERVICE }} + CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }} + steps: + - uses: actions/checkout@v6 + with: + persist-credentials: false + - uses: google-github-actions/auth@v3 + with: + workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }} + service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }} + - uses: google-github-actions/setup-gcloud@v3 + - name: Stage and verify the serving image with the snapshot flag + run: python3 scripts/enable_sg_serving_snapshot.py diff --git a/config/hk_snapshot_verification.workflow.yaml b/config/hk_snapshot_verification.workflow.yaml index c78cbe04..c9a77b19 100644 --- a/config/hk_snapshot_verification.workflow.yaml +++ b/config/hk_snapshot_verification.workflow.yaml @@ -6,6 +6,7 @@ main: assign: - origin: ${sys.get_env("HK_SNAPSHOT_ORIGIN")} - expected_binding: ${sys.get_env("HK_SNAPSHOT_EXPECTED_BINDING")} + - expected_scope: ${sys.get_env("SNAPSHOT_VERIFICATION_SCOPE", "HK")} - requested_at: ${sys.now()} - archive_bucket: ${sys.get_env("HK_DIGEST_BUCKET", "")} - archive_prefix: ${sys.get_env("HK_DIGEST_PREFIX", "")} @@ -14,7 +15,7 @@ main: - strategy_profile: ${sys.get_env("HK_DIGEST_STRATEGY_PROFILE", "")} - check_configuration: switch: - - condition: '${not text.match_regex(origin, "^https://[a-z0-9-]+[.][a-z0-9-]+[.]run[.]app$") or not text.match_regex(expected_binding, "^[a-f0-9]{64}$")}' + - condition: '${not text.match_regex(origin, "^https://[a-z0-9-]+[.][a-z0-9-]+[.]run[.]app$") or not text.match_regex(expected_binding, "^[a-f0-9]{64}$") or (expected_scope != "HK" and expected_scope != "SG")}' raise: hk_snapshot_configuration_invalid - read_snapshot: try: @@ -39,7 +40,7 @@ main: - now: ${sys.now()} - check_binding_and_fields: switch: - - condition: '${response.code != 200 or get_type(snapshot) != "map" or map.get(snapshot, "schema_version") != "longbridge_account_snapshot.v1" or map.get(snapshot, "status") != "partial" or map.get(snapshot, "account_scope") != "HK"}' + - condition: '${response.code != 200 or get_type(snapshot) != "map" or map.get(snapshot, "schema_version") != "longbridge_account_snapshot.v1" or map.get(snapshot, "status") != "partial" or map.get(snapshot, "account_scope") != expected_scope}' raise: hk_snapshot_contract_invalid - condition: '${map.get(snapshot, "no_order") != true or map.get(snapshot, "live_authority_granted") != false or map.get(snapshot, "cash_complete") != true or map.get(snapshot, "positions_complete") != true}' raise: hk_snapshot_contract_invalid @@ -59,7 +60,7 @@ main: switch: - condition: ${archive_bucket == "" and archive_prefix == "" and account_uid == "" and target_id == "" and strategy_profile == ""} next: receipt - - condition: '${not text.match_regex(archive_bucket, "^[a-z0-9][a-z0-9._-]{2,62}$") or not text.match_regex(archive_prefix, "^[A-Za-z0-9_/-]+$") or account_uid == "" or target_id == "" or strategy_profile == ""}' + - condition: '${expected_scope != "HK" or not text.match_regex(archive_bucket, "^[a-z0-9][a-z0-9._-]{2,62}$") or not text.match_regex(archive_prefix, "^[A-Za-z0-9_/-]+$") or account_uid == "" or target_id == "" or strategy_profile == ""}' raise: hk_digest_archive_configuration_invalid - start_projection: assign: @@ -149,7 +150,7 @@ main: position_count: ${len(snapshot.positions)} - receipt: return: - account_scope: HK + account_scope: ${expected_scope} source_binding_verified: true read_only_snapshot_verified: true observed_at: ${snapshot.observed_finished_at} diff --git a/docs/account-snapshot-history.md b/docs/account-snapshot-history.md index 47576b49..e7d35f6e 100644 --- a/docs/account-snapshot-history.md +++ b/docs/account-snapshot-history.md @@ -197,6 +197,10 @@ producer 为每次观察写入 `{prefix}/{target_id}/{source_binding_id}/{observ 这份记录不是 TWR，不是收益率，也不授予 live 权限。 +## SG 快照开关的单独发布 + +`Deploy Cloud Run` 的 `sg-serving-snapshot` 模式只接受已核验的 SG serving inventory 版本。流程先用云端预检核对现役镜像、停用的交易开关和当前固定流量，然后暂存同一镜像并单独开启快照。新版本的配置和身份核对通过后，才将流量切到该版本。SG 的日报采集还需要独立验证真实快照及私有归档。 + ## HK 独立只读观察模式（默认关闭，仅手动） 新增的 `ACCOUNT_HISTORY_OBSERVATION_MODE=independent_get` 是与旧归档路线分开的 HK-only 模式。缺省仍为 `scheduler_archive`，旧模式继续保留一次 Scheduler 触发、受限 GCS listing、固定 generation、原字节发布及原有 SG/HK 受限恢复保证。选择独立模式后，失败不会回退到 Scheduler、`/probe`、`/run`、`/dry-run` 或策略 canary。 diff --git a/docs/read-only-account-observation.zh-CN.md b/docs/read-only-account-observation.zh-CN.md index e0edd55d..8b615513 100644 --- a/docs/read-only-account-observation.zh-CN.md +++ b/docs/read-only-account-observation.zh-CN.md @@ -28,7 +28,7 @@ `independent_get` 适用于可由当前执行器访问的已鉴权 Cloud Run 服务。内部 ingress 返回固定原因 `internal_observer_required`，在调用券商或发送账户事实前停止；不得为采集开放 ingress。内部服务应复用受支持的内部执行入口，另行验证只读采集、私有归档与接收闭环。此文档不表示该内部闭环已经完成。 -HK 内网连通性可用 `config/hk_snapshot_verification.workflow.yaml` 单独验证。它在同一 GCP 项目通过 OIDC 调用固定的 `GET /account-snapshot`，从受保护的部署配置读取服务地址和预期绑定。验证只返回时间、字段数量和绑定结果。Cloud Workflows 的调用日志须关闭，运行身份单独配置，调用者不能传入地址、方法或请求参数。未配置归档时只做手动验证。HK 生产归档使用受保护的 `HK_DIGEST_BUCKET`、`HK_DIGEST_PREFIX`、`HK_DIGEST_ACCOUNT_UID`、`HK_DIGEST_TARGET_ID` 和 `HK_DIGEST_STRATEGY_PROFILE`，每天在新加坡时间 06:00 前采集，按 06:00 分界写入唯一的账户候选对象。只保留有来源绑定的净资产和股票持仓，不从快照推断订单或成交。归档权限仅允许在 HK 候选路径创建对象，候选已存在时保留原对象。 +HK 内网连通性可用 `config/hk_snapshot_verification.workflow.yaml` 单独验证。它在同一 GCP 项目通过 OIDC 调用固定的 `GET /account-snapshot`，从受保护的部署配置读取服务地址和预期绑定。验证只返回时间、字段数量和绑定结果。Cloud Workflows 的调用日志须关闭，运行身份单独配置，调用者不能传入地址、方法或请求参数。未配置归档时只做手动验证。SG 的部署可在只读快照开启后将 `SNAPSHOT_VERIFICATION_SCOPE` 设为 `SG`，使用其独立身份和服务来源绑定。SG 验证模式不写入 HK 归档。HK 生产归档使用受保护的 `HK_DIGEST_BUCKET`、`HK_DIGEST_PREFIX`、`HK_DIGEST_ACCOUNT_UID`、`HK_DIGEST_TARGET_ID` 和 `HK_DIGEST_STRATEGY_PROFILE`，每天在新加坡时间 06:00 前采集，按 06:00 分界写入唯一的账户候选对象。只保留有来源绑定的净资产和股票持仓，不从快照推断订单或成交。归档权限仅允许在 HK 候选路径创建对象，候选已存在时保留原对象。 单次操作最多一次快照 GET 和一次事实 POST，不自动重试未知结果。投影保留券商实际币种、金额、观察时间及来源；账户来源不匹配时拒收，不补零、不改旧数据时间。交易启停、风险预算和通知不由此观察器控制。 diff --git a/scripts/enable_sg_serving_snapshot.py b/scripts/enable_sg_serving_snapshot.py new file mode 100644 index 00000000..7373a655 --- /dev/null +++ b/scripts/enable_sg_serving_snapshot.py @@ -0,0 +1,118 @@ +"""Enable the SG account snapshot on the verified serving image only.""" +from __future__ import annotations + +import copy +import json +import os +import re +import subprocess +import time +from urllib.parse import urlencode +from urllib.request import Request, urlopen + +SOURCE = "2bace9b851495dc25f2b4772bcf602b09eb9e61e" +FLAG = "LONGBRIDGE_ACCOUNT_SNAPSHOT_ENABLED" +API = "https://run.googleapis.com/v2/" + + +def _require(value): + if not value: + raise ValueError("sg_snapshot_release_readback_mismatch") + + +def _environment(container): + return {item["name"]: item for item in container.get("env", [])} + + +def _revision_path(service, revision): + if "/revisions/" in revision: + _require(revision.startswith(service + "/revisions/")) + return revision + _require(re.fullmatch(r"[a-z0-9-]+", revision)) + return service + "/revisions/" + revision + + +def enable_snapshot(request, service): + baseline = request(service) + _require(baseline["name"] == service and baseline["ingress"] == "INGRESS_TRAFFIC_INTERNAL_ONLY") + current_traffic, = baseline["traffic"] + active, = [r for r in baseline["trafficStatuses"] if r.get("percent", 0) > 0] + _require(current_traffic["type"] == "TRAFFIC_TARGET_ALLOCATION_TYPE_REVISION") + _require(current_traffic["percent"] == active["percent"] == 100) + _require(not current_traffic.get("tag") and not active.get("tag")) + old_path = _revision_path(service, active["revision"]) + _require(old_path == _revision_path(service, current_traffic["revision"])) + old = request(old_path) + _require(old.get("labels", {}).get("commit-sha") == SOURCE) + template = copy.deepcopy(baseline["template"]) + container, = template["containers"] + serving_container, = old["containers"] + env = _environment(container) + _require(container["image"] == serving_container["image"] and env == _environment(serving_container)) + _require(re.search(r"@sha256:[a-f0-9]{64}$", container["image"])) + _require(env["RUNTIME_TARGET_ENABLED"].get("value") == "false") + target = json.loads(env["RUNTIME_TARGET_JSON"]["value"]) + _require(target["platform_id"] == "longbridge" and target["account_scope"].upper() == "SG") + if FLAG in env: + _require(env[FLAG].get("value") == "false" and "valueSource" not in env[FLAG]) + env[FLAG]["value"] = "true" + else: + container["env"].append({"name": FLAG, "value": "true"}) + template.pop("revision", None) + body = {"name": service, "etag": baseline["etag"], "template": template} + request(service + "?" + urlencode({"updateMask": "template", "validateOnly": "true"}), "PATCH", body) + _require(request(service)["etag"] == baseline["etag"]) + staged = request(service + "?updateMask=template", "PATCH", body) + _wait(request, staged) + + live = request(service) + _require(live["traffic"] == baseline["traffic"] and live["trafficStatuses"] == baseline["trafficStatuses"]) + _require(live["ingress"] == baseline["ingress"] and live["template"]["serviceAccount"] == template["serviceAccount"]) + new_path = _revision_path(service, live["latestReadyRevision"]) + _require(new_path != old_path and new_path == _revision_path(service, live["latestCreatedRevision"])) + ready = request(new_path) + new_container, = ready["containers"] + _require(new_container["image"] == container["image"] and _environment(new_container) == _environment(container)) + _require(ready.get("labels", {}).get("commit-sha") == SOURCE and ready["serviceAccount"] == old["serviceAccount"]) + route = {**current_traffic, "revision": new_path.rsplit("/", 1)[-1], "percent": 100} + moved = request(service + "?updateMask=traffic", "PATCH", {"name": service, "etag": live["etag"], "traffic": [route]}) + _wait(request, moved) + + final = request(service) + current, = [r for r in final["trafficStatuses"] if r.get("percent", 0) > 0] + _require(current["percent"] == 100 and _revision_path(service, current["revision"]) == new_path) + _require(final["template"] == live["template"] and final["ingress"] == baseline["ingress"]) + + +def _wait(request, operation): + deadline = time.monotonic() + 240 + while not operation.get("done") and time.monotonic() < deadline: + time.sleep(4) + operation = request(operation["name"]) + _require(operation.get("done") and not operation.get("error")) + + +def main(): + project, region, service = (os.environ[k] for k in ("GCP_PROJECT_ID", "CLOUD_RUN_REGION", "CLOUD_RUN_SERVICE")) + _require(all(re.fullmatch(r"[a-z0-9-]+", value) for value in (project, region, service))) + token = subprocess.check_output(["gcloud", "auth", "print-access-token"], text=True, + stderr=subprocess.DEVNULL, timeout=30).strip() + + def request(path, method="GET", payload=None): + headers = {"Authorization": "Bearer " + token, "Content-Type": "application/json"} + body = json.dumps(payload).encode() if payload is not None else None + with urlopen(Request(API + path, headers=headers, method=method, data=body), timeout=30) as response: + result = json.load(response) + _require(not result.get("error")) + return result + + enable_snapshot(request, f"projects/{project}/locations/{region}/services/{service}") + + +if __name__ == "__main__": + try: + main() + except Exception as error: + print(json.dumps({"sg_snapshot_release_verified": False, "error_type": type(error).__name__})) + raise SystemExit(1) from None + print('{"sg_snapshot_release_verified":true,"serving_image_preserved":true,"trading_target_still_disabled":true}') diff --git a/tests/test_enable_sg_serving_snapshot.py b/tests/test_enable_sg_serving_snapshot.py new file mode 100644 index 00000000..5a003265 --- /dev/null +++ b/tests/test_enable_sg_serving_snapshot.py @@ -0,0 +1,68 @@ +import copy +import json + +import pytest + +from scripts.enable_sg_serving_snapshot import FLAG, SOURCE, enable_snapshot + +SERVICE = "projects/example/locations/example/services/sg" +IMAGE = "example.invalid/source@sha256:" + "1" * 64 + + +def fixture(*, enabled=False, drift=False): + old_env = [ + {"name": "RUNTIME_TARGET_ENABLED", "value": "true" if enabled else "false"}, + {"name": "RUNTIME_TARGET_JSON", "value": json.dumps({"platform_id": "longbridge", "account_scope": "SG"})}, + ] + template = {"containers": [{"image": IMAGE, "env": old_env}], "serviceAccount": "example"} + baseline = {"name": SERVICE, "ingress": "INGRESS_TRAFFIC_INTERNAL_ONLY", + "template": template, "etag": "before", + "traffic": [{"type": "TRAFFIC_TARGET_ALLOCATION_TYPE_REVISION", "revision": "sg-old", "percent": 100}], + "trafficStatuses": [{"revision": "sg-old", "percent": 100}]} + old = {"containers": template["containers"], "labels": {"commit-sha": SOURCE}, "serviceAccount": "example"} + writes, live = [], copy.deepcopy(baseline) + + def request(path, method="GET", body=None): + nonlocal live + if method == "PATCH": + writes.append((path, copy.deepcopy(body))) + if "validateOnly" in path: + return {"done": True} + if path.endswith("updateMask=template"): + assert body["etag"] == "before" + live = {**live, "template": body["template"], "etag": "staged", + "latestReadyRevision": "sg-new", "latestCreatedRevision": "sg-new"} + else: + assert body["etag"] == "staged" + live = {**live, "traffic": body["traffic"], "trafficStatuses": [{"revision": "sg-new", "percent": 100}]} + return {"done": True} + if path == SERVICE: + return copy.deepcopy(live) + if path.endswith("sg-old"): + return copy.deepcopy(old) + assert path.endswith("sg-new") + data = {**old, "containers": copy.deepcopy(live["template"]["containers"])} + if drift: + data["containers"][0]["env"][0]["value"] = "true" + return data + + return request, writes + + +def test_reuses_the_serving_image_and_preserves_other_environment_values(): + request, writes = fixture() + enable_snapshot(request, SERVICE) + assert len(writes) == 3 + staged = writes[1][1]["template"]["containers"][0] + assert staged["image"] == IMAGE + assert [entry for entry in staged["env"] if entry["name"] == FLAG] == [{"name": FLAG, "value": "true"}] + assert staged["env"][0]["value"] == "false" + assert writes[2][1]["traffic"][0]["revision"] == "sg-new" + + +@pytest.mark.parametrize(("options", "writes_before_stop"), [({"enabled": True}, 0), ({"drift": True}, 2)]) +def test_rejects_enabled_trading_or_staged_configuration_drift(options, writes_before_stop): + request, writes = fixture(**options) + with pytest.raises(ValueError, match="readback_mismatch"): + enable_snapshot(request, SERVICE) + assert len(writes) == writes_before_stop diff --git a/tests/test_runtime_monitor_workflows.py b/tests/test_runtime_monitor_workflows.py index b54ff0fa..88fae9a1 100644 --- a/tests/test_runtime_monitor_workflows.py +++ b/tests/test_runtime_monitor_workflows.py @@ -215,3 +215,10 @@ def test_internal_hk_archive_is_one_account_and_create_only() -> None: assert "ifGenerationMatch: 0" in source assert "actually_ran: false" in source assert "order_count:" not in source and "fill_count:" not in source + + +def test_internal_sg_verification_uses_sg_identity_without_hk_archive() -> None: + source = (ROOT / "config/hk_snapshot_verification.workflow.yaml").read_text() + assert 'sys.get_env("SNAPSHOT_VERIFICATION_SCOPE", "HK")' in source + assert 'map.get(snapshot, "account_scope") != expected_scope' in source + assert 'expected_scope != "HK" or not text.match_regex(archive_bucket' in source From e2591ea3899fbd9884fb701376c4dd05f7845b34 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 11 Nov 2026 19:43:34 +0800 Subject: [PATCH 2/2] fix(observation): verify the full SG serving configuration --- .github/workflows/sync-cloud-run-env.yml | 4 ++- docs/account-snapshot-history.md | 2 +- scripts/enable_sg_serving_snapshot.py | 32 +++++++++++++++++------- tests/test_enable_sg_serving_snapshot.py | 29 +++++++++++++++------ 4 files changed, 49 insertions(+), 18 deletions(-) diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml index 315c08dc..597c1fbf 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -2317,7 +2317,7 @@ jobs: github.run_attempt == 1 && inputs.deployment_mode == 'sg-serving-snapshot' && inputs.target == 'SG' && inputs.approved_ref == 'main' && inputs.account_snapshot_enabled == 'true' && - inputs.source_commit == '2bace9b851495dc25f2b4772bcf602b09eb9e61e' && + inputs.source_commit == '' && inputs.account_history_recording_enabled == '' && inputs.account_history_gcs_prefix == '' && inputs.account_history_target_id == '' && inputs.account_history_expected_scope == '' runs-on: ubuntu-latest @@ -2340,4 +2340,6 @@ jobs: service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }} - uses: google-github-actions/setup-gcloud@v3 - name: Stage and verify the serving image with the snapshot flag + env: + SG_SNAPSHOT_APPROVED_SERVING_SOURCE: ${{ secrets.SG_SNAPSHOT_APPROVED_SERVING_SOURCE }} run: python3 scripts/enable_sg_serving_snapshot.py diff --git a/docs/account-snapshot-history.md b/docs/account-snapshot-history.md index e7d35f6e..eb4eaab3 100644 --- a/docs/account-snapshot-history.md +++ b/docs/account-snapshot-history.md @@ -199,7 +199,7 @@ producer 为每次观察写入 `{prefix}/{target_id}/{source_binding_id}/{observ ## SG 快照开关的单独发布 -`Deploy Cloud Run` 的 `sg-serving-snapshot` 模式只接受已核验的 SG serving inventory 版本。流程先用云端预检核对现役镜像、停用的交易开关和当前固定流量，然后暂存同一镜像并单独开启快照。新版本的配置和身份核对通过后，才将流量切到该版本。SG 的日报采集还需要独立验证真实快照及私有归档。 +`Deploy Cloud Run` 的 `sg-serving-snapshot` 模式从受保护的 SG Environment 读取已核准的现役来源。流程先用云端预检核对完整现役配置、停用的交易开关和当前固定流量，然后暂存同一镜像并单独开启快照。新版本的配置和身份核对通过后，才将流量切到该版本。SG 的日报采集还需要独立验证真实快照及私有归档。 ## HK 独立只读观察模式（默认关闭，仅手动） diff --git a/scripts/enable_sg_serving_snapshot.py b/scripts/enable_sg_serving_snapshot.py index 7373a655..afdde90b 100644 --- a/scripts/enable_sg_serving_snapshot.py +++ b/scripts/enable_sg_serving_snapshot.py @@ -10,9 +10,14 @@ from urllib.parse import urlencode from urllib.request import Request, urlopen -SOURCE = "2bace9b851495dc25f2b4772bcf602b09eb9e61e" FLAG = "LONGBRIDGE_ACCOUNT_SNAPSHOT_ENABLED" API = "https://run.googleapis.com/v2/" +REVISION_METADATA = { + "name", "uid", "generation", "createTime", "updateTime", "deleteTime", "expireTime", + "launchStage", "service", "reconciling", "conditions", "observedGeneration", "logUri", + "satisfiesPzs", "scalingStatus", "containerStatuses", "creator", "etag", +} +TEMPLATE_METADATA = {"revision", "client", "clientVersion"} def _require(value): @@ -32,7 +37,18 @@ def _revision_path(service, revision): return service + "/revisions/" + revision -def enable_snapshot(request, service): +def _require_matching_revision(template, revision): + wanted = {k: v for k, v in template.items() if k not in TEMPLATE_METADATA} + actual = {k: v for k, v in revision.items() if k not in REVISION_METADATA | TEMPLATE_METADATA} + expected_container, = wanted["containers"] + actual_container, = actual["containers"] + if "name" not in expected_container: + actual["containers"] = [{k: v for k, v in actual_container.items() if k != "name"}] + _require(wanted == actual) + + +def enable_snapshot(request, service, approved_source): + _require(re.fullmatch(r"[a-f0-9]{40}", approved_source)) baseline = request(service) _require(baseline["name"] == service and baseline["ingress"] == "INGRESS_TRAFFIC_INTERNAL_ONLY") current_traffic, = baseline["traffic"] @@ -43,12 +59,11 @@ def enable_snapshot(request, service): old_path = _revision_path(service, active["revision"]) _require(old_path == _revision_path(service, current_traffic["revision"])) old = request(old_path) - _require(old.get("labels", {}).get("commit-sha") == SOURCE) + _require(old.get("labels", {}).get("commit-sha") == approved_source) template = copy.deepcopy(baseline["template"]) container, = template["containers"] - serving_container, = old["containers"] + _require_matching_revision(template, old) env = _environment(container) - _require(container["image"] == serving_container["image"] and env == _environment(serving_container)) _require(re.search(r"@sha256:[a-f0-9]{64}$", container["image"])) _require(env["RUNTIME_TARGET_ENABLED"].get("value") == "false") target = json.loads(env["RUNTIME_TARGET_JSON"]["value"]) @@ -71,9 +86,7 @@ def enable_snapshot(request, service): new_path = _revision_path(service, live["latestReadyRevision"]) _require(new_path != old_path and new_path == _revision_path(service, live["latestCreatedRevision"])) ready = request(new_path) - new_container, = ready["containers"] - _require(new_container["image"] == container["image"] and _environment(new_container) == _environment(container)) - _require(ready.get("labels", {}).get("commit-sha") == SOURCE and ready["serviceAccount"] == old["serviceAccount"]) + _require_matching_revision(template, ready) route = {**current_traffic, "revision": new_path.rsplit("/", 1)[-1], "percent": 100} moved = request(service + "?updateMask=traffic", "PATCH", {"name": service, "etag": live["etag"], "traffic": [route]}) _wait(request, moved) @@ -106,7 +119,8 @@ def request(path, method="GET", payload=None): _require(not result.get("error")) return result - enable_snapshot(request, f"projects/{project}/locations/{region}/services/{service}") + approved_source = os.environ["SG_SNAPSHOT_APPROVED_SERVING_SOURCE"] + enable_snapshot(request, f"projects/{project}/locations/{region}/services/{service}", approved_source) if __name__ == "__main__": diff --git a/tests/test_enable_sg_serving_snapshot.py b/tests/test_enable_sg_serving_snapshot.py index 5a003265..dc3ec4a3 100644 --- a/tests/test_enable_sg_serving_snapshot.py +++ b/tests/test_enable_sg_serving_snapshot.py @@ -3,23 +3,28 @@ import pytest -from scripts.enable_sg_serving_snapshot import FLAG, SOURCE, enable_snapshot +from scripts.enable_sg_serving_snapshot import FLAG, enable_snapshot +SOURCE = "2" * 40 SERVICE = "projects/example/locations/example/services/sg" IMAGE = "example.invalid/source@sha256:" + "1" * 64 -def fixture(*, enabled=False, drift=False): +def fixture(*, enabled=False, drift=False, stale_template=False): old_env = [ {"name": "RUNTIME_TARGET_ENABLED", "value": "true" if enabled else "false"}, {"name": "RUNTIME_TARGET_JSON", "value": json.dumps({"platform_id": "longbridge", "account_scope": "SG"})}, ] - template = {"containers": [{"image": IMAGE, "env": old_env}], "serviceAccount": "example"} + template = {"containers": [{"image": IMAGE, "env": old_env}], "serviceAccount": "example", + "labels": {"commit-sha": SOURCE}} baseline = {"name": SERVICE, "ingress": "INGRESS_TRAFFIC_INTERNAL_ONLY", "template": template, "etag": "before", "traffic": [{"type": "TRAFFIC_TARGET_ALLOCATION_TYPE_REVISION", "revision": "sg-old", "percent": 100}], "trafficStatuses": [{"revision": "sg-old", "percent": 100}]} - old = {"containers": template["containers"], "labels": {"commit-sha": SOURCE}, "serviceAccount": "example"} + old = copy.deepcopy(template) + old["containers"][0]["name"] = "generated-container-name" + if stale_template: + template["containers"][0]["command"] = ["/different-command"] writes, live = [], copy.deepcopy(baseline) def request(path, method="GET", body=None): @@ -42,6 +47,7 @@ def request(path, method="GET", body=None): return copy.deepcopy(old) assert path.endswith("sg-new") data = {**old, "containers": copy.deepcopy(live["template"]["containers"])} + data["containers"][0]["name"] = "generated-container-name" if drift: data["containers"][0]["env"][0]["value"] = "true" return data @@ -51,7 +57,7 @@ def request(path, method="GET", body=None): def test_reuses_the_serving_image_and_preserves_other_environment_values(): request, writes = fixture() - enable_snapshot(request, SERVICE) + enable_snapshot(request, SERVICE, SOURCE) assert len(writes) == 3 staged = writes[1][1]["template"]["containers"][0] assert staged["image"] == IMAGE @@ -60,9 +66,18 @@ def test_reuses_the_serving_image_and_preserves_other_environment_values(): assert writes[2][1]["traffic"][0]["revision"] == "sg-new" -@pytest.mark.parametrize(("options", "writes_before_stop"), [({"enabled": True}, 0), ({"drift": True}, 2)]) +@pytest.mark.parametrize(("options", "writes_before_stop"), [ + ({"enabled": True}, 0), ({"stale_template": True}, 0), ({"drift": True}, 2), +]) def test_rejects_enabled_trading_or_staged_configuration_drift(options, writes_before_stop): request, writes = fixture(**options) with pytest.raises(ValueError, match="readback_mismatch"): - enable_snapshot(request, SERVICE) + enable_snapshot(request, SERVICE, SOURCE) assert len(writes) == writes_before_stop + + +def test_rejects_a_serving_source_that_does_not_match_the_protected_pin(): + request, writes = fixture() + with pytest.raises(ValueError, match="readback_mismatch"): + enable_snapshot(request, SERVICE, "3" * 40) + assert writes == []