Harden private Observatory staging preflight

.github/workflows/gcp-observatory-read-adapter.yml

@@ -34,7 +34,7 @@ env:
   DB_IAM_USER: sa-observatory-read-adapter@teleo-501523.iam
   API_KEY_SECRET: observatory-read-api-key-staging
   BUILD_WORKLOAD_IDENTITY_PROVIDER: projects/785938879453/locations/global/workloadIdentityPools/github-actions/providers/living-ip-github
-  DEPLOY_WORKLOAD_IDENTITY_PROVIDER: projects/785938879453/locations/global/workloadIdentityPools/github-actions/providers/living-ip-github
+  DEPLOY_WORKLOAD_IDENTITY_PROVIDER: projects/785938879453/locations/global/workloadIdentityPools/github-actions/providers/observatory-read-adapter-main
   EXPECTED_WORKFLOW_REF: living-ip/teleo-infrastructure/.github/workflows/gcp-observatory-read-adapter.yml@refs/heads/main
 
 jobs:
@@ -55,8 +55,16 @@ jobs:
       - name: Run focused adapter tests
         run: |
           python -m pip install -e '.[dev]'
-          python -m pytest tests/test_observatory_read_adapter.py -q
+          python -m pytest \
-          python -m ruff check observatory_read_adapter tests/test_observatory_read_adapter.py
+            tests/test_observatory_read_adapter.py \
+            tests/test_observatory_read_adapter_gcp_plan.py \
+            -q
+          python -m ruff check \
+            observatory_read_adapter \
+            ops/check_observatory_read_adapter_gcp_preflight.py \
+            ops/plan_observatory_read_adapter_gcp.py \
+            tests/test_observatory_read_adapter.py \
+            tests/test_observatory_read_adapter_gcp_plan.py
 
       - id: auth
         uses: google-github-actions/auth@v3
@@ -106,6 +114,12 @@ jobs:
     steps:
       - uses: actions/checkout@v4
 
+      - name: Initialize fail-closed preflight receipt
+        run: |
+          python3 ops/check_observatory_read_adapter_gcp_preflight.py \
+            --initialize-fail-closed \
+            --output observatory-read-adapter-preflight.json
+
       - name: Enforce reviewed main-only staging path
         shell: bash
         run: |

docs/observatory-read-adapter-gcp-staging.md

@@ -84,6 +84,10 @@ they do not route production traffic.
    `run.operations.get`; it contains no delete permission. Do not add
    `run.admin`, `run.developer`, `cloudsql.admin`, `secretmanager.admin`,
    `compute.admin`, Editor, or Owner to either lane identity.
+   The deployer binding uses the provider-unique
+   `attribute.observatory_workflow` principal set. A pool-wide subject binding
+   or the general `living-ip-github` provider is diagnostic-only and must not
+   authenticate the preflight or deploy jobs.
 3. Add the dedicated runtime service account as a Cloud SQL IAM
    service-account user on `teleo-pgvector-standby`.
 4. From the existing private GCP VM database-admin path, run

ops/check_observatory_read_adapter_gcp_preflight.py

@@ -26,6 +26,9 @@ SECRET = "observatory-read-api-key-staging"
 REPOSITORY = "teleo"
 WIF_POOL = "github-actions"
 DEPLOY_WIF_PROVIDER = "observatory-read-adapter-main"
+DEPLOY_WIF_PROVIDER_RESOURCE = (
+    f"projects/{PROJECT_NUMBER}/locations/global/workloadIdentityPools/{WIF_POOL}/providers/{DEPLOY_WIF_PROVIDER}"
+)
 DEPLOY_WORKFLOW_REF = (
     "living-ip/teleo-infrastructure/.github/workflows/"
     "gcp-observatory-read-adapter.yml@refs/heads/main"
@@ -36,9 +39,8 @@ DEPLOY_WIF_ATTRIBUTE_CONDITION = (
     f"assertion.workflow_ref=='{DEPLOY_WORKFLOW_REF}'"
 )
 DEPLOY_WIF_MEMBER = (
-    f"principal://iam.googleapis.com/projects/{PROJECT_NUMBER}/locations/global/"
+    f"principalSet://iam.googleapis.com/projects/{PROJECT_NUMBER}/locations/global/"
-    f"workloadIdentityPools/{WIF_POOL}/subject/"
+    f"workloadIdentityPools/{WIF_POOL}/attribute.observatory_workflow/{DEPLOY_WORKFLOW_REF}"
-    "repo:living-ip/teleo-infrastructure:ref:refs/heads/main"
 )
 PREFLIGHT_ROLE = f"projects/{PROJECT}/roles/observatoryStagingPreflight"
 DEPLOY_ROLE = f"projects/{PROJECT}/roles/observatoryStagingDeployer"
@@ -215,6 +217,22 @@ def member_roles(policy: dict[str, object], member: str) -> set[str]:
     return roles
 
 
+def role_has_exact_unconditional_members(
+    policy: dict[str, object],
+    role: str,
+    expected_members: set[str],
+) -> bool:
+    bindings = policy.get("bindings", [])
+    if not isinstance(bindings, list):
+        return False
+    role_bindings = [binding for binding in bindings if isinstance(binding, dict) and binding.get("role") == role]
+    if len(role_bindings) != 1:
+        return False
+    binding = role_bindings[0]
+    members = binding.get("members", [])
+    return isinstance(members, list) and set(members) == expected_members and binding.get("condition") in (None, {})
+
+
 def project_iam_is_exact(value: str) -> bool:
     policy = json.loads(value)
     deployer = f"serviceAccount:{DEPLOY_SERVICE_ACCOUNT}"
@@ -277,6 +295,7 @@ def wif_provider_is_exact(value: str) -> bool:
     mapping = payload.get("attributeMapping", {})
     expected_mapping = {
         "google.subject": "assertion.sub",
+        "attribute.observatory_workflow": "assertion.workflow_ref",
         "attribute.repository": "assertion.repository",
         "attribute.ref": "assertion.ref",
         "attribute.workflow_ref": "assertion.workflow_ref",
@@ -290,7 +309,11 @@ def wif_provider_is_exact(value: str) -> bool:
 
 def deployer_service_account_policy_is_exact(value: str) -> bool:
     policy = json.loads(value)
-    return policy_has_binding(policy, "roles/iam.workloadIdentityUser", DEPLOY_WIF_MEMBER)
+    return role_has_exact_unconditional_members(
+        policy,
+        "roles/iam.workloadIdentityUser",
+        {DEPLOY_WIF_MEMBER},
+    )
 
 
 def runtime_service_account_policy_is_exact(value: str) -> bool:
@@ -508,7 +531,7 @@ def build_checks(image_ref: str) -> list[Check]:
                     "--format=json",
                 ],
                 deployer_service_account_policy_is_exact,
-                "exact_main_subject_workload_identity_user",
+                "exact_workflow_attribute_principal_set_only",
             ),
             command_check(
                 "runtime_act_as_policy",
@@ -633,23 +656,23 @@ def build_checks(image_ref: str) -> list[Check]:
     return checks
 
 
-def main() -> int:
+def build_receipt(
-    parser = argparse.ArgumentParser()
+    checks: list[Check],
-    parser.add_argument("--image-ref", required=True)
+    *,
-    parser.add_argument("--output", type=Path)
+    current_tier: str,
-    args = parser.parse_args()
+    claim_ceiling: str,
-
+) -> dict[str, object]:
-    checks = build_checks(args.image_ref)
     pass_count = sum(check.status == "pass" for check in checks)
     blocked_count = len(checks) - pass_count
-    receipt = {
+    return {
         "schema": "livingip.observatory-read-adapter-gcp-preflight.v1",
         "project": PROJECT,
         "required_tier": "T3_live_readonly",
-        "current_tier": "T2_authenticated_gcp_preflight",
+        "current_tier": current_tier,
         "ready_for_staging_deploy": blocked_count == 0,
-        "claim_ceiling": "Preflight only; T3 requires the live private Cloud SQL claim and negative receipts.",
+        "claim_ceiling": claim_ceiling,
         "identity": DEPLOY_SERVICE_ACCOUNT,
+        "workload_identity_provider": DEPLOY_WIF_PROVIDER_RESOURCE,
         "checks": [asdict(check) for check in checks],
         "pass_count": pass_count,
         "blocked_count": blocked_count,
@@ -657,12 +680,49 @@ def main() -> int:
         "database_role_live_verification_deferred_to_fail_closed_canary": True,
         "production_repoint_executed": False,
     }
+
+
+def write_receipt(receipt: dict[str, object], output: Path | None) -> None:
     rendered = json.dumps(receipt, indent=2, sort_keys=True) + "\n"
-    if args.output:
+    if output:
-        args.output.write_text(rendered, encoding="utf-8")
+        output.write_text(rendered, encoding="utf-8")
     else:
         print(rendered, end="")
-    return 0 if blocked_count == 0 else 1
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    mode = parser.add_mutually_exclusive_group(required=True)
+    mode.add_argument("--image-ref")
+    mode.add_argument("--initialize-fail-closed", action="store_true")
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+
+    if args.initialize_fail_closed:
+        receipt = build_receipt(
+            [
+                Check(
+                    "identity_authentication_and_gcp_preflight",
+                    "blocked",
+                    "dedicated staging identity authentication and GCP preflight did not complete",
+                )
+            ],
+            current_tier="T2_identity_preflight_not_completed",
+            claim_ceiling=(
+                "Fail-closed initialization only; no authenticated GCP preflight or staging deploy has completed."
+            ),
+        )
+        write_receipt(receipt, args.output)
+        return 0
+
+    checks = build_checks(args.image_ref)
+    receipt = build_receipt(
+        checks,
+        current_tier="T2_authenticated_gcp_preflight",
+        claim_ceiling="Preflight only; T3 requires the live private Cloud SQL claim and negative receipts.",
+    )
+    write_receipt(receipt, args.output)
+    return 0 if receipt["ready_for_staging_deploy"] else 1
 
 
 if __name__ == "__main__":

ops/plan_observatory_read_adapter_gcp.py

@@ -13,12 +13,16 @@ REGION = "europe-west6"
 GITHUB_REPOSITORY = "living-ip/teleo-infrastructure"
 WIF_POOL = "github-actions"
 DEPLOY_WIF_PROVIDER = "observatory-read-adapter-main"
+DEPLOY_WIF_PROVIDER_RESOURCE = (
+    f"projects/{PROJECT_NUMBER}/locations/global/workloadIdentityPools/{WIF_POOL}/providers/{DEPLOY_WIF_PROVIDER}"
+)
 DEPLOY_WORKFLOW_REF = (
     f"{GITHUB_REPOSITORY}/.github/workflows/gcp-observatory-read-adapter.yml@refs/heads/main"
 )
 DEPLOY_WIF_ATTRIBUTE_MAPPING = (
     "google.subject=assertion.sub,attribute.repository=assertion.repository,"
-    "attribute.ref=assertion.ref,attribute.workflow_ref=assertion.workflow_ref"
+    "attribute.ref=assertion.ref,attribute.workflow_ref=assertion.workflow_ref,"
+    "attribute.observatory_workflow=assertion.workflow_ref"
 )
 DEPLOY_WIF_ATTRIBUTE_CONDITION = (
     "assertion.repository=='living-ip/teleo-infrastructure' && "
@@ -93,6 +97,13 @@ def shell_join(parts: list[str]) -> str:
 
 
 def github_wif_member() -> str:
+    return (
+        f"principalSet://iam.googleapis.com/projects/{PROJECT_NUMBER}/locations/global/"
+        f"workloadIdentityPools/{WIF_POOL}/attribute.observatory_workflow/{DEPLOY_WORKFLOW_REF}"
+    )
+
+
+def legacy_pool_wide_github_wif_member() -> str:
     return (
         f"principal://iam.googleapis.com/projects/{PROJECT_NUMBER}/locations/global/"
         f"workloadIdentityPools/{WIF_POOL}/subject/repo:{GITHUB_REPOSITORY}:ref:refs/heads/main"
@@ -268,6 +279,35 @@ def build_plan() -> dict[str, object]:
     )
     commands.extend(
         [
+            (
+                "DEPLOYER_POLICY_JSON=$(gcloud iam service-accounts get-iam-policy "
+                + shlex.quote(DEPLOY_SERVICE_ACCOUNT)
+                + " --project "
+                + shlex.quote(PROJECT)
+                + " --format=json)"
+            ),
+            (
+                "if jq -e --arg legacy "
+                + shlex.quote(legacy_pool_wide_github_wif_member())
+                + " 'any(.bindings[]?; .role == \"roles/iam.workloadIdentityUser\" "
+                "and ((.members // []) | index($legacy)))' <<<\"${DEPLOYER_POLICY_JSON}\" >/dev/null; then "
+                + shell_join(
+                    [
+                        "gcloud",
+                        "iam",
+                        "service-accounts",
+                        "remove-iam-policy-binding",
+                        DEPLOY_SERVICE_ACCOUNT,
+                        "--project",
+                        PROJECT,
+                        "--member",
+                        legacy_pool_wide_github_wif_member(),
+                        "--role",
+                        "roles/iam.workloadIdentityUser",
+                    ]
+                )
+                + "; fi"
+            ),
             project_binding(deployer_member, PREFLIGHT_ROLE),
             project_binding(deployer_member, DEPLOY_ROLE),
             shell_join(
@@ -398,10 +438,16 @@ def build_plan() -> dict[str, object]:
             "deployer": "workflow-specific main-only WIF plus a five-permission Cloud Run role, repository read, exact runtime actAs, exact secret access",
             "runtime": "exact Cloud SQL instance connect/login plus exact secret access",
         },
-        "github_wif_provider": (
+        "github_wif_provider": DEPLOY_WIF_PROVIDER_RESOURCE,
-            f"projects/{PROJECT_NUMBER}/locations/global/workloadIdentityPools/{WIF_POOL}/providers/{DEPLOY_WIF_PROVIDER}"
-        ),
         "github_wif_member": github_wif_member(),
+        "legacy_pool_wide_github_wif_member_removed": legacy_pool_wide_github_wif_member(),
+        "workflow_authentication": {
+            "provider": DEPLOY_WIF_PROVIDER_RESOURCE,
+            "service_account": DEPLOY_SERVICE_ACCOUNT,
+            "principal_set": github_wif_member(),
+            "fallback_provider_allowed": False,
+            "required_condition": DEPLOY_WIF_ATTRIBUTE_CONDITION,
+        },
         "required_apis": list(REQUIRED_APIS),
         "preflight_custom_role": {
             "name": PREFLIGHT_ROLE,
@@ -446,6 +492,7 @@ def build_plan() -> dict[str, object]:
         "forbidden_deployer_roles": list(FORBIDDEN_DEPLOYER_ROLES),
         "forbidden_actions": [
             "grant deploy or infra roles to sa-artifact-builder",
+            "authenticate the deploy or canary jobs through the general living-ip-github provider",
             "reuse a pre-existing image tag instead of building the current workflow revision",
             "add a public Cloud SQL address",
             "expose a secret value in Git, logs, or a receipt",

tests/test_observatory_read_adapter_gcp_plan.py

@@ -38,12 +38,24 @@ def test_plan_splits_build_deploy_and_runtime_identities() -> None:
     assert "roles/cloudsql.instanceUser" in commands
     assert "roles/artifactregistry.reader" in commands
     assert "roles/secretmanager.secretAccessor" in commands
-    assert "subject/repo:living-ip/teleo-infrastructure:ref:refs/heads/main" in plan["github_wif_member"]
+    assert (
+        "attribute.observatory_workflow/living-ip/teleo-infrastructure/.github/workflows/"
+        "gcp-observatory-read-adapter.yml@refs/heads/main"
+    ) in plan["github_wif_member"]
     assert plan["conditions"]["deployer_wif"].endswith(
         "assertion.workflow_ref=='living-ip/teleo-infrastructure/.github/workflows/"
         "gcp-observatory-read-adapter.yml@refs/heads/main'"
     )
     assert plan["github_wif_provider"].endswith("/providers/observatory-read-adapter-main")
+    assert plan["workflow_authentication"] == {
+        "provider": plan["github_wif_provider"],
+        "service_account": "sa-observatory-deployer@teleo-501523.iam.gserviceaccount.com",
+        "principal_set": plan["github_wif_member"],
+        "fallback_provider_allowed": False,
+        "required_condition": plan["conditions"]["deployer_wif"],
+    }
+    assert "remove-iam-policy-binding" in commands
+    assert plan["legacy_pool_wide_github_wif_member_removed"] in commands
     assert "roles/run.serviceAgent" in commands
     assert "teleo-pgvector-standby" in plan["conditions"]["cloud_sql"]
     assert set(plan["deploy_custom_role"]["permissions"]) == {
@@ -84,7 +96,10 @@ def test_workflow_requires_dedicated_preflight_before_staging_deploy() -> None:
 
     assert workflow["env"]["BUILD_SERVICE_ACCOUNT"].startswith("sa-artifact-builder@")
     assert workflow["env"]["DEPLOY_SERVICE_ACCOUNT"].startswith("sa-observatory-deployer@")
-    assert workflow["env"]["DEPLOY_WORKLOAD_IDENTITY_PROVIDER"].endswith("/providers/living-ip-github")
+    assert workflow["env"]["DEPLOY_WORKLOAD_IDENTITY_PROVIDER"].endswith(
+        "/providers/observatory-read-adapter-main"
+    )
+    assert workflow["env"]["DEPLOY_WORKLOAD_IDENTITY_PROVIDER"] == preflight.DEPLOY_WIF_PROVIDER_RESOURCE
     assert workflow["jobs"]["deploy-staging"]["needs"] == ["build", "preflight-staging"]
     assert "action=preflight_staging" not in text
     assert "inputs.action != 'build_only'" in text
@@ -97,6 +112,8 @@ def test_workflow_requires_dedicated_preflight_before_staging_deploy() -> None:
     assert "API_KEY_VERSION" in text
     assert "EXPECTED_IMAGE_REF" in text
     assert "check_observatory_read_adapter_gcp_preflight.py" in text
+    assert "--initialize-fail-closed" in text
+    assert text.index("--initialize-fail-closed") < text.index("workload_identity_provider: ${{ env.DEPLOY_WORKLOAD_IDENTITY_PROVIDER }}")
     assert 'service_account: ${{ env.BUILD_SERVICE_ACCOUNT }}' in text
     assert text.count('service_account: ${{ env.DEPLOY_SERVICE_ACCOUNT }}') == 2
     assert "positive-redacted.json" in text
@@ -107,6 +124,37 @@ def test_workflow_requires_dedicated_preflight_before_staging_deploy() -> None:
     assert "rm -f positive.json" in text
 
 
+def test_fail_closed_preflight_initialization_retains_a_blocked_receipt(tmp_path: Path) -> None:
+    output = tmp_path / "preflight.json"
+    subprocess.run(
+        [
+            "python3",
+            "ops/check_observatory_read_adapter_gcp_preflight.py",
+            "--initialize-fail-closed",
+            "--output",
+            str(output),
+        ],
+        cwd=ROOT,
+        check=True,
+    )
+
+    receipt = json.loads(output.read_text(encoding="utf-8"))
+    assert receipt["ready_for_staging_deploy"] is False
+    assert receipt["current_tier"] == "T2_identity_preflight_not_completed"
+    assert receipt["identity"] == "sa-observatory-deployer@teleo-501523.iam.gserviceaccount.com"
+    assert receipt["workload_identity_provider"].endswith("/providers/observatory-read-adapter-main")
+    assert receipt["pass_count"] == 0
+    assert receipt["blocked_count"] == 1
+    assert receipt["checks"] == [
+        {
+            "name": "identity_authentication_and_gcp_preflight",
+            "status": "blocked",
+            "detail": "dedicated staging identity authentication and GCP preflight did not complete",
+        }
+    ]
+    assert receipt["secret_values_retained"] is False
+
+
 def test_preflight_private_cloud_sql_and_secret_validators() -> None:
     instance = {
         "name": "teleo-pgvector-standby",
@@ -225,6 +273,7 @@ def test_preflight_end_to_end_snapshot_retains_no_secret(monkeypatch) -> None:
         "attributeCondition": preflight.DEPLOY_WIF_ATTRIBUTE_CONDITION,
         "attributeMapping": {
             "google.subject": "assertion.sub",
+            "attribute.observatory_workflow": "assertion.workflow_ref",
             "attribute.repository": "assertion.repository",
             "attribute.ref": "assertion.ref",
             "attribute.workflow_ref": "assertion.workflow_ref",
@@ -304,6 +353,42 @@ def test_preflight_end_to_end_snapshot_retains_no_secret(monkeypatch) -> None:
     assert secret_value not in json.dumps([check.__dict__ for check in checks])
 
 
+def test_deployer_wif_binding_rejects_pool_wide_or_additional_impersonators() -> None:
+    exact = {
+        "bindings": [
+            {
+                "role": "roles/iam.workloadIdentityUser",
+                "members": [preflight.DEPLOY_WIF_MEMBER],
+            }
+        ]
+    }
+    assert preflight.deployer_service_account_policy_is_exact(json.dumps(exact)) is True
+
+    old_pool_wide_subject = (
+        "principal://iam.googleapis.com/projects/785938879453/locations/global/"
+        "workloadIdentityPools/github-actions/subject/"
+        "repo:living-ip/teleo-infrastructure:ref:refs/heads/main"
+    )
+    exact["bindings"][0]["members"].append(old_pool_wide_subject)
+    assert preflight.deployer_service_account_policy_is_exact(json.dumps(exact)) is False
+
+
+def test_dedicated_provider_requires_provider_unique_workflow_mapping() -> None:
+    provider = {
+        "state": "ACTIVE",
+        "attributeCondition": preflight.DEPLOY_WIF_ATTRIBUTE_CONDITION,
+        "attributeMapping": {
+            "google.subject": "assertion.sub",
+            "attribute.repository": "assertion.repository",
+            "attribute.ref": "assertion.ref",
+            "attribute.workflow_ref": "assertion.workflow_ref",
+        },
+    }
+    assert preflight.wif_provider_is_exact(json.dumps(provider)) is False
+    provider["attributeMapping"]["attribute.observatory_workflow"] = "assertion.workflow_ref"
+    assert preflight.wif_provider_is_exact(json.dumps(provider)) is True
+
+
 def test_malformed_successful_readback_becomes_blocked_check(monkeypatch) -> None:
     monkeypatch.setattr(
         preflight,