diff --git a/.github/workflows/apply-paper-scheduler-jobs-get.yml b/.github/workflows/apply-paper-scheduler-jobs-get.yml new file mode 100644 index 0000000..c763e95 --- /dev/null +++ b/.github/workflows/apply-paper-scheduler-jobs-get.yml @@ -0,0 +1,66 @@ +name: Apply PAPER Scheduler jobs.get IAM + +on: + workflow_dispatch: + inputs: + mode: + description: Dry-run first; apply needs a separate exact-revision owner card + required: true + type: choice + options: [dry-run, apply] + default: dry-run + confirm: + description: Must be APPLY_PAPER_SCHEDULER_JOBS_GET + required: true + type: string + +permissions: + contents: read + +# One project IAM writer across invocations; never cancel a write in flight. +# External writers must honor the same owner window; etag CAS detects drift. +concurrency: + group: paper-scheduler-jobs-get-iam + cancel-in-progress: false + +jobs: + bounded-iam: + if: >- + github.event_name == 'workflow_dispatch' + && github.repository == 'QuantStrategyLab/LongBridgePlatform' + && github.ref == 'refs/heads/main' + && github.run_attempt == 1 + && inputs.confirm == 'APPLY_PAPER_SCHEDULER_JOBS_GET' + && (inputs.mode == 'dry-run' || inputs.mode == 'apply') + runs-on: ubuntu-latest + timeout-minutes: 10 + environment: longbridge-paper + permissions: + contents: read + id-token: write + env: + IAM_ENVIRONMENT: longbridge-paper + IAM_MODE: ${{ inputs.mode }} + IAM_CONFIRM: ${{ inputs.confirm }} + GCP_PROJECT_ID: ${{ vars.GCP_PROJECT_ID }} + PAPER_SCHEDULER_GET_PROJECT_ID: ${{ vars.PAPER_SCHEDULER_GET_PROJECT_ID }} + PAPER_SCHEDULER_GET_READER_EMAIL: ${{ vars.PAPER_SCHEDULER_GET_READER_EMAIL }} + steps: + - name: Checkout exact protected revision + uses: actions/checkout@v6 + with: + ref: ${{ github.sha }} + persist-credentials: false + - name: Validate protected PAPER mappings before authentication + run: python3 -c 'import os; from scripts.paper_scheduler_get_iam import identity, role_definition; identity(os.environ); role_definition()' + - name: Authenticate using existing PAPER Environment WIF + uses: google-github-actions/auth@v3 + with: + workload_identity_provider: ${{ vars.GCP_WORKLOAD_IDENTITY_PROVIDER }} + service_account: ${{ vars.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }} + - name: Set up existing cloud client + uses: google-github-actions/setup-gcloud@v3 + with: + project_id: ${{ vars.GCP_PROJECT_ID }} + - name: Plan then apply only the fixed IAM delta and verify readback + run: python3 scripts/paper_scheduler_get_iam.py diff --git a/docs/iam/paper-scheduler-get.md b/docs/iam/paper-scheduler-get.md index 08d8c21..bb2ef33 100644 --- a/docs/iam/paper-scheduler-get.md +++ b/docs/iam/paper-scheduler-get.md @@ -65,3 +65,66 @@ access, deployment, Scheduler enablement or PAPER start is performed by this proposal. PAPER remains stopped; `READY_FOR_START_CARD=false`. Live, Kelly and all stop-loss switches must remain OFF. A future successful IAM readback would prove only that IAM delta, not runtime safety or permission to start PAPER. + +## Restricted Actions entrypoint + +[Apply PAPER Scheduler jobs.get IAM](../../.github/workflows/apply-paper-scheduler-jobs-get.yml) +provides the bounded entrypoint using +[the single-purpose adapter](../../scripts/paper_scheduler_get_iam.py). +This workflow proposal uses authority `t1293s7532=A` for Draft publication only. +The earlier merge/apply authority `t1293s7526=B` ended at +`MERGED_APPLY_BLOCKED_NO_ENTRYPOINT`; it is not reused. A separate owner card +must pin the merged workflow revision and authorize its dry-run and ONE apply. +No dispatch, IAM operation or PAPER start is authorized by this document. + +The only trigger is `workflow_dispatch` on this repository's `main`, first run +attempt only, protected by Environment `longbridge-paper`. Enter exactly +`APPLY_PAPER_SCHEDULER_JOBS_GET`; `mode` defaults to `dry-run`. The concurrency +group `paper-scheduler-jobs-get-iam` queues writers without cancellation. +The single-writer window also requires other project IAM/role writers to stay +out; Actions concurrency cannot lock external clients. + +Existing PAPER Environment variables `GCP_PROJECT_ID`, +`GCP_WORKLOAD_IDENTITY_PROVIDER` and `GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT` +supply existing WIF authentication. Two additional protected mapping variables +must already be verified at the later authorized boundary: +`PAPER_SCHEDULER_GET_PROJECT_ID` must equal `GCP_PROJECT_ID` for the approved +PAPER project, and `PAPER_SCHEDULER_GET_READER_EMAIL` must identify the existing +`grok-gcp-ro` reader. Neither accepts dispatch input. Missing or inconsistent +mappings fail before authentication. This PR neither configures these variables +nor verifies their deployed presence, Environment protection or the executor's +current IAM capabilities. Missing authority or permissions must block the later +operation, never prompt an automatic admin grant or new trust configuration. + +Dry-run reads the role and version-3 project policy; it prints fixed aliases +and the permission list only. Apply repeats those reads, refuses snapshot drift, +creates the role only when absent, and never updates or undeletes an existing +role. An existing role must have exactly `cloudscheduler.jobs.get`, GA stage +and an etag. Conditional, duplicate or extra-member bindings for this role are +collisions and stop the operation. An already-correct role/binding is a no-op. + +Project bindings use the Resource Manager `setIamPolicy` API with the explicit +field mask `bindings,etag` and the freshly read etag. This is a conditional +read-modify-write of the binding field, adding one member/role binding while +preserving every unrelated binding; it is not a whole-policy replacement or +an unconditional policy-file replay. Audit configuration is excluded from the +write. A concurrent policy change produces an etag conflict and stops. Before +and after reads validate exact role permissions, the single existing-reader +binding and unchanged unrelated policy fields. This API cannot atomically +lock a separate custom role: an external role writer remains a reason to +stop and investigate, and the owner must reserve a single-writer window. + +Every HTTP mutation is attempted once, without redirects or automatic retry. +Transport failures, collisions, unreadable results and drift exit non-zero. +If role creation succeeds but the binding fails, the partial result remains; +there is no rollback, automatic second apply or recovery dispatch. Independently +read back remote state under a new owner decision before further action. +A process interruption has the same uncertain-result treatment; re-running an +Actions attempt is gated out. Logs emit fixed categories only, with no remote +payload, project ID, email, token or exception message. + +The adapter has no permission, member, role or command input and exposes no +image, Artifact Registry, Storage, Container Analysis, Secret IAM, Scheduler +mutation, deployment, traffic or PAPER-start operation. Live, Kelly and all +stop-loss switches stay OFF. Image metadata remains +`UNKNOWN_PERMISSION_CLASS` / `OWNER_NEEDED_FOLLOWUP`. diff --git a/scripts/paper_scheduler_get_iam.py b/scripts/paper_scheduler_get_iam.py new file mode 100644 index 0000000..f0fad04 --- /dev/null +++ b/scripts/paper_scheduler_get_iam.py @@ -0,0 +1,190 @@ +"""One bounded PAPER IAM delta. Remote data and diagnostics never leave memory. + +Only fixed status messages are emitted, as in stop_hk_runtime.py. There is no +arbitrary command, project, permission or member argument and no write retry. +""" +import copy +import json +import os +from pathlib import Path +import re +import subprocess +import urllib.error +import urllib.request + +ROLE_ID = "paperSchedulerJobsGet" +PERMISSION = "cloudscheduler.jobs.get" +ROLE_FILE = Path(__file__).resolve().parents[1] / "docs/iam/paper-scheduler-get-role.json" + + +class Blocked(ValueError): + """Fixed failure category; never attach remote payloads or identifiers.""" + + +def role_definition(path=ROLE_FILE): + value = json.loads(path.read_text()) + if (set(value) != {"title", "description", "stage", "includedPermissions"} + or value["includedPermissions"] != [PERMISSION] or value["stage"] != "GA"): + raise Blocked("role_definition_rejected") + return value + + +def identity(env): + required = { + "GITHUB_EVENT_NAME": "workflow_dispatch", + "GITHUB_REPOSITORY": "QuantStrategyLab/LongBridgePlatform", + "GITHUB_REF": "refs/heads/main", "GITHUB_RUN_ATTEMPT": "1", + "IAM_ENVIRONMENT": "longbridge-paper", "IAM_CONFIRM": "APPLY_PAPER_SCHEDULER_JOBS_GET", + } + project = env.get("GCP_PROJECT_ID", "") + reader = env.get("PAPER_SCHEDULER_GET_READER_EMAIL", "") + if (any(env.get(k) != v for k, v in required.items()) + or env.get("IAM_MODE", "dry-run") not in {"dry-run", "apply"} + or not re.fullmatch(r"[a-z][a-z0-9-]{4,28}[a-z0-9]", project) + or project != env.get("PAPER_SCHEDULER_GET_PROJECT_ID") + or not re.fullmatch(r"grok-gcp-ro@[a-z][a-z0-9-]{4,28}[a-z0-9]\.iam\.gserviceaccount\.com", reader)): + raise Blocked("authority_mapping_rejected") + return project, f"serviceAccount:{reader}" + + +def validate_role(role, name): + if (not isinstance(role, dict) or role.get("name") != name + or role.get("includedPermissions") != [PERMISSION] + or role.get("deleted", False) is not False or role.get("stage") != "GA" + or not isinstance(role.get("etag"), str) or not role["etag"]): + raise Blocked("role_collision_or_integrity_failure") + + +def validate_policy(policy, name, member): + if (not isinstance(policy, dict) or not isinstance(policy.get("etag"), str) + or not policy["etag"] or policy.get("version") not in {1, 3} + or not isinstance(policy.get("bindings"), list)): + raise Blocked("policy_integrity_failure") + matches = [] + for binding in policy["bindings"]: + if (not isinstance(binding, dict) or not isinstance(binding.get("role"), str) + or not isinstance(binding.get("members"), list) + or not binding["members"] or any(not isinstance(m, str) for m in binding["members"])): + raise Blocked("policy_integrity_failure") + if binding["role"] == name: + matches.append(binding) + if matches and (len(matches) != 1 or matches[0] != {"role": name, "members": [member]}): + raise Blocked("binding_collision") + return bool(matches) + + +def execute(env, request): + project, member = identity(env) + definition = role_definition() + name = f"projects/{project}/roles/{ROLE_ID}" + role_url = f"https://iam.googleapis.com/v1/{name}" + project_url = f"https://cloudresourcemanager.googleapis.com/v1/projects/{project}" + + def read_policy(): + policy = request("POST", project_url + ":getIamPolicy", {"options": {"requestedPolicyVersion": 3}}) + validate_policy(policy, name, member) + return policy + + before_role = request("GET", role_url, missing=True) + if before_role is not None: + validate_role(before_role, name) + before_policy = read_policy() + already_bound = validate_policy(before_policy, name, member) + if before_role is None and already_bound: + raise Blocked("orphan_binding_collision") + print("before: project=longbridge-paper reader=existing-box-reader role=" + ROLE_ID) + print("plan: permissions=[cloudscheduler.jobs.get]; additive binding only") + if env.get("IAM_MODE", "dry-run") == "dry-run": + print("dry-run: no writes; PAPER remains stopped") + return + + # Re-read both objects immediately before any write. Never adopt a role + # created by another writer after the initial absent snapshot. + current_role = request("GET", role_url, missing=True) + current_policy = read_policy() + if current_role != before_role or current_policy != before_policy: + raise Blocked("concurrent_drift") + if before_role is None: + request("POST", f"https://iam.googleapis.com/v1/projects/{project}/roles", + {"roleId": ROLE_ID, "role": definition}) + created_role = request("GET", role_url) + validate_role(created_role, name) + else: + created_role = before_role + # Role creation and policy mutation are separate APIs, not a transaction. + # A failure can leave the role created with no binding. Stop; never rollback + # or retry. A later owner card must inspect the partial result. + fresh_policy = read_policy() + fresh_role = request("GET", role_url) + validate_role(fresh_role, name) + if fresh_policy != before_policy or fresh_role != created_role: + raise Blocked("concurrent_drift") + if not already_bound: + bindings = copy.deepcopy(fresh_policy["bindings"]) + bindings.append({"role": name, "members": [member]}) + # Google exposes additive project bindings through setIamPolicy. Use + # ONLY the binding field and etag CAS: no whole-policy replacement, + # auditConfig update, stale replay, CLI retry or unconditional write. + request("POST", project_url + ":setIamPolicy", { + "policy": {"version": fresh_policy["version"], "etag": fresh_policy["etag"], "bindings": bindings}, + "updateMask": "bindings,etag", + }) + after_role = request("GET", role_url) + validate_role(after_role, name) + after_policy = read_policy() + expected = copy.deepcopy(before_policy) + if not already_bound: + expected["bindings"].append({"role": name, "members": [member]}) + # Etag changes are expected. All unrelated policy fields must be identical. + if ({k: v for k, v in after_policy.items() if k != "etag"} + != {k: v for k, v in expected.items() if k != "etag"} + or after_role != created_role + or not validate_policy(after_policy, name, member)): + raise Blocked("after_readback_drift") + print("after: permissions=[cloudscheduler.jobs.get]; existing-box-reader binding verified") + print("complete: PAPER remains stopped; no further operation authorized") + + +class Cloud: + """Single-attempt REST adapter. No raw stderr, URL or response is printed.""" + + def __init__(self): + result = subprocess.run(["gcloud", "auth", "print-access-token"], capture_output=True, timeout=60, check=True) + self.token = result.stdout.decode().strip() + if not self.token or "\n" in self.token or "\r" in self.token: + raise Blocked("authentication_failed") + + def __call__(self, method, url, body=None, missing=False): + data = None if body is None else json.dumps(body).encode() + request = urllib.request.Request(url, data=data, method=method, headers={ + "Authorization": "Bearer " + self.token, "Content-Type": "application/json", + }) + # Disable redirects so neither the bearer token nor a mutation can be + # replayed at another endpoint. urllib performs no automatic retries. + class NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + try: + with urllib.request.build_opener(NoRedirect).open(request, timeout=60) as response: + return json.load(response) + except urllib.error.HTTPError as error: + if missing and method == "GET" and error.code == 404: + return None + raise Blocked("remote_rejected_or_unknown_stop_no_retry") from None + except Exception: + raise Blocked("remote_result_unknown_stop_no_retry") from None + + +def main(): + try: + identity(os.environ) # Reject wrong scope before token access. + execute(os.environ, Cloud()) + return 0 + except Exception: + # Fixed allowlisted text; suppress even unexpected exception messages. + print("BLOCKED: verify mappings, permissions and remote state; do not retry an uncertain write") + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_paper_scheduler_get_iam.py b/tests/test_paper_scheduler_get_iam.py new file mode 100644 index 0000000..399cdee --- /dev/null +++ b/tests/test_paper_scheduler_get_iam.py @@ -0,0 +1,190 @@ +"""Offline adverse tests: no cloud clients, tokens or real identifiers.""" +import copy +import importlib.util +import json +from pathlib import Path + +import pytest + +SPEC = importlib.util.spec_from_file_location("paper_iam", Path(__file__).parents[1] / "scripts/paper_scheduler_get_iam.py") +# A missing implementation is the expected fail-before result. +m = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(m) + +PROJECT = "fake-paper-project" +READER = "grok-gcp-ro@fake-reader-project.iam.gserviceaccount.com" +ENV = { + "GITHUB_EVENT_NAME": "workflow_dispatch", "GITHUB_REPOSITORY": "QuantStrategyLab/LongBridgePlatform", + "GITHUB_REF": "refs/heads/main", "GITHUB_RUN_ATTEMPT": "1", + "IAM_ENVIRONMENT": "longbridge-paper", "IAM_CONFIRM": "APPLY_PAPER_SCHEDULER_JOBS_GET", + "IAM_MODE": "apply", "GCP_PROJECT_ID": PROJECT, "PAPER_SCHEDULER_GET_PROJECT_ID": PROJECT, + "PAPER_SCHEDULER_GET_READER_EMAIL": READER, +} + + +class Cloud: + def __init__(self, role=None): + self.role = role + self.policy = {"version": 3, "etag": "fake-etag", "bindings": [ + {"role": "roles/viewer", "members": ["serviceAccount:fake@fake.iam.gserviceaccount.com"]} + ], "auditConfigs": [{"service": "allServices"}]} + self.calls = [] + self.policy_reads = 0 + self.drift = False + self.ambiguous = False + + def __call__(self, method, url, body=None, missing=False): + self.calls.append((method, url, copy.deepcopy(body))) + if url.endswith(":getIamPolicy"): + self.policy_reads += 1 + if self.drift and self.policy_reads > 1: + self.policy["etag"] = "concurrent-writer" + return copy.deepcopy(self.policy) + if url.endswith(":setIamPolicy"): + if self.ambiguous: + raise m.Blocked("remote_result_unknown") + assert body["updateMask"] == "bindings,etag" + assert body["policy"]["etag"] == self.policy["etag"] + self.policy["bindings"] = copy.deepcopy(body["policy"]["bindings"]) + return copy.deepcopy(self.policy) + if method == "POST": + assert body["roleId"] == m.ROLE_ID + self.role = {**body["role"], "name": f"projects/{PROJECT}/roles/{m.ROLE_ID}", "etag": "fake-role-etag"} + return copy.deepcopy(self.role) + return copy.deepcopy(self.role) + + +def exact_role(): + return {"name": f"projects/{PROJECT}/roles/{m.ROLE_ID}", "stage": "GA", "includedPermissions": [m.PERMISSION], "etag": "fake-role-etag"} + + +@pytest.mark.parametrize("key,value", [ + ("GITHUB_REF", "refs/heads/other"), ("GITHUB_RUN_ATTEMPT", "2"), + ("GITHUB_REPOSITORY", "fake/fork"), ("GITHUB_EVENT_NAME", "push"), + ("IAM_ENVIRONMENT", "longbridge-hk"), ("IAM_CONFIRM", "wrong"), + ("IAM_MODE", "unknown"), ("GCP_PROJECT_ID", "fake-live-project"), + ("PAPER_SCHEDULER_GET_PROJECT_ID", ""), ("PAPER_SCHEDULER_GET_READER_EMAIL", "other@fake.iam.gserviceaccount.com"), +]) +def test_invalid_authority_blocks_before_network(key, value): + cloud = Cloud() + with pytest.raises(m.Blocked): + m.execute({**ENV, key: value}, cloud) + assert cloud.calls == [] + + +@pytest.mark.parametrize("permissions", [["cloudscheduler.jobs.run"], [m.PERMISSION, "storage.objects.get"], []]) +def test_role_collision_never_adopted(permissions): + role = {**exact_role(), "includedPermissions": permissions} + cloud = Cloud(role) + with pytest.raises(m.Blocked): + m.execute(ENV, cloud) + assert all(c[0] == "GET" for c in cloud.calls) + + +def test_dry_run_is_read_only_and_default(): + cloud = Cloud() + env = {k: v for k, v in ENV.items() if k != "IAM_MODE"} + m.execute(env, cloud) + assert cloud.role is None + assert not any(c[1].endswith(":setIamPolicy") or (c[0] == "POST" and not c[1].endswith(":getIamPolicy")) for c in cloud.calls) + + +def test_additive_cas_preserves_unrelated_and_is_idempotent(): + cloud = Cloud() + before = copy.deepcopy(cloud.policy) + m.execute(ENV, cloud) + assert cloud.policy["bindings"][0] == before["bindings"][0] + assert cloud.policy["auditConfigs"] == before["auditConfigs"] + assert cloud.policy["bindings"][1] == {"role": exact_role()["name"], "members": [f"serviceAccount:{READER}"]} + count = len([c for c in cloud.calls if c[1].endswith(":setIamPolicy")]) + m.execute(ENV, cloud) + assert len([c for c in cloud.calls if c[1].endswith(":setIamPolicy")]) == count + + +def test_concurrent_policy_change_stops_before_binding(): + cloud = Cloud(exact_role()) + cloud.drift = True + with pytest.raises(m.Blocked): + m.execute(ENV, cloud) + assert not any(c[1].endswith(":setIamPolicy") for c in cloud.calls) + + +def test_ambiguous_write_is_not_retried(): + cloud = Cloud(exact_role()) + cloud.ambiguous = True + with pytest.raises(m.Blocked): + m.execute(ENV, cloud) + assert len([c for c in cloud.calls if c[1].endswith(":setIamPolicy")]) == 1 + + +@pytest.mark.parametrize("mutation", ["no-etag", "conditional", "duplicate", "extra-member"]) +def test_invalid_policy_fails_closed(mutation): + cloud = Cloud(exact_role()) + target = {"role": exact_role()["name"], "members": [f"serviceAccount:{READER}"]} + if mutation == "no-etag": + cloud.policy.pop("etag") + else: + cloud.policy["bindings"].append(target) + if mutation == "conditional": + target["condition"] = {"expression": "true"} + if mutation == "duplicate": + cloud.policy["bindings"].append(copy.deepcopy(target)) + if mutation == "extra-member": + target["members"].append("serviceAccount:fake-other@fake.iam.gserviceaccount.com") + with pytest.raises(m.Blocked): + m.execute(ENV, cloud) + assert not any(c[1].endswith(":setIamPolicy") for c in cloud.calls) + + +def test_template_permission_drift_blocks(tmp_path): + path = tmp_path / "role.json" + path.write_text(json.dumps({"includedPermissions": ["secretmanager.versions.access"]})) + with pytest.raises(m.Blocked): + m.role_definition(path) + + +def test_remote_failure_never_prints_private_data(monkeypatch, capsys): + monkeypatch.setattr(m.os, "environ", ENV) + def fail(): + raise RuntimeError("fake-private-token fake@fake.invalid") + monkeypatch.setattr(m, "Cloud", fail) + assert m.main() == 1 + output = capsys.readouterr().out + assert "fake-private" not in output and "fake@" not in output + assert output.startswith("BLOCKED:") + + +def test_role_drift_after_create_blocks_binding(): + cloud = Cloud() + original = cloud.__call__ + def request(method, url, body=None, missing=False): + value = original(method, url, body, missing) + if url.endswith(":getIamPolicy") and cloud.policy_reads == 3: + cloud.role["includedPermissions"].append("cloudscheduler.jobs.run") + return value + with pytest.raises(m.Blocked): + m.execute(ENV, request) + assert not any(c[1].endswith(":setIamPolicy") for c in cloud.calls) + + +def test_transport_does_not_retry_or_follow_redirects(monkeypatch): + calls = [] + monkeypatch.setattr(m.subprocess, "run", lambda *a, **k: type("Result", (), {"stdout": b"fake-token"})()) + class Opener: + def open(self, req, timeout): + calls.append(req) + raise TimeoutError("fake-private-diagnostic") + monkeypatch.setattr(m.urllib.request, "build_opener", lambda handler: Opener()) + with pytest.raises(m.Blocked, match="remote_result_unknown"): + m.Cloud()("POST", "https://fake.invalid/test", {}) + assert len(calls) == 1 + + +def test_workflow_boundaries(): + text = (Path(__file__).parents[1] / ".github/workflows/apply-paper-scheduler-jobs-get.yml").read_text() + for gate in ["workflow_dispatch:", "github.ref == 'refs/heads/main'", "github.run_attempt == 1", + "github.repository == 'QuantStrategyLab/LongBridgePlatform'", "environment: longbridge-paper", + "cancel-in-progress: false", "default: dry-run", "APPLY_PAPER_SCHEDULER_JOBS_GET"]: + assert gate in text + assert "secrets." not in text and "longbridge-hk" not in text + assert "gcloud " not in text