diff --git a/.github/workflows/notify-route-check-protected.yml b/.github/workflows/notify-route-check-protected.yml index 6ed891c..41fbb5c 100644 --- a/.github/workflows/notify-route-check-protected.yml +++ b/.github/workflows/notify-route-check-protected.yml @@ -1,44 +1,55 @@ name: Notify Route Check (protected execution) -# Register the original JSON as a workflow secret before any runner step header. -# This masks the passed value, not every extracted field or reserialized variant. +# Reusable secrets register the original private values before runner headers. +# No environment/secret/IAM/route changes; no Telegram message is sent. on: workflow_call: secrets: + GCP_PROJECT_ID: + required: true + GCP_WORKLOAD_IDENTITY_PROVIDER: + required: true + GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT: + required: true + RUNTIME_GUARD_CLOUD_RUN_SERVICES: + required: false CLOUD_RUN_SERVICE: required: false CLOUD_RUN_SERVICES: required: false CLOUD_RUN_SERVICE_TARGETS_JSON: required: false + CLOUD_RUN_REGION: + required: false + TELEGRAM_TOKEN_SECRET_NAME: + required: false permissions: contents: read id-token: write -env: - GCP_PROJECT_ID: interactivebrokersquant - GCP_WORKLOAD_IDENTITY_PROVIDER: projects/303168642265/locations/global/workloadIdentityPools/github-actions/providers/github-main - GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT: ibkr-platform-deploy@interactivebrokersquant.iam.gserviceaccount.com - jobs: route-check: - name: Telegram route check (read-only) + name: Telegram route identity (read-only) + if: ${{ github.event_name == 'workflow_dispatch' && github.ref == 'refs/heads/main' }} runs-on: ubuntu-latest timeout-minutes: 10 - permissions: - contents: read - id-token: write env: - RUNTIME_GUARD_CLOUD_RUN_SERVICES: ${{ vars.RUNTIME_GUARD_CLOUD_RUN_SERVICES }} + GCP_PROJECT_ID: ${{ secrets.GCP_PROJECT_ID }} + GCP_WORKLOAD_IDENTITY_PROVIDER: ${{ secrets.GCP_WORKLOAD_IDENTITY_PROVIDER }} + GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT: ${{ secrets.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }} + RUNTIME_GUARD_CLOUD_RUN_SERVICES: ${{ secrets.RUNTIME_GUARD_CLOUD_RUN_SERVICES }} CLOUD_RUN_SERVICE: ${{ secrets.CLOUD_RUN_SERVICE }} CLOUD_RUN_SERVICES: ${{ secrets.CLOUD_RUN_SERVICES }} CLOUD_RUN_SERVICE_TARGETS_JSON: ${{ secrets.CLOUD_RUN_SERVICE_TARGETS_JSON }} - CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }} - TELEGRAM_TOKEN_SECRET_NAME: ${{ vars.TELEGRAM_TOKEN_SECRET_NAME }} + CLOUD_RUN_REGION: ${{ secrets.CLOUD_RUN_REGION }} + TELEGRAM_TOKEN_SECRET_NAME: ${{ secrets.TELEGRAM_TOKEN_SECRET_NAME }} + SOURCE_SHA: ${{ github.sha }} steps: - name: Checkout repository uses: actions/checkout@v6 + with: + persist-credentials: false - name: Authenticate to Google Cloud uses: google-github-actions/auth@v3 @@ -49,5 +60,5 @@ jobs: - name: Set up gcloud uses: google-github-actions/setup-gcloud@v3 - - name: Report Telegram route (no secret values) + - name: Observe serving Telegram binding (equality only) run: python3 scripts/notify_route_check.py diff --git a/.github/workflows/notify-route-check.yml b/.github/workflows/notify-route-check.yml index 8f71fb0..6e2c7b0 100644 --- a/.github/workflows/notify-route-check.yml +++ b/.github/workflows/notify-route-check.yml @@ -1,19 +1,25 @@ name: Notify Route Check -# Read-only: reports which Secret Manager secret each live Cloud Run service binds -# to TELEGRAM_TOKEN and resolves it to the public bot identity (getMe id/username). -# Never prints token or chat id; never changes Cloud Run, secrets, or trading. +# Read-only, equality-only serving binding diagnosis. Private inputs cross the +# reusable-workflow secret boundary before any runner step header. on: workflow_dispatch: jobs: route-check: - name: Telegram route check (read-only) + name: Telegram route identity (read-only) + if: ${{ github.event_name == 'workflow_dispatch' && github.ref == 'refs/heads/main' }} permissions: contents: read id-token: write uses: ./.github/workflows/notify-route-check-protected.yml secrets: + GCP_PROJECT_ID: ${{ 'interactivebrokersquant' }} + GCP_WORKLOAD_IDENTITY_PROVIDER: ${{ 'projects/303168642265/locations/global/workloadIdentityPools/github-actions/providers/github-main' }} + GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT: ${{ 'ibkr-platform-deploy@interactivebrokersquant.iam.gserviceaccount.com' }} + RUNTIME_GUARD_CLOUD_RUN_SERVICES: ${{ vars.RUNTIME_GUARD_CLOUD_RUN_SERVICES }} CLOUD_RUN_SERVICE: ${{ secrets.CLOUD_RUN_SERVICE }} CLOUD_RUN_SERVICES: ${{ secrets.CLOUD_RUN_SERVICES }} CLOUD_RUN_SERVICE_TARGETS_JSON: ${{ vars.CLOUD_RUN_SERVICE_TARGETS_JSON || secrets.CLOUD_RUN_SERVICE_TARGETS_JSON }} + CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }} + TELEGRAM_TOKEN_SECRET_NAME: ${{ vars.TELEGRAM_TOKEN_SECRET_NAME }} diff --git a/.github/workflows/runtime-target-lifecycle.yml b/.github/workflows/runtime-target-lifecycle.yml index 4124230..f00df07 100644 --- a/.github/workflows/runtime-target-lifecycle.yml +++ b/.github/workflows/runtime-target-lifecycle.yml @@ -157,7 +157,8 @@ jobs: exit 0 fi - export RUNTIME_TARGET_JSON="$(cat runtime-target.json)" + export RUNTIME_TARGET_JSON + RUNTIME_TARGET_JSON="$(cat runtime-target.json)" set +e uv run --no-sync python scripts/execution_report_heartbeat.py > execution-heartbeat.log 2>&1 exit_code=$? @@ -178,7 +179,8 @@ jobs: id: production_drift run: | set -uo pipefail - export RUNTIME_TARGET_JSON="$(cat runtime-target.json)" + export RUNTIME_TARGET_JSON + RUNTIME_TARGET_JSON="$(cat runtime-target.json)" set +e uv run --no-sync python scripts/production_drift_health_observe.py > production-drift.log 2>&1 exit_code=$? diff --git a/docs/notify-route-identity.zh-CN.md b/docs/notify-route-identity.zh-CN.md new file mode 100644 index 0000000..314f0d0 --- /dev/null +++ b/docs/notify-route-identity.zh-CN.md @@ -0,0 +1,39 @@ +# Serving Telegram 只读身份诊断 + +入口为 `notify-route-check.yml` → `notify-route-check-protected.yml` → `scripts/notify_route_check.py`。仅接受 main 分支的手动 workflow_dispatch；不改生产设置、secret、IAM、Cloud Run、通知路由、交易或预算。 + +## 实际读取与证明范围 + +- 读取既有服务配置；private list/target JSON/region 与既有 GCP 设置以同值 reusable secrets 传入，在 runner header 前注册掩码。不新建生产 secret，不打印 `add-mask` 命令或原值。权限仍为原 `contents: read` 与 `id-token: write`，沿用原 WIF / SA。 +- 从 Cloud Run **status positive traffic** 逐 revision describe 其实际 containers/env；不把 latestReady 或 service template 当 serving。要求流量记录合法且覆盖 100%，不猜缺失 revision。多流量/split 仍读取各 revision，但全体与每行 bot 比较都保持 unknown/null；重复或不完整流量也 unknown。 +- 只有单一 container 的 TELEGRAM_TOKEN 可以解析。明确 plain env 仅在内存用于 getMe；duplicate/缺失/非法引用不猜。该 runtime selector 不使用 TG_TOKEN；主 token 无效或探测失败时不借次优 token 救成功。 +- secretKeyRef 的版本必须是精确正整数。读取绑定项目/secret/该固定版本；支持完整 resource 及 `run.googleapis.com/secrets` 的明确 alias→完整 resource 映射。错误/重复 alias 不回落到同项目同名 secret。**latest/其他版本别名不能证明正在运行进程加载的 immutable token，故保持 unknown，不自动改绑定。** +- 比较参考是**本项目**既有 `quant-sentinel-telegram-bot-token` 当次 latest 读取后 getMe 的 bot 身份。它不是中央已批准 canonical reference，也不因 secret 名相同证明跨项目相同。两端 getMe 只 GET，检查 ok=true、is_bot=true、有效正整数 id；缺失、读失败、网络/坏响应均 null/unknown，不写 false。仅两端身份都已知且 sole-serving 无歧义时输出 match/true 或 mismatch/false。 +- 所有 gcloud 命令 stdout/stderr 捕获，30 秒超时；getMe 15 秒超时并限制响应 64 KiB。没有 update、sendMessage、broker、真实业务或交易调用。服务/引用值与 token 只在私有内存中使用。 + +这是一次读取快照的 Telegram **配置绑定**比较；不能由诊断 source SHA 反推 runtime 镜像、已安装代码、有效 channel 或真实交付。服务若正改变流量，需重新观察稳定状态，不把单次快照当原子发布证明。 + +## 公开输出 + +stdout 和 GitHub summary 使用 `qsl.notify_route_identity.v1`：固定 status/reason enum、nullable bool、`service#N` / `revision#N` ordinal，以及经 40 位小写 hex 校验的诊断 `source_sha`。不输出 token/chat/hash/last4、project/service/revision/SA/secret 引用/版本、bot 原 ID/name、自由错误/URL。没有公开私有文件 artifact。 + +`status=ok` / exit 0 仅表示诊断形成结果，**不表示 route 已就绪**。看逐服务 `bot_status=match|mismatch|unknown`、`same_bot_as_contract=true|false|null` 与固定 reason。scope 始终 `project_contract_only`，`canonical_route_status=unknown`。 + +## 尚未验收的入口 + +`same_chat_as_canonical` 和 `same_topic_as_canonical` 固定 null、对应 status unknown：当前没有批准的中央 private canonical reference，本批不创造新 secret、全局路由规则或 topic 值，也不额外读取 chat secret。既有 runtime chat 为 QSL_GLOBAL 优先 GLOBAL；heartbeat/guard 只遍历 GLOBAL 全列表；中央日报优先 QSL_GLOBAL/GLOBAL/plugin 且只取首 ID。它们需要按各自原 selector 在受保护比较边界取得三态证据，不能重新统一解析来使其匹配。 + +本诊断没有核验 GHA heartbeat/guard 的注入、scanner/plugin/drill/VPS 实际安装或中央 Environment；本仓所查发送器未给 message_thread_id，不等于自然消息已经进入同一 topic。same bot 不等于相同 chat/topic、shared dedup 或 exactly-once。保留既有 alert/订单可达与 quiet，不切换 dedup mode。 + +B7/B8 的中央生产 reader/sections_v2 采用、批准 producer/runtime/source 身份、自然周期安全 receipt 和最终 Telegram 截图仍是独立验收项；不会用只读 success、合成夹具或旧截图代替。 + +## 本地验证 + +```bash +uv sync --frozen --extra test +PYTHONPATH=. PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 uv run --no-sync python -m pytest -q tests/test_notify_route_check.py +uv run --no-sync ruff check --exclude external . +PATH=/path/to/shellcheck:$PATH actionlint +``` + +回归仅 fake gcloud/urllib 与固定合成路由；还应运行 CONTRIBUTING 全量与锁/pin/启动检查。源码工程验收后由主智能体在最终 main 执行受控只读诊断，结果依真实 read/getMe 状态分项记录；不把缺失事实补成匹配。 diff --git a/scripts/notify_route_check.py b/scripts/notify_route_check.py index 045f28e..21570f6 100644 --- a/scripts/notify_route_check.py +++ b/scripts/notify_route_check.py @@ -1,168 +1,278 @@ #!/usr/bin/env python3 -"""Read-only Telegram route check for Cloud Run services (no token/chat printed). +"""Observe serving Telegram bindings without disclosing private route material. -For each configured Cloud Run service, report which Secret Manager secret the -live service binds to TELEGRAM_TOKEN, then resolve each distinct secret (plus -the QuantSentinel contract secret) to its public bot identity via Telegram -``getMe``. Only bot id / username are printed; token values are masked and -never written. Does not change Cloud Run, secrets, or trading state. +Only a sole serving revision with one container and a pinned token binding can +be compared with this project's current contract bot. This is not central +canonical route approval. Cloud commands are read-only; Telegram uses getMe +only. No route values, identifiers, token masks or exception text are printed. """ from __future__ import annotations import json import os +from pathlib import Path +import re import subprocess import sys import urllib.request CONTRACT_SECRET = "quant-sentinel-telegram-bot-token" +_PROJECT = r"(?:[a-z][a-z0-9-]{4,28}[a-z0-9]|[1-9][0-9]{5,19})" +_SECRET = r"[A-Za-z0-9_-]{1,255}" +_SERVICE = r"[a-z](?:[a-z0-9-]{0,61}[a-z0-9])?" +_MAX_GETME_BYTES = 64 * 1024 def _services() -> list[str]: - raw: list[str] = [] + names: list[str] = [] for key in ("CLOUD_RUN_SERVICE", "CLOUD_RUN_SERVICES", "RUNTIME_GUARD_CLOUD_RUN_SERVICES"): - raw.extend(p.strip() for p in (os.environ.get(key) or "").replace("\n", ",").split(",")) - targets = (os.environ.get("CLOUD_RUN_SERVICE_TARGETS_JSON") or "").strip() - if targets: - try: - data = json.loads(targets) - items = data if isinstance(data, list) else data.get("targets", []) if isinstance(data, dict) else [] - for item in items: - if isinstance(item, dict): - for key in ("service", "service_name", "cloud_run_service"): - value = item.get(key) or (item.get("runtime_target") or {}).get(key) - if isinstance(value, str) and value.strip(): - raw.append(value.strip()) - break - elif isinstance(item, str): - raw.append(item.strip()) - except json.JSONDecodeError: - pass - seen: list[str] = [] - for name in raw: - if name and name not in seen: - seen.append(name) - return seen + names.extend(item.strip() for item in (os.environ.get(key) or "").replace("\n", ",").split(",") + if item.strip()) + raw = (os.environ.get("CLOUD_RUN_SERVICE_TARGETS_JSON") or "").strip() + if raw: + data = json.loads(raw) + items = data if isinstance(data, list) else data.get("targets") if isinstance(data, dict) else None + if not isinstance(items, list): + raise ValueError("invalid_targets") + for item in items: + if isinstance(item, str): + candidates = [item.strip()] + elif isinstance(item, dict): + nested = item.get("runtime_target") or {} + if not isinstance(nested, dict): + raise ValueError("invalid_targets") + candidates = [obj[key].strip() for obj in (item, nested) + for key in ("service", "service_name", "cloud_run_service") + if isinstance(obj.get(key), str) and obj[key].strip()] + else: + raise ValueError("invalid_targets") + if len(set(candidates)) != 1: + raise ValueError("invalid_targets") + names.append(candidates[0]) + if not names or any(not re.fullmatch(_SERVICE, name) for name in names): + raise ValueError("invalid_targets") + return list(dict.fromkeys(names)) def _gcloud(*args: str) -> str: - return subprocess.run(["gcloud", *args], check=True, capture_output=True, text=True).stdout - - -def _telegram_binding(service: str, project: str, region: str) -> dict: - spec = json.loads(_gcloud("run", "services", "describe", service, f"--project={project}", - f"--region={region}", "--format=json")) - containers = spec.get("spec", {}).get("template", {}).get("spec", {}).get("containers", []) - status = spec.get("status", {}) - out = { - "latest_ready_revision": status.get("latestReadyRevisionName"), - "traffic": [ - {"revision": t.get("revisionName"), "percent": t.get("percent")} - for t in status.get("traffic", []) if t.get("percent") - ], - "telegram_token_source": "missing", - "telegram_token_secret": None, - "telegram_token_secret_version": None, - } - out["template_service_account"] = ( - spec.get("spec", {}).get("template", {}).get("spec", {}).get("serviceAccountName") - ) - serving = [] - for entry in out["traffic"]: - rev = entry.get("revision") - sa = None - if rev: - try: - rev_spec = json.loads(_gcloud("run", "revisions", "describe", rev, f"--project={project}", - f"--region={region}", "--format=json")) - sa = rev_spec.get("spec", {}).get("serviceAccountName") - except subprocess.CalledProcessError: - sa = "describe_failed" - serving.append({"revision": rev, "percent": entry.get("percent"), "runtime_service_account": sa}) - out["serving_revisions"] = serving - for container in containers: - for env in container.get("env", []) or []: - if env.get("name") != "TELEGRAM_TOKEN": - continue - ref = (env.get("valueFrom") or {}).get("secretKeyRef") - if ref: - out["telegram_token_source"] = "secret_ref" - out["telegram_token_secret"] = ref.get("name") - out["telegram_token_secret_version"] = ref.get("key") - elif env.get("value"): - out["telegram_token_source"] = "plain_env_value" - return out + return subprocess.run(["gcloud", *args], check=True, capture_output=True, text=True, timeout=30).stdout -def _bot_identity(secret: str, project: str) -> dict: +def _bot_identity(token: str) -> tuple[int | None, str]: + if not re.fullmatch(r"[0-9]+:[A-Za-z0-9_-]+", token): + return None, "token_invalid" try: - token = _gcloud("secrets", "versions", "access", "latest", f"--secret={secret}", - f"--project={project}").strip() - except subprocess.CalledProcessError as exc: - stderr = (exc.stderr or "").upper() - reason = "unknown" - for marker in ("NOT_FOUND", "PERMISSION_DENIED", "FAILED_PRECONDITION", "UNAUTHENTICATED"): - if marker in stderr: - reason = marker.lower() - break - return {"secret": secret, "status": "secret_inaccessible", "reason": reason} - if not token: - return {"secret": secret, "status": "secret_empty"} - print(f"::add-mask::{token}") + request = urllib.request.Request(f"https://api.telegram.org/bot{token}/getMe", method="GET") + with urllib.request.urlopen(request, timeout=15) as response: + raw = response.read(_MAX_GETME_BYTES + 1) + if len(raw) > _MAX_GETME_BYTES: + return None, "getme_invalid" + body = json.loads(raw) + result = body.get("result") if isinstance(body, dict) and body.get("ok") is True else None + bot_id = result.get("id") if isinstance(result, dict) else None + if type(bot_id) is not int or bot_id <= 0 or result.get("is_bot") is not True: + return None, "getme_invalid" + return bot_id, "known" + except Exception: # noqa: BLE001 - never disclose URLs, response bodies or exceptions + return None, "getme_failed" + + +def _secret_token(project: str, secret: str, version: str) -> tuple[str | None, str]: try: - with urllib.request.urlopen(f"https://api.telegram.org/bot{token}/getMe", timeout=15) as resp: - body = json.loads(resp.read().decode("utf-8")) - except Exception as exc: # noqa: BLE001 - report class only, never the URL - return {"secret": secret, "status": f"getme_failed:{type(exc).__name__}"} - finally: - token = "" - result = body.get("result") or {} + token = _gcloud("secrets", "versions", "access", version, f"--secret={secret}", f"--project={project}") + except Exception: # noqa: BLE001 - private command streams and arguments stay in memory + return None, "secret_read_failed" + if not token: + return None, "token_missing" + return token, "known" + + +def _secret_reference(name: str, revision: dict, project: str) -> tuple[str, str] | None: + metadata = revision.get("metadata") or {} + annotations = metadata.get("annotations") or {} if isinstance(metadata, dict) else None + if not isinstance(annotations, dict): + return None + aliases = annotations.get("run.googleapis.com/secrets") + aliased = False + if aliases is not None: + if not isinstance(aliases, str): + return None + matches = [] + for declaration in aliases.split(","): + alias, separator, reference = declaration.strip().partition(":") + if not separator or not alias or not reference: + return None + if alias == name: + matches.append(reference) + if len(matches) > 1: + return None + if matches: + name = matches[0] + aliased = True + qualified = re.fullmatch(r"projects/(" + _PROJECT + r")/secrets/(" + _SECRET + r")", name) + if qualified: + return qualified.group(1), qualified.group(2) + if not aliased and re.fullmatch(_SECRET, name): + return project, name + return None + + +def _binding_token(revision: dict, project: str) -> tuple[str | None, str, str]: + spec = revision.get("spec") + containers = spec.get("containers") if isinstance(spec, dict) else None + if not isinstance(containers, list) or not containers: + return None, "unknown", "containers_missing" + if len(containers) != 1 or not isinstance(containers[0], dict): + return None, "unknown", "containers_ambiguous" + envs = containers[0].get("env") + if not isinstance(envs, list) or any(not isinstance(item, dict) for item in envs): + return None, "unknown", "token_environment_invalid" + winners = [item for item in envs if item.get("name") == "TELEGRAM_TOKEN"] + if not winners: + return None, "unknown", "token_binding_missing" + if len(winners) != 1: + return None, "unknown", "token_binding_ambiguous" + winner = winners[0] + if "value" in winner and "valueFrom" in winner: + return None, "unknown", "token_binding_ambiguous" + if "value" in winner: + token = winner["value"] + if not isinstance(token, str) or not token: + return None, "plain_env", "token_missing" + return token, "plain_env", "known" + value_from = winner.get("valueFrom") + ref = value_from.get("secretKeyRef") if isinstance(value_from, dict) else None + if not isinstance(ref, dict) or not isinstance(ref.get("name"), str): + return None, "unknown", "token_binding_invalid" + version = ref.get("key") + # Alias/latest cannot identify the immutable token loaded by a running process. + if not isinstance(version, str) or not re.fullmatch(r"[1-9][0-9]{0,19}", version): + return None, "secret_version", "token_version_unpinned" + reference = _secret_reference(ref["name"], revision, project) + if reference is None: + return None, "secret_version", "token_binding_invalid" + token, status = _secret_token(*reference, version) + return token, "secret_version", status + + +def _traffic(service: dict) -> tuple[list[str], str, str]: + status = service.get("status") + rows = status.get("traffic") if isinstance(status, dict) else None + if not isinstance(rows, list): + return [], "unknown", "traffic_invalid" + serving = [] + total = 0 + for row in rows: + percent = row.get("percent") if isinstance(row, dict) else None + if type(percent) is not int or not 0 <= percent <= 100: + return [], "unknown", "traffic_invalid" + if percent == 0: + continue + name = row.get("revisionName") + if not isinstance(name, str) or not re.fullmatch(_SERVICE, name) or name in serving: + return [], "unknown", "traffic_ambiguous" + serving.append(name) + total += percent + if not serving: + return [], "none", "no_positive_traffic" + if total != 100: + return [], "unknown", "traffic_incomplete" + return serving, "single" if len(serving) == 1 else "split", "none" + + +def _unknown_comparison() -> dict: return { - "secret": secret, - "status": "ok" if body.get("ok") else "getme_not_ok", - "bot_id": result.get("id"), - "bot_username": result.get("username"), - "bot_first_name": result.get("first_name"), + "bot_status": "unknown", "same_bot_as_contract": None, + "chat_status": "unknown", "same_chat_as_canonical": None, + "topic_status": "unknown", "same_topic_as_canonical": None, } -def main() -> int: - project = os.environ["GCP_PROJECT_ID"] - region = os.environ.get("CLOUD_RUN_REGION") or "us-central1" - services = _services() - report: dict = {"project": project, "region": region, "services": [], "bots": []} - secrets = {CONTRACT_SECRET} - gh_secret_name = (os.environ.get("TELEGRAM_TOKEN_SECRET_NAME") or "").strip() - if gh_secret_name: - secrets.add(gh_secret_name) - report["github_variable_TELEGRAM_TOKEN_SECRET_NAME"] = gh_secret_name or None - for index, service in enumerate(services, start=1): +def _observe_service(service: str, ordinal: int, project: str, region: str, contract_id: int | None) -> dict: + out = {"service_label": f"service#{ordinal}", "traffic_status": "unknown", + "reason_code": "service_describe_failed", **_unknown_comparison(), "revisions": []} + try: + spec = json.loads(_gcloud("run", "services", "describe", service, f"--project={project}", + f"--region={region}", "--format=json")) + except Exception: # noqa: BLE001 - fixed reason only + return out + if not isinstance(spec, dict): + out["reason_code"] = "service_metadata_invalid" + return out + serving, traffic_status, reason = _traffic(spec) + out.update(traffic_status=traffic_status, reason_code=reason) + if not serving: + return out + for index, name in enumerate(serving, start=1): + observed = {"revision_label": f"revision#{index}", "token_source": "unknown", + "token_status": "revision_describe_failed", "reason_code": "revision_describe_failed", + **_unknown_comparison()} try: - binding = _telegram_binding(service, project, region) - except subprocess.CalledProcessError: - binding = {"telegram_token_source": "describe_failed"} - binding["service_label"] = f"service#{index}" - # Service name / region are not secrets (owner needs them for gcloud). - # GitHub may still redact the name if it equals a repository secret value. - binding["service_name"] = service - binding["region"] = region - if binding.get("telegram_token_secret"): - secrets.add(binding["telegram_token_secret"]) - report["services"].append(binding) - for secret in sorted(secrets): - report["bots"].append(_bot_identity(secret, project)) - contract_bot = next((b for b in report["bots"] if b["secret"] == CONTRACT_SECRET), {}) - for svc in report["services"]: - bound = next((b for b in report["bots"] if b["secret"] == svc.get("telegram_token_secret")), {}) - svc["same_bot_as_contract"] = ( - bool(bound.get("bot_id")) and bound.get("bot_id") == contract_bot.get("bot_id") - ) + revision = json.loads(_gcloud("run", "revisions", "describe", name, f"--project={project}", + f"--region={region}", "--format=json")) + if not isinstance(revision, dict): + observed.update(token_status="revision_metadata_invalid", reason_code="revision_metadata_invalid") + else: + token, source, token_status = _binding_token(revision, project) + observed.update(token_source=source, token_status=token_status, reason_code=token_status) + if token is not None: + bot_id, bot_status = _bot_identity(token) + token = None + observed.update(token_status=bot_status, reason_code=bot_status) + if traffic_status == "single" and bot_id is not None and contract_id is not None: + same = bot_id == contract_id + observed.update(same_bot_as_contract=same, bot_status="match" if same else "mismatch", + reason_code="project_contract_compared") + elif bot_id is not None: + observed["reason_code"] = "split_traffic" if traffic_status == "split" else "contract_unavailable" + except Exception: # noqa: BLE001 - never emit metadata, private values or tracebacks + pass + out["revisions"].append(observed) + if traffic_status == "single": + only = out["revisions"][0] + out.update(bot_status=only["bot_status"], same_bot_as_contract=only["same_bot_as_contract"], + reason_code=only["reason_code"]) + else: + out["reason_code"] = "split_traffic" + return out + + +def _report() -> dict: + source = os.environ.get("SOURCE_SHA") or "" + report = { + "schema_version": "qsl.notify_route_identity.v1", + "source_sha": source if re.fullmatch(r"[0-9a-f]{40}", source) else None, + "status": "config_unknown", "reference_scope": "project_contract_only", + "canonical_route_status": "unknown", "contract_bot_status": "unavailable", "services": [], + } + project = os.environ.get("GCP_PROJECT_ID") or "" + region = os.environ.get("CLOUD_RUN_REGION") or "us-central1" + if not re.fullmatch(_PROJECT, project): + return report + try: + services = _services() + except Exception: # noqa: BLE001 - do not report invalid private configuration + return report + token, status = _secret_token(project, CONTRACT_SECRET, "latest") + contract_id = None + if token is not None: + contract_id, status = _bot_identity(token) + token = None + report.update(status="ok", contract_bot_status=status) + report["services"] = [_observe_service(name, index, project, region, contract_id) + for index, name in enumerate(services, start=1)] + return report + + +def main() -> int: + report = _report() text = json.dumps(report, ensure_ascii=False, indent=2) print(text) summary = os.environ.get("GITHUB_STEP_SUMMARY") if summary: - with open(summary, "a", encoding="utf-8") as fh: - fh.write("## Telegram route check (read-only)\n\n```json\n" + text + "\n```\n") + with Path(summary).open("a", encoding="utf-8") as stream: + stream.write("## Telegram route identity (read-only)\n\n```json\n" + text + "\n```\n") + # Diagnosis completion is not route readiness; nullable result fields carry unknown. return 0 diff --git a/tests/test_notify_route_check.py b/tests/test_notify_route_check.py new file mode 100644 index 0000000..5089841 --- /dev/null +++ b/tests/test_notify_route_check.py @@ -0,0 +1,393 @@ +from __future__ import annotations + +import importlib.util +import io +import json +import os +from pathlib import Path +import re +import subprocess +from unittest import mock + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +SPEC = importlib.util.spec_from_file_location("readonly_notify_route_check", ROOT / "scripts/notify_route_check.py") +route = importlib.util.module_from_spec(SPEC) +assert SPEC.loader is not None +SPEC.loader.exec_module(route) + +PROJECT = "synthetic-project" +SERVICE = "private-service" +REVISION = "private-serving" +VERSION = "97007" +BOUND_SECRET = "private-bound-secret" +CONTRACT_TOKEN = "98765001:PRIVATE_CONTRACT_TOKEN" +BOUND_TOKEN = "98765002:PRIVATE_BOUND_TOKEN" +OTHER_TOKEN = "98765003:PRIVATE_OTHER_TOKEN" +CONTRACT_BOT = 98765001 +OTHER_BOT = 98765002 +SOURCE_SHA = "a" * 40 + + +def secret_env(name=BOUND_SECRET, version=VERSION): + return {"name": "TELEGRAM_TOKEN", "valueFrom": {"secretKeyRef": {"name": name, "key": version}}} + + +def revision(env=None, containers=None, annotations=None): + return {"metadata": {"annotations": annotations or {}}, + "spec": {"serviceAccountName": "private-sa@example.test", + "containers": containers if containers is not None else [{"env": env if env is not None else [ + secret_env(), {"name": "QSL_GLOBAL_TELEGRAM_CHAT_ID", "value": "PRIVATE_CHAT_QSL"}, + {"name": "GLOBAL_TELEGRAM_CHAT_ID", "value": "PRIVATE_CHAT_GLOBAL"}, + ]}]}} + + +class FakeCloud: + def __init__(self): + self.calls = [] + self.service = { + "spec": {"template": revision([secret_env(route.CONTRACT_SECRET, "latest")])}, + "status": {"latestReadyRevisionName": "private-ready", + "traffic": [{"revisionName": REVISION, "percent": 100}]}, + } + self.revisions = {REVISION: revision()} + self.secrets = { + (PROJECT, route.CONTRACT_SECRET, "latest"): CONTRACT_TOKEN, + (PROJECT, BOUND_SECRET, VERSION): BOUND_TOKEN, + (PROJECT, BOUND_SECRET, "latest"): CONTRACT_TOKEN, + } + self.fail = set() + + def __call__(self, *args): + self.calls.append(args) + if args[:3] == ("run", "services", "describe"): + key = ("service", args[3]) + value = self.service + elif args[:3] == ("run", "revisions", "describe"): + key = ("revision", args[3]) + value = self.revisions[args[3]] + elif args[:3] == ("secrets", "versions", "access"): + project = next(arg.removeprefix("--project=") for arg in args if arg.startswith("--project=")) + secret = next(arg.removeprefix("--secret=") for arg in args if arg.startswith("--secret=")) + key = ("secret", project, secret, args[3]) + value = self.secrets[(project, secret, args[3])] + else: + raise AssertionError("Unexpected cloud operation") + if key in self.fail: + raise subprocess.CalledProcessError(1, ["PRIVATE_COMMAND"], output="PRIVATE_STDOUT", stderr="PRIVATE_STDERR") + return value if isinstance(value, str) else json.dumps(value) + + +class FakeReply(io.BytesIO): + def __enter__(self): + return self + + def __exit__(self, *args): + self.close() + + +class FakeTelegram: + def __init__(self): + self.calls = [] + self.results = { + CONTRACT_TOKEN: {"ok": True, "result": {"id": CONTRACT_BOT, "is_bot": True, + "username": "PRIVATE_BOT_NAME"}}, + BOUND_TOKEN: {"ok": True, "result": {"id": OTHER_BOT, "is_bot": True}}, + OTHER_TOKEN: {"ok": True, "result": {"id": CONTRACT_BOT, "is_bot": True}}, + } + + def __call__(self, request, **kwargs): + url = request.full_url if hasattr(request, "full_url") else request + method = request.get_method() if hasattr(request, "get_method") else "GET" + assert method == "GET" + assert url.startswith("https://api.telegram.org/bot") and url.endswith("/getMe") + token = url.removeprefix("https://api.telegram.org/bot").removesuffix("/getMe") + self.calls.append(token) + value = self.results[token] + if isinstance(value, Exception): + raise value + raw = value if isinstance(value, bytes) else json.dumps(value).encode() + return FakeReply(raw) + + +def observe(tmp_path, cloud=None, telegram=None, **env_overrides): + cloud = cloud or FakeCloud() + telegram = telegram or FakeTelegram() + env = {"GCP_PROJECT_ID": PROJECT, "CLOUD_RUN_REGION": "private-region", "CLOUD_RUN_SERVICE": SERVICE, + "SOURCE_SHA": SOURCE_SHA, "GITHUB_STEP_SUMMARY": str(tmp_path / "summary"), **env_overrides} + output = io.StringIO() + errors = io.StringIO() + with mock.patch.dict(os.environ, env, clear=True), mock.patch.object(route, "_gcloud", cloud), \ + mock.patch.object(route.urllib.request, "urlopen", telegram), \ + mock.patch("sys.stdout", output), mock.patch("sys.stderr", errors): + code = route.main() + text = output.getvalue() + summary = (tmp_path / "summary").read_text() + for private in [PROJECT, SERVICE, REVISION, VERSION, BOUND_SECRET, route.CONTRACT_SECRET, + CONTRACT_TOKEN, BOUND_TOKEN, "private-ready", "private-sa@example.test", "private-region", + "PRIVATE_CHAT", "PRIVATE_STDOUT", "PRIVATE_STDERR", "PRIVATE_BOT", "PRIVATE_COMMAND", + "api.telegram.org", str(CONTRACT_BOT), str(OTHER_BOT), "::add-mask::"]: + assert private not in text + errors.getvalue() + summary + assert errors.getvalue() == "" + assert code == 0 + report = json.loads(text) + assert report["schema_version"] == "qsl.notify_route_identity.v1" + assert report["reference_scope"] == "project_contract_only" + assert report["canonical_route_status"] == "unknown" + for service in report["services"]: + for row in [service, *service["revisions"]]: + assert row["same_chat_as_canonical"] is None + assert row["same_topic_as_canonical"] is None + assert row["chat_status"] == row["topic_status"] == "unknown" + return report, cloud, telegram + + +def test_template_and_secret_latest_are_not_serving_evidence(tmp_path): + report, cloud, _ = observe(tmp_path) + row = report["services"][0] + assert row["same_bot_as_contract"] is False + assert row["bot_status"] == "mismatch" + assert report["source_sha"] == SOURCE_SHA + assert ("run", "revisions", "describe", REVISION, f"--project={PROJECT}", + "--region=private-region", "--format=json") in cloud.calls + assert ("secrets", "versions", "access", VERSION, f"--secret={BOUND_SECRET}", + f"--project={PROJECT}") in cloud.calls + assert not any(args[:4] == ("secrets", "versions", "access", "latest") + and f"--secret={BOUND_SECRET}" in args for args in cloud.calls) + + +def test_fixed_version_can_match_while_latest_and_template_differ(tmp_path): + cloud = FakeCloud() + cloud.secrets[(PROJECT, BOUND_SECRET, VERSION)] = OTHER_TOKEN + cloud.secrets[(PROJECT, BOUND_SECRET, "latest")] = BOUND_TOKEN + report, _, _ = observe(tmp_path, cloud) + assert report["services"][0]["same_bot_as_contract"] is True + assert report["services"][0]["bot_status"] == "match" + + +def test_same_secret_name_does_not_prove_same_bot(tmp_path): + cloud = FakeCloud() + cloud.revisions[REVISION] = revision([secret_env(route.CONTRACT_SECRET)]) + cloud.secrets[(PROJECT, route.CONTRACT_SECRET, VERSION)] = BOUND_TOKEN + report, _, _ = observe(tmp_path, cloud) + assert report["services"][0]["same_bot_as_contract"] is False + + +@pytest.mark.parametrize("traffic", [ + None, {}, [], [{"revisionName": REVISION, "percent": 0}], [None], + [{"revisionName": REVISION}], [{"revisionName": REVISION, "percent": True}], + [{"revisionName": REVISION, "percent": "100"}], [{"revisionName": REVISION, "percent": -1}], + [{"revisionName": REVISION, "percent": 101}], [{"revisionName": REVISION, "percent": 50}], + [{"latestRevision": True, "percent": 100}], [{"revisionName": "", "percent": 100}], + [{"revisionName": REVISION, "percent": 50}, {"revisionName": REVISION, "percent": 50}], +]) +def test_missing_partial_invalid_and_ambiguous_traffic_is_unknown(tmp_path, traffic): + cloud = FakeCloud() + cloud.service["status"]["traffic"] = traffic + report, _, _ = observe(tmp_path, cloud) + row = report["services"][0] + assert row["same_bot_as_contract"] is None + assert row["bot_status"] == "unknown" + assert not any(args[:3] == ("run", "revisions", "describe") for args in cloud.calls) + + +def test_split_traffic_reads_each_revision_and_remains_unknown(tmp_path): + cloud = FakeCloud() + cloud.service["status"]["traffic"] = [ + {"revisionName": REVISION, "percent": 50}, {"revisionName": "private-serving-two", "percent": 50}, + {"revisionName": "private-zero-traffic", "percent": 0}, + ] + cloud.revisions["private-serving-two"] = revision([secret_env(version="97008")]) + cloud.secrets[(PROJECT, BOUND_SECRET, "97008")] = OTHER_TOKEN + report, _, _ = observe(tmp_path, cloud) + row = report["services"][0] + assert row["traffic_status"] == "split" + assert row["same_bot_as_contract"] is None + assert len(row["revisions"]) == 2 + assert all(item["same_bot_as_contract"] is None for item in row["revisions"]) + described = [args[3] for args in cloud.calls if args[:3] == ("run", "revisions", "describe")] + assert described == [REVISION, "private-serving-two"] + assert not any("private-zero-traffic" in args for args in cloud.calls) + + +@pytest.mark.parametrize("containers", [[], [{}, {}], None]) +def test_missing_or_multiple_containers_do_not_guess(tmp_path, containers): + cloud = FakeCloud() + cloud.revisions[REVISION] = {"spec": {"containers": containers}} + report, _, _ = observe(tmp_path, cloud) + assert report["services"][0]["same_bot_as_contract"] is None + assert not any(f"--secret={BOUND_SECRET}" in args for args in cloud.calls) + + +@pytest.mark.parametrize("env", [ + [], [{"name": "TG_TOKEN", "value": OTHER_TOKEN}], [secret_env(), secret_env()], + [{"name": "TELEGRAM_TOKEN"}], [{"name": "TELEGRAM_TOKEN", "value": ""}], + [{"name": "TELEGRAM_TOKEN", "value": 7}], + [{"name": "TELEGRAM_TOKEN", "value": OTHER_TOKEN, + "valueFrom": {"secretKeyRef": {"name": BOUND_SECRET, "key": VERSION}}}], + [secret_env(version="latest")], [secret_env(version="current")], [secret_env(version=None)], + [secret_env(version=7)], [secret_env(version="0")], [secret_env(version="-1")], + [secret_env(name="")], [secret_env(name="PRIVATE_BAD_SECRET/path")], +]) +def test_missing_ambiguous_or_unpinned_winner_is_unknown_without_fallback(tmp_path, env): + cloud = FakeCloud() + cloud.revisions[REVISION] = revision(env) + report, _, telegram = observe(tmp_path, cloud) + assert report["services"][0]["same_bot_as_contract"] is None + assert telegram.calls == [CONTRACT_TOKEN] + + +@pytest.mark.parametrize("name,annotations,secret_project", [ + (f"projects/{PROJECT}/secrets/{BOUND_SECRET}", {}, PROJECT), + ("private-alias", {"run.googleapis.com/secrets": f"private-alias:projects/{PROJECT}/secrets/{BOUND_SECRET}"}, PROJECT), + ("private-alias", {"run.googleapis.com/secrets": f"private-alias:projects/synthetic-other/secrets/{BOUND_SECRET}"}, + "synthetic-other"), +]) +def test_qualified_and_annotated_secret_reference_keeps_exact_project_version(tmp_path, name, annotations, secret_project): + cloud = FakeCloud() + cloud.revisions[REVISION] = revision([secret_env(name=name)], annotations=annotations) + cloud.secrets[(secret_project, BOUND_SECRET, VERSION)] = OTHER_TOKEN + report, _, _ = observe(tmp_path, cloud) + assert report["services"][0]["same_bot_as_contract"] is True + assert ("secrets", "versions", "access", VERSION, f"--secret={BOUND_SECRET}", + f"--project={secret_project}") in cloud.calls + + +@pytest.mark.parametrize("annotations", [ + {"run.googleapis.com/secrets": "private-alias:PRIVATE_BAD_PATH"}, + {"run.googleapis.com/secrets": "private-alias:projects/synthetic-project/secrets/a,private-alias:projects/synthetic-project/secrets/b"}, + {"run.googleapis.com/secrets": {"private-alias": "PRIVATE_BAD_VALUE"}}, +]) +def test_invalid_or_duplicate_secret_alias_cannot_fall_back_to_same_project_name(tmp_path, annotations): + cloud = FakeCloud() + cloud.secrets[(PROJECT, "PRIVATE_BAD_PATH", VERSION)] = OTHER_TOKEN + cloud.revisions[REVISION] = revision([secret_env(name="private-alias")], annotations=annotations) + report, _, telegram = observe(tmp_path, cloud) + assert report["services"][0]["same_bot_as_contract"] is None + assert telegram.calls == [CONTRACT_TOKEN] + + +@pytest.mark.parametrize("stage", ["service", "revision", "bound_secret", "contract_secret", "bound_getme", "contract_getme"]) +def test_private_command_and_network_failures_are_nullable_and_redacted(tmp_path, stage): + cloud, telegram = FakeCloud(), FakeTelegram() + if stage == "service": + cloud.fail.add(("service", SERVICE)) + elif stage == "revision": + cloud.fail.add(("revision", REVISION)) + elif stage == "bound_secret": + cloud.fail.add(("secret", PROJECT, BOUND_SECRET, VERSION)) + elif stage == "contract_secret": + cloud.fail.add(("secret", PROJECT, route.CONTRACT_SECRET, "latest")) + else: + telegram.results[BOUND_TOKEN if stage == "bound_getme" else CONTRACT_TOKEN] = RuntimeError("PRIVATE_EXCEPTION_URL") + report, _, _ = observe(tmp_path, cloud, telegram) + assert report["services"][0]["same_bot_as_contract"] is None + + +@pytest.mark.parametrize("body", [ + b"{broken", [], {}, {"ok": False}, {"ok": True, "result": None}, + {"ok": True, "result": {"id": True, "is_bot": True}}, + {"ok": True, "result": {"id": "98765001", "is_bot": True}}, + {"ok": True, "result": {"id": -1, "is_bot": True}}, + {"ok": True, "result": {"id": CONTRACT_BOT, "is_bot": False}}, + {"ok": True, "result": {"id": CONTRACT_BOT}}, + b"x" * (64 * 1024 + 1), +]) +def test_invalid_getme_payload_is_unknown_not_false_or_match(tmp_path, body): + telegram = FakeTelegram() + telegram.results[BOUND_TOKEN] = body + report, _, _ = observe(tmp_path, telegram=telegram) + assert report["services"][0]["same_bot_as_contract"] is None + + +def test_plain_env_single_container_can_be_compared_without_secret_value_output(tmp_path): + cloud = FakeCloud() + cloud.revisions[REVISION] = revision([{"name": "TELEGRAM_TOKEN", "value": OTHER_TOKEN}]) + report, _, _ = observe(tmp_path, cloud) + assert report["services"][0]["same_bot_as_contract"] is True + assert report["services"][0]["revisions"][0]["token_source"] == "plain_env" + + +def test_primary_token_failure_is_not_rescued_by_secondary_token(tmp_path): + cloud, telegram = FakeCloud(), FakeTelegram() + cloud.revisions[REVISION] = revision([ + {"name": "TELEGRAM_TOKEN", "value": BOUND_TOKEN}, {"name": "TG_TOKEN", "value": OTHER_TOKEN}, + ]) + telegram.results[BOUND_TOKEN] = RuntimeError("PRIVATE_WINNER_FAILED") + report, _, _ = observe(tmp_path, cloud, telegram) + assert report["services"][0]["same_bot_as_contract"] is None + assert OTHER_TOKEN not in telegram.calls + + +@pytest.mark.parametrize("env_overrides", [ + {"CLOUD_RUN_SERVICE": ""}, {"CLOUD_RUN_SERVICE_TARGETS_JSON": "{broken"}, + {"CLOUD_RUN_SERVICE_TARGETS_JSON": '{"targets":[{"runtime_target":7}]}'}, + {"CLOUD_RUN_SERVICE_TARGETS_JSON": '{"targets":[{}]}'}, + {"CLOUD_RUN_SERVICE_TARGETS_JSON": '{"targets":"PRIVATE_BAD_TARGETS"}'}, + {"GCP_PROJECT_ID": ""}, {"GCP_PROJECT_ID": "PRIVATE_BAD_PROJECT/path"}, +]) +def test_missing_invalid_config_and_target_mapping_are_fixed_unknown(tmp_path, env_overrides): + report, cloud, _ = observe(tmp_path, **env_overrides) + assert report["status"] == "config_unknown" + assert report["services"] == [] + assert cloud.calls == [] + + +def test_target_selection_keeps_existing_sources_and_deduplicates_without_disclosing_names(tmp_path): + report, cloud, _ = observe( + tmp_path, CLOUD_RUN_SERVICES=SERVICE, RUNTIME_GUARD_CLOUD_RUN_SERVICES=SERVICE, + CLOUD_RUN_SERVICE_TARGETS_JSON=json.dumps({"targets": [ + {"runtime_target": {"service_name": SERVICE}}, SERVICE, + ]}), + ) + assert len(report["services"]) == 1 + assert sum(args[:3] == ("run", "services", "describe") for args in cloud.calls) == 1 + + +@pytest.mark.parametrize("stage", ["service", "revision"]) +def test_malformed_metadata_cannot_escape_as_private_traceback(tmp_path, stage): + cloud = FakeCloud() + if stage == "service": + cloud.service = "{PRIVATE_BAD_SERVICE_JSON" + else: + cloud.revisions[REVISION] = "{PRIVATE_BAD_REVISION_JSON" + report, _, _ = observe(tmp_path, cloud) + assert report["services"][0]["same_bot_as_contract"] is None + + +def test_subprocess_cloud_command_captures_both_streams_and_has_bounded_timeout(): + with mock.patch.object(route.subprocess, "run", return_value=subprocess.CompletedProcess( + ["gcloud"], 0, stdout="PRIVATE_RESULT", stderr="PRIVATE_ERROR")) as run: + assert route._gcloud("run", "services", "describe", SERVICE) == "PRIVATE_RESULT" + kwargs = run.call_args.kwargs + assert kwargs["capture_output"] is True + assert kwargs["text"] is True + assert 0 < kwargs["timeout"] <= 60 + + +def test_invalid_source_sha_does_not_publish_untrusted_metadata(tmp_path): + report, _, _ = observe(tmp_path, SOURCE_SHA="PRIVATE_SOURCE_VALUE") + assert report["source_sha"] is None + + +def test_workflow_protects_existing_inputs_before_headers_and_keeps_manual_main_only(): + caller = (ROOT / ".github/workflows/notify-route-check.yml").read_text() + protected_path = ROOT / ".github/workflows/notify-route-check-protected.yml" + assert protected_path.exists() + protected = protected_path.read_text() + assert "workflow_dispatch:" in caller and "schedule:" not in caller + assert "uses: ./.github/workflows/notify-route-check-protected.yml" in caller + assert "github.ref == 'refs/heads/main'" in caller + assert "github.event_name == 'workflow_dispatch'" in caller + assert "\n env:" not in caller and "\n with:" not in caller + assert "persist-credentials: false" in protected + for name in ["GCP_PROJECT_ID", "GCP_WORKLOAD_IDENTITY_PROVIDER", "GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT", + "RUNTIME_GUARD_CLOUD_RUN_SERVICES", "CLOUD_RUN_SERVICE", "CLOUD_RUN_SERVICES", + "CLOUD_RUN_SERVICE_TARGETS_JSON", "CLOUD_RUN_REGION", "TELEGRAM_TOKEN_SECRET_NAME"]: + assert re.search(r"^ " + name + r":", caller, re.MULTILINE) + assert re.search(r"^ " + name + r":", protected, re.MULTILINE) + assert " " + name + ": ${{ secrets." + name + " }}" in protected + assert "SOURCE_SHA: ${{ github.sha }}" in protected + assert "::add-mask::" not in protected + assert "run: python3 scripts/notify_route_check.py" in protected diff --git a/tests/test_workflow_json_secret_boundary.py b/tests/test_workflow_json_secret_boundary.py index 8216e65..b5cf3b0 100644 --- a/tests/test_workflow_json_secret_boundary.py +++ b/tests/test_workflow_json_secret_boundary.py @@ -12,6 +12,28 @@ "env-sync-drift-check", "notify-route-check", "collect-reconciliation-evidence", ) +# The route observer forwards these existing same-value settings as reusable +# secrets so runner headers mask them; it does not create repository secrets. +# This explicit contract is limited to this caller's six new fields. +NOTIFY_ROUTE_ADDITIONAL_SECRET_CONTRACTS = { + "GCP_PROJECT_ID": (True, "'interactivebrokersquant'"), + "GCP_WORKLOAD_IDENTITY_PROVIDER": ( + True, "'projects/303168642265/locations/global/workloadIdentityPools/github-actions/providers/github-main'", + ), + "GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT": ( + True, "'ibkr-platform-deploy@interactivebrokersquant.iam.gserviceaccount.com'", + ), + "RUNTIME_GUARD_CLOUD_RUN_SERVICES": (False, "vars.RUNTIME_GUARD_CLOUD_RUN_SERVICES"), + "CLOUD_RUN_REGION": (False, "vars.CLOUD_RUN_REGION"), + "TELEGRAM_TOKEN_SECRET_NAME": (False, "vars.TELEGRAM_TOKEN_SECRET_NAME"), +} + + +def _assert_notify_route_additional_secret_contract(key: str, declared: str, passed: str) -> None: + required, expression = NOTIFY_ROUTE_ADDITIONAL_SECRET_CONTRACTS[key] + assert f" {key}:\n required: {str(required).lower()}" in declared + assert f" {key}: ${{{{ {expression} }}}}" in passed + def _block(text: str, key: str, indent: int) -> str: match = re.search(rf"(?m)^{' ' * indent}{re.escape(key)}:\s*\n", text) @@ -68,9 +90,12 @@ def test_json_crosses_secret_boundary_before_any_execution(name: str) -> None: referenced = set(re.findall(r"secrets\.([A-Z][A-Z0-9_]*)", _block(callee, "jobs", 0))) assert referenced for key in referenced: - assert f" {key}:\n required: false" in declared - value = f"vars.{key} || secrets.{key}" if key in PROTECTED else f"secrets.{key}" - assert f" {key}: ${{{{ {value} }}}}" in passed + if name == "notify-route-check" and key in NOTIFY_ROUTE_ADDITIONAL_SECRET_CONTRACTS: + _assert_notify_route_additional_secret_contract(key, declared, passed) + else: + assert f" {key}:\n required: false" in declared + value = f"vars.{key} || secrets.{key}" if key in PROTECTED else f"secrets.{key}" + assert f" {key}: ${{{{ {value} }}}}" in passed for key in PROTECTED: assert f"vars.{key}" not in callee if key in referenced: @@ -79,6 +104,21 @@ def test_json_crosses_secret_boundary_before_any_execution(name: str) -> None: assert not _unsafe_json_env(callee) +@pytest.mark.parametrize("key", NOTIFY_ROUTE_ADDITIONAL_SECRET_CONTRACTS) +@pytest.mark.parametrize("mutation", ("required", "source")) +def test_route_additional_contract_rejects_changed_required_or_source(key: str, mutation: str) -> None: + required, expression = NOTIFY_ROUTE_ADDITIONAL_SECRET_CONTRACTS[key] + declared = f" {key}:\n required: {str(required).lower()}\n" + passed = f" {key}: ${{{{ {expression} }}}}\n" + _assert_notify_route_additional_secret_contract(key, declared, passed) + if mutation == "required": + declared = f" {key}:\n required: {str(not required).lower()}\n" + else: + passed = f" {key}: ${{{{ vars.UNRELATED_PRIVATE_SETTING }}}}\n" + with pytest.raises(AssertionError): + _assert_notify_route_additional_secret_contract(key, declared, passed) + + def test_all_workflows_reject_variable_backed_json_in_env() -> None: for path in ROOT.glob("*.yml"): assert not _unsafe_json_env(path.read_text()), path.name