fix: Expose safe IAM read failure diagnostics

scripts/paper_scheduler_get_iam.py

@@ -20,6 +20,10 @@
 class Blocked(ValueError):
     """Fixed failure category; never attach remote payloads or identifiers."""
 
+    def __init__(self, *args, phase="unknown"):
+        super().__init__(*args)
+        self.phase = phase
+
 
 def role_definition(path=ROLE_FILE):
     value = json.loads(path.read_text())
@@ -149,12 +153,21 @@ 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()
+        try:
+            result = subprocess.run(["gcloud", "auth", "print-access-token"], capture_output=True, timeout=60, check=True)
+            self.token = result.stdout.decode().strip()
+        except (subprocess.SubprocessError, OSError, UnicodeError):
+            raise Blocked("authentication_failed", phase="token_acquisition") from None
         if not self.token or "\n" in self.token or "\r" in self.token:
-            raise Blocked("authentication_failed")
+            raise Blocked("authentication_failed", phase="token_acquisition")
 
     def __call__(self, method, url, body=None, missing=False):
+        # Only read operations receive a diagnostic label; never echo a URL.
+        phase = "unknown"
+        if method == "GET":
+            phase = "role_get"
+        elif method == "POST" and url.endswith(":getIamPolicy"):
+            phase = "get_iam_policy"
         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",
@@ -170,19 +183,36 @@ def redirect_request(self, req, fp, code, msg, headers, newurl):
         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
+            raise Blocked("remote_rejected_or_unknown_stop_no_retry", phase=phase) from None
         except Exception:
-            raise Blocked("remote_result_unknown_stop_no_retry") from None
+            raise Blocked("remote_result_unknown_stop_no_retry", phase=phase) from None
 
 
 def main():
     try:
         identity(os.environ)  # Reject wrong scope before token access.
         execute(os.environ, Cloud())
         return 0
+    except Blocked as error:
+        # Never format exception text, remote data, or arbitrary phase labels.
+        categories = {
+            "authority_mapping_rejected", "role_definition_rejected",
+            "role_collision_or_integrity_failure", "policy_integrity_failure",
+            "binding_collision", "orphan_binding_collision", "concurrent_drift",
+            "after_readback_drift", "authentication_failed",
+            "remote_rejected_or_unknown_stop_no_retry", "remote_result_unknown_stop_no_retry",
+        }
+        category = error.args[0] if len(error.args) == 1 else None
+        if type(category) is not str or category not in categories:
+            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"}:
+                phase = "unknown"
+            print(f"BLOCKED: {category} phase={phase}")
+        return 1
     except Exception:
-        # Fixed allowlisted text; suppress even unexpected exception messages.
-        print("BLOCKED: verify mappings, permissions and remote state; do not retry an uncertain write")
+        print("BLOCKED: unknown_failure phase=unknown")
         return 1
 
 

tests/test_paper_scheduler_get_iam.py

@@ -188,3 +188,95 @@ def test_workflow_boundaries():
         assert gate in text
     assert "secrets." not in text and "longbridge-hk" not in text
     assert "gcloud " not in text
+
+
+SAFE_CATEGORIES = [
+    "authority_mapping_rejected", "role_definition_rejected",
+    "role_collision_or_integrity_failure", "policy_integrity_failure",
+    "binding_collision", "orphan_binding_collision", "concurrent_drift",
+    "after_readback_drift", "authentication_failed",
+    "remote_rejected_or_unknown_stop_no_retry", "remote_result_unknown_stop_no_retry",
+]
+HOSTILE = "fake-secret https://fake.invalid/payload fake@example.invalid\nPASS"
+
+
+@pytest.mark.parametrize("category", SAFE_CATEGORIES)
+def test_main_emits_only_allowlisted_category(category, monkeypatch, capsys):
+    monkeypatch.setattr(m.os, "environ", ENV)
+    def fail():
+        raise m.Blocked(category)
+    monkeypatch.setattr(m, "Cloud", fail)
+    assert m.main() == 1
+    assert capsys.readouterr() == (f"BLOCKED: {category} phase=unknown\n", "")
+
+
+@pytest.mark.parametrize("error", [m.Blocked(HOSTILE), RuntimeError(HOSTILE), m.Blocked(), m.Blocked([HOSTILE])])
+def test_main_unknown_failure_cannot_leak(error, monkeypatch, capsys):
+    monkeypatch.setattr(m.os, "environ", ENV)
+    def fail():
+        raise error
+    monkeypatch.setattr(m, "Cloud", fail)
+    assert m.main() == 1
+    assert capsys.readouterr() == ("BLOCKED: unknown_failure phase=unknown\n", "")
+
+
+@pytest.mark.parametrize("error", [
+    m.subprocess.CalledProcessError(1, [HOSTILE], stderr=HOSTILE),
+    m.subprocess.TimeoutExpired([HOSTILE], 60, stderr=HOSTILE),
+    FileNotFoundError(HOSTILE), UnicodeError(HOSTILE),
+])
+def test_token_subprocess_failure_is_fixed(error, monkeypatch, capsys):
+    monkeypatch.setattr(m.os, "environ", ENV)
+    calls = []
+    def fail(*args, **kwargs):
+        calls.append((args, kwargs))
+        raise error
+    monkeypatch.setattr(m.subprocess, "run", fail)
+    assert m.main() == 1
+    assert capsys.readouterr() == ("BLOCKED: authentication_failed phase=token_acquisition\n", "")
+    assert len(calls) == 1
+    assert calls[0][1] == {"capture_output": True, "timeout": 60, "check": True}
+
+
+@pytest.mark.parametrize("token", [b"", b"fake\nsecret", b"fake\rsecret", b"\xff"])
+def test_invalid_token_has_safe_phase(token, monkeypatch, capsys):
+    monkeypatch.setattr(m.os, "environ", ENV)
+    monkeypatch.setattr(m.subprocess, "run", lambda *a, **k: type("Result", (), {"stdout": token})())
+    assert m.main() == 1
+    assert capsys.readouterr() == ("BLOCKED: authentication_failed phase=token_acquisition\n", "")
+
+
+@pytest.mark.parametrize("phase", ["role_get", "get_iam_policy"])
+@pytest.mark.parametrize("failure", ["http", "transport", "json"])
+def test_remote_diagnostic_phase_and_no_retry(phase, failure, monkeypatch, capsys):
+    monkeypatch.setattr(m.os, "environ", {**ENV, "IAM_MODE": "dry-run"})
+    monkeypatch.setattr(m.subprocess, "run", lambda *a, **k: type("Result", (), {"stdout": b"fake-token"})())
+    calls = []
+    class Opener:
+        def open(self, req, timeout):
+            calls.append(req)
+            if phase == "get_iam_policy" and len(calls) == 1:
+                raise m.urllib.error.HTTPError(req.full_url, 404, HOSTILE, {}, None)
+            if failure == "http":
+                raise m.urllib.error.HTTPError(req.full_url, 403, HOSTILE, {}, None)
+            if failure == "json":
+                from io import BytesIO
+                return BytesIO(HOSTILE.encode())
+            raise TimeoutError(HOSTILE)
+    monkeypatch.setattr(m.urllib.request, "build_opener", lambda handler: Opener())
+    assert m.main() == 1
+    category = "remote_rejected_or_unknown_stop_no_retry" if failure == "http" else "remote_result_unknown_stop_no_retry"
+    assert capsys.readouterr() == (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)
+
+
+def test_unknown_phase_is_not_printed(monkeypatch, capsys):
+    monkeypatch.setattr(m.os, "environ", ENV)
+    error = m.Blocked("authentication_failed")
+    error.phase = HOSTILE
+    def fail():
+        raise error
+    monkeypatch.setattr(m, "Cloud", fail)
+    assert m.main() == 1
+    assert capsys.readouterr() == ("BLOCKED: authentication_failed phase=unknown\n", "")