ci: Add restricted PAPER Scheduler jobs.get IAM workflow

.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

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`.

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())