From c481c3140c446c15ecf653c8e0556de84d6ca493 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 6 Nov 2026 18:13:12 +0800 Subject: [PATCH] fix: Expose safe IAM read failure diagnostics --- scripts/paper_scheduler_get_iam.py | 44 +++++++++++-- tests/test_paper_scheduler_get_iam.py | 92 +++++++++++++++++++++++++++ 2 files changed, 129 insertions(+), 7 deletions(-) diff --git a/scripts/paper_scheduler_get_iam.py b/scripts/paper_scheduler_get_iam.py index f0fad04..87c116b 100644 --- a/scripts/paper_scheduler_get_iam.py +++ b/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,9 +183,9 @@ 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(): @@ -180,9 +193,26 @@ def main(): 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 diff --git a/tests/test_paper_scheduler_get_iam.py b/tests/test_paper_scheduler_get_iam.py index 399cdee..087aa55 100644 --- a/tests/test_paper_scheduler_get_iam.py +++ b/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", "")