diff --git a/docs/iam/paper-scheduler-get.md b/docs/iam/paper-scheduler-get.md index bb2ef33..7e748e8 100644 --- a/docs/iam/paper-scheduler-get.md +++ b/docs/iam/paper-scheduler-get.md @@ -128,3 +128,22 @@ 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`. + + +### Executor permission observations + +After authentication, a single read-only `projects.testIamPermissions` request +checks only `iam.roles.get`, `resourcemanager.projects.getIamPolicy`, +`iam.roles.create`, and `resourcemanager.projects.setIamPolicy`. The response is +limited to 4096 bytes. Output contains only `executor_preflight=held_all|missing` +and `executor_missing`, in the fixed allowlist order. An omitted permissions +field means none held; invalid, duplicate or unexpected values stop with a fixed +unknown category. HTTP, transport and JSON failures stop without retry or raw +response output, with phase `test_iam_permissions`. + +These observations do not grant authority. A missing result still reaches the +existing role/policy validation; `held_all` proves neither API availability nor +successful later operations. The original role GET, accepted 404, policy checks, +CAS, dry-run default and apply gates remain required. A later executor grant, +especially project `setIamPolicy`, needs a separate owner decision. Publishing +this diagnostic does not authorize its execution or any IAM change. diff --git a/scripts/paper_scheduler_get_iam.py b/scripts/paper_scheduler_get_iam.py index 75e7793..3672c10 100644 --- a/scripts/paper_scheduler_get_iam.py +++ b/scripts/paper_scheduler_get_iam.py @@ -12,6 +12,12 @@ import urllib.error import urllib.request +EXECUTOR_PERMISSIONS = ( + "iam.roles.get", "resourcemanager.projects.getIamPolicy", + "iam.roles.create", "resourcemanager.projects.setIamPolicy", +) +PREFLIGHT_MAX_BYTES = 4096 + ROLE_ID = "paperSchedulerJobsGet" PERMISSION = "cloudscheduler.jobs.get" ROLE_FILE = Path(__file__).resolve().parents[1] / "docs/iam/paper-scheduler-get-role.json" @@ -89,6 +95,20 @@ def read_policy(): validate_policy(policy, name, member) return policy + # Observations only: even held_all cannot establish API availability or + # authorize writes. Missing permissions never replace the existing reads. + result = request("POST", project_url + ":testIamPermissions", + {"permissions": list(EXECUTOR_PERMISSIONS)}) + held = result.get("permissions", []) if isinstance(result, dict) else None + if (not isinstance(result, dict) or set(result) - {"permissions"} + or not isinstance(held, list) + or any(type(p) is not str or p not in EXECUTOR_PERMISSIONS for p in held) + or len(held) != len(set(held))): + raise Blocked("executor_preflight_unknown_stop_no_retry", phase="test_iam_permissions") + missing_permissions = [p for p in EXECUTOR_PERMISSIONS if p not in held] + status = "missing" if missing_permissions else "held_all" + print("executor_preflight=" + status + " executor_missing=" + json.dumps(missing_permissions)) + before_role = request("GET", role_url, missing=True) if before_role is not None: validate_role(before_role, name) @@ -168,6 +188,8 @@ def __call__(self, method, url, body=None, missing=False): phase = "role_get" elif method == "POST" and url.endswith(":getIamPolicy"): phase = "get_iam_policy" + elif method == "POST" and url.endswith(":testIamPermissions"): + phase = "test_iam_permissions" 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", @@ -179,6 +201,11 @@ 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: + if phase == "test_iam_permissions": + payload = response.read(PREFLIGHT_MAX_BYTES + 1) + if len(payload) > PREFLIGHT_MAX_BYTES: + raise ValueError("oversized_preflight") + return json.loads(payload) return json.load(response) except urllib.error.HTTPError as error: if missing and method == "GET" and error.code == 404: @@ -208,6 +235,7 @@ def main(): "role_collision_or_integrity_failure", "policy_integrity_failure", "binding_collision", "orphan_binding_collision", "concurrent_drift", "after_readback_drift", "authentication_failed", + "executor_preflight_unknown_stop_no_retry", "remote_rejected_or_unknown_stop_no_retry", "remote_result_unknown_stop_no_retry", "remote_http_401_stop_no_retry", "remote_http_403_stop_no_retry", "remote_http_429_stop_no_retry", "remote_http_5xx_stop_no_retry", @@ -218,7 +246,7 @@ def main(): print("BLOCKED: unknown_failure phase=unknown") else: phase = error.phase - if type(phase) is not str or phase not in {"token_acquisition", "role_get", "get_iam_policy"}: + if type(phase) is not str or phase not in {"token_acquisition", "role_get", "get_iam_policy", "test_iam_permissions"}: phase = "unknown" print(f"BLOCKED: {category} phase={phase}") return 1 diff --git a/tests/test_paper_scheduler_get_iam.py b/tests/test_paper_scheduler_get_iam.py index 39067b0..466d33c 100644 --- a/tests/test_paper_scheduler_get_iam.py +++ b/tests/test_paper_scheduler_get_iam.py @@ -35,6 +35,8 @@ def __init__(self, role=None): def __call__(self, method, url, body=None, missing=False): self.calls.append((method, url, copy.deepcopy(body))) + if url.endswith(":testIamPermissions"): + return {"permissions": list(m.EXECUTOR_PERMISSIONS)} if url.endswith(":getIamPolicy"): self.policy_reads += 1 if self.drift and self.policy_reads > 1: @@ -78,7 +80,7 @@ def test_role_collision_never_adopted(permissions): cloud = Cloud(role) with pytest.raises(m.Blocked): m.execute(ENV, cloud) - assert all(c[0] == "GET" for c in cloud.calls) + assert all(c[0] == "GET" or c[1].endswith(":testIamPermissions") for c in cloud.calls) def test_dry_run_is_read_only_and_default(): @@ -86,7 +88,7 @@ def test_dry_run_is_read_only_and_default(): 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) + assert not any(c[1].endswith(":setIamPolicy") or (c[0] == "POST" and not c[1].endswith((":getIamPolicy", ":testIamPermissions"))) for c in cloud.calls) def test_additive_cas_preserves_unrelated_and_is_idempotent(): @@ -254,6 +256,9 @@ def test_remote_diagnostic_phase_and_no_retry(phase, failure, monkeypatch, capsy calls = [] class Opener: def open(self, req, timeout): + if req.full_url.endswith(":testIamPermissions"): + from io import BytesIO + return BytesIO(json.dumps({"permissions": list(m.EXECUTOR_PERMISSIONS)}).encode()) calls.append(req) if phase == "get_iam_policy" and len(calls) == 1: raise m.urllib.error.HTTPError(req.full_url, 404, HOSTILE, {}, None) @@ -266,7 +271,7 @@ def open(self, req, timeout): monkeypatch.setattr(m.urllib.request, "build_opener", lambda handler: Opener()) assert m.main() == 1 category = "remote_http_403_stop_no_retry" if failure == "http" else "remote_result_unknown_stop_no_retry" - assert capsys.readouterr() == (f"BLOCKED: {category} phase={phase}\n", "") + assert capsys.readouterr() == ('executor_preflight=held_all executor_missing=[]\n' + f"BLOCKED: {category} phase={phase}\n", "") assert len(calls) == (1 if phase == "role_get" else 2) assert all(req.get_method() == "GET" or req.full_url.endswith(":getIamPolicy") for req in calls) @@ -329,3 +334,84 @@ def open(self, req, timeout): assert m.Cloud()("GET", "https://fake.invalid/role", missing=True) is None assert len(calls) == 1 assert capsys.readouterr() == ("", "") + + +EXECUTOR_PERMISSIONS = ( + "iam.roles.get", "resourcemanager.projects.getIamPolicy", + "iam.roles.create", "resourcemanager.projects.setIamPolicy", +) + + +@pytest.mark.parametrize("held", [[], list(EXECUTOR_PERMISSIONS), [EXECUTOR_PERMISSIONS[2]]]) +def test_executor_preflight_fixed_allowlist_and_diagnostic_only(held, capsys): + cloud = Cloud() + original = cloud.__call__ + def request(method, url, body=None, missing=False): + if url.endswith(":testIamPermissions"): + cloud.calls.append((method, url, body)) + assert method == "POST" and body == {"permissions": list(EXECUTOR_PERMISSIONS)} + return {"permissions": held} + return original(method, url, body, missing) + m.execute({**ENV, "IAM_MODE": "dry-run"}, request) + missing = [p for p in EXECUTOR_PERMISSIONS if p not in held] + category = "missing" if missing else "held_all" + output = capsys.readouterr().out + assert output.startswith(f"executor_preflight={category} executor_missing=" + json.dumps(missing) + "\n") + assert cloud.calls[0][1].endswith(":testIamPermissions") + assert cloud.calls[1][0] == "GET" # diagnostic never replaces role validation + assert cloud.role is None + + +@pytest.mark.parametrize("result", [None, [], {"permissions": None}, + {"permissions": [HOSTILE]}, {"permissions": [1]}, + {"permissions": [EXECUTOR_PERMISSIONS[0]] * 2}, {"extra": HOSTILE}, + {"permissions": [], "extra": HOSTILE}]) +def test_executor_preflight_malformed_stops_without_leak(result, monkeypatch, capsys): + monkeypatch.setattr(m.os, "environ", ENV) + calls = [] + def request(*args, **kwargs): + calls.append(args) + return result + monkeypatch.setattr(m, "Cloud", lambda: request) + assert m.main() == 1 + assert capsys.readouterr() == ("BLOCKED: executor_preflight_unknown_stop_no_retry phase=test_iam_permissions\n", "") + assert len(calls) == 1 + + +@pytest.mark.parametrize("failure", ["http", "transport", "json", "oversized"]) +def test_executor_preflight_transport_bounded_and_single_attempt(failure, monkeypatch, capsys): + from io import BytesIO + monkeypatch.setattr(m.os, "environ", ENV) + monkeypatch.setattr(m.subprocess, "run", lambda *a, **k: type("Result", (), {"stdout": b"fake-token"})()) + calls = [] + class Response(BytesIO): + def read(self, size=-1): + assert size == 4097 + return super().read(size) + class Opener: + def open(self, req, timeout): + calls.append(req) + assert req.full_url.endswith(":testIamPermissions") + if failure == "http": + raise m.urllib.error.HTTPError(req.full_url, 403, HOSTILE, {}, BytesIO(HOSTILE.encode())) + if failure == "transport": + raise TimeoutError(HOSTILE) + return Response((HOSTILE if failure == "json" else " " * 4097).encode()) + monkeypatch.setattr(m.urllib.request, "build_opener", lambda handler: Opener()) + assert m.main() == 1 + category = "remote_http_403_stop_no_retry" if failure == "http" else "remote_result_unknown_stop_no_retry" + assert capsys.readouterr() == (f"BLOCKED: {category} phase=test_iam_permissions\n", "") + assert len(calls) == 1 + + +def test_executor_preflight_empty_object_means_no_permissions(capsys): + calls = [] + def request(method, url, body=None, missing=False): + calls.append(url) + if url.endswith(":testIamPermissions"): + return {} # API omits an empty repeated field + raise m.Blocked("remote_http_403_stop_no_retry", phase="role_get") + with pytest.raises(m.Blocked, match="remote_http_403"): + m.execute(ENV, request) + assert len(calls) == 2 + assert capsys.readouterr().out == 'executor_preflight=missing executor_missing=' + json.dumps(list(EXECUTOR_PERMISSIONS)) + '\n'