|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""404-vs-unreachable gate — exception-type-blind classification enforcement. |
| 3 | +
|
| 4 | +RomM answers a request for an entity it no longer has with HTTP 404. That is |
| 5 | +the server *answering*, not the server being unreachable, so it must reach the |
| 6 | +frontend as ``not_found`` — never as ``server_unreachable``. The plugin already |
| 7 | +owns the funnel that decides this (:func:`lib.errors.classify_error`, which maps |
| 8 | +``RommNotFoundError`` to :data:`ErrorCode.NOT_FOUND`); the recurring defect is a |
| 9 | +catch-all ``except Exception`` that ignores the funnel and hardcodes the |
| 10 | +unreachable verdict for every exception it sees. The whole UI then claims "RomM |
| 11 | +offline" while the server is plainly answering. |
| 12 | +
|
| 13 | +This class of bug was swept once before (#971) and came back, because |
| 14 | +``check_failure_shape.py`` inspects key *presence* only — a hardcoded |
| 15 | +``server_unreachable`` is a perfectly canonical failure shape, so every one of |
| 16 | +these sites is green there. This check inspects the slug *choice* instead. |
| 17 | +
|
| 18 | +The rule |
| 19 | +======== |
| 20 | +
|
| 21 | +Inside ``py_modules/services/``, a **catch-all** exception handler (``except |
| 22 | +Exception``, ``except BaseException``, or a bare ``except:``) may not *return* a |
| 23 | +hardcoded ``server_unreachable`` verdict. A handler returns one when a dict |
| 24 | +literal in its body binds a verdict key — ``reason``, ``status``, or |
| 25 | +``recommended_action`` — to ``ErrorCode.SERVER_UNREACHABLE`` or to the bare |
| 26 | +``"server_unreachable"`` string. Both spellings count: the canonical failure |
| 27 | +shape uses the enum, while the discriminated-status and ``recommended_action`` |
| 28 | +carve-outs spell the slug out as a literal. |
| 29 | +
|
| 30 | +Binding the key to a *name* (``{"reason": reason}`` after ``reason, message = |
| 31 | +classify_error(e)``) is the correct shape and is what this check is steering |
| 32 | +towards, so it is never flagged. Keying on the verdict's position — rather than |
| 33 | +on the mere mention of the slug anywhere in the handler — is deliberate: it |
| 34 | +catches the sharpest form of this bug, a handler that calls ``classify_error`` |
| 35 | +and then **discards** its verdict in favour of the hardcoded one, which a |
| 36 | +"does the handler call the funnel?" test would wave through. |
| 37 | +
|
| 38 | +One escape clears an otherwise-flagged handler: |
| 39 | +
|
| 40 | +* **A sibling handler peels the 404 off first** — the same ``try`` carries an |
| 41 | + ``except RommNotFoundError`` clause, so the definitive-404 case never reaches |
| 42 | + the catch-all. This is the shape the partial-success carve-outs use, where the |
| 43 | + verdict is a flag rather than a ``reason`` slug and there is no funnel to call. |
| 44 | +
|
| 45 | +``EXEMPT`` holds service modules deliberately kept out of the scan — a module |
| 46 | +that does not talk to RomM at all, so ``classify_error`` (a RomM-shaped funnel) |
| 47 | +has nothing to say about its failures. Adding to it is a deliberate act that |
| 48 | +shows up in review. |
| 49 | +
|
| 50 | +Two modes: |
| 51 | +
|
| 52 | + * report (default) — print every catch-all handler carrying an unreachable |
| 53 | + verdict, grouped by verdict, then exit 0. Report mode never fails; it is the |
| 54 | + inventory. |
| 55 | + * ``--check`` — enforce mode. Exit 1 on any violation. |
| 56 | +
|
| 57 | +A **typed** handler is never flagged. ``except RommConnectionError`` or |
| 58 | +``except SaveSyncTimeoutError`` returning ``server_unreachable`` is a deliberate |
| 59 | +statement about a known exception type, which is exactly what this check wants |
| 60 | +more of. |
| 61 | +
|
| 62 | +The AST heuristic is intentionally conservative, and a guardrail rather than a |
| 63 | +prover: it reads one ``try`` statement at a time and does not follow a verdict |
| 64 | +returned by a helper the handler calls, nor one assembled across statements |
| 65 | +(``resp = {...}; resp["reason"] = ...``). A ``try`` nested *inside* a catch-all |
| 66 | +handler contributes its own dict literals to that handler's scan, so a nested |
| 67 | +``except RommNotFoundError`` peel can clear the outer handler too. |
| 68 | +""" |
| 69 | + |
| 70 | +from __future__ import annotations |
| 71 | + |
| 72 | +import ast |
| 73 | +import sys |
| 74 | +from dataclasses import dataclass |
| 75 | +from pathlib import Path |
| 76 | + |
| 77 | +REPO_ROOT = Path(__file__).resolve().parent.parent |
| 78 | +SERVICES_DIR = REPO_ROOT / "py_modules" / "services" |
| 79 | + |
| 80 | +# The canonical slug, in both spellings a service can write it. |
| 81 | +UNREACHABLE_ENUM_MEMBER = "SERVER_UNREACHABLE" |
| 82 | +UNREACHABLE_SLUG = "server_unreachable" |
| 83 | + |
| 84 | +# Dict keys whose value is the verdict a consumer routes on: the canonical |
| 85 | +# failure shape's ``reason``, plus the two documented carve-out discriminants. |
| 86 | +VERDICT_KEYS = frozenset({"reason", "status", "recommended_action"}) |
| 87 | + |
| 88 | +# The exception whose presence as a sibling clause proves the 404 was peeled off. |
| 89 | +NOT_FOUND_EXCEPTION = "RommNotFoundError" |
| 90 | + |
| 91 | +# Catch-all handler types (``except:`` with no type is handled separately). |
| 92 | +CATCH_ALL_NAMES = frozenset({"Exception", "BaseException"}) |
| 93 | + |
| 94 | +# Service modules deliberately out of scope, as ``<path>: <why>``. Only for |
| 95 | +# modules that never talk to RomM — classify_error cannot classify their errors. |
| 96 | +EXEMPT: dict[str, str] = { |
| 97 | + "steamgrid.py": ( |
| 98 | + "talks to SteamGridDB, not RomM — SGDB failures arrive as SgdbApiError " |
| 99 | + "(carrying its own status_code) and classify_error has no SGDB branch" |
| 100 | + ), |
| 101 | +} |
| 102 | + |
| 103 | + |
| 104 | +@dataclass(frozen=True) |
| 105 | +class Finding: |
| 106 | + """One catch-all handler that hardcodes the unreachable verdict.""" |
| 107 | + |
| 108 | + path: Path |
| 109 | + lineno: int |
| 110 | + spelling: str |
| 111 | + function: str |
| 112 | + |
| 113 | + @property |
| 114 | + def rel(self) -> str: |
| 115 | + return str(self.path.relative_to(REPO_ROOT)) |
| 116 | + |
| 117 | + def render(self) -> str: |
| 118 | + return f"{self.rel}:{self.lineno} in {self.function}() [{self.spelling}]" |
| 119 | + |
| 120 | + |
| 121 | +def _handler_is_catch_all(handler: ast.ExceptHandler) -> bool: |
| 122 | + """True when *handler* catches everything (bare, Exception, BaseException). |
| 123 | +
|
| 124 | + A tuple clause counts when any member is a catch-all name: ``except |
| 125 | + (Exception, X)`` still swallows every exception. |
| 126 | + """ |
| 127 | + if handler.type is None: # bare ``except:`` |
| 128 | + return True |
| 129 | + candidates = handler.type.elts if isinstance(handler.type, ast.Tuple) else [handler.type] |
| 130 | + return any(isinstance(node, ast.Name) and node.id in CATCH_ALL_NAMES for node in candidates) |
| 131 | + |
| 132 | + |
| 133 | +def _handler_catches_not_found(handler: ast.ExceptHandler) -> bool: |
| 134 | + """True when *handler* explicitly catches ``RommNotFoundError``.""" |
| 135 | + if handler.type is None: |
| 136 | + return False |
| 137 | + candidates = handler.type.elts if isinstance(handler.type, ast.Tuple) else [handler.type] |
| 138 | + return any(isinstance(node, ast.Name) and node.id == NOT_FOUND_EXCEPTION for node in candidates) |
| 139 | + |
| 140 | + |
| 141 | +def _hardcoded_verdict_spelling(value: ast.expr) -> str | None: |
| 142 | + """Return how *value* spells a hardcoded unreachable verdict, else None. |
| 143 | +
|
| 144 | + ``ErrorCode.SERVER_UNREACHABLE`` and ``ErrorCode.SERVER_UNREACHABLE.value`` |
| 145 | + both surface as an ``Attribute`` named ``SERVER_UNREACHABLE``; the carve-out |
| 146 | + shapes write the bare ``"server_unreachable"`` string instead. A ``Name`` |
| 147 | + (the classified slug held in a variable) is neither, and returns None. |
| 148 | + """ |
| 149 | + for node in ast.walk(value): |
| 150 | + if isinstance(node, ast.Attribute) and node.attr == UNREACHABLE_ENUM_MEMBER: |
| 151 | + return f"ErrorCode.{UNREACHABLE_ENUM_MEMBER}" |
| 152 | + if isinstance(node, ast.Constant) and node.value == UNREACHABLE_SLUG: |
| 153 | + return f'"{UNREACHABLE_SLUG}"' |
| 154 | + return None |
| 155 | + |
| 156 | + |
| 157 | +def _unreachable_spelling(handler: ast.ExceptHandler) -> str | None: |
| 158 | + """Return how *handler* hardcodes the unreachable verdict, or None. |
| 159 | +
|
| 160 | + Only a verdict *key* counts (see :data:`VERDICT_KEYS`) — mentioning the slug |
| 161 | + anywhere else in the handler (a log line, a comparison against a classified |
| 162 | + reason) is not a hardcoded verdict. |
| 163 | + """ |
| 164 | + for node in ast.walk(handler): |
| 165 | + if not isinstance(node, ast.Dict): |
| 166 | + continue |
| 167 | + for key, value in zip(node.keys, node.values, strict=True): |
| 168 | + if not (isinstance(key, ast.Constant) and key.value in VERDICT_KEYS): |
| 169 | + continue |
| 170 | + spelling = _hardcoded_verdict_spelling(value) |
| 171 | + if spelling is not None: |
| 172 | + return spelling |
| 173 | + return None |
| 174 | + |
| 175 | + |
| 176 | +def _enclosing_functions(tree: ast.AST) -> dict[int, str]: |
| 177 | + """Map every node's line to the name of the function that contains it.""" |
| 178 | + owner: dict[int, str] = {} |
| 179 | + for node in ast.walk(tree): |
| 180 | + if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): |
| 181 | + continue |
| 182 | + for child in ast.walk(node): |
| 183 | + lineno = getattr(child, "lineno", None) |
| 184 | + if lineno is not None: |
| 185 | + # Inner functions win: they are visited after the outer one only |
| 186 | + # when nested, so record the most specific (shortest) span. |
| 187 | + owner[lineno] = node.name |
| 188 | + return owner |
| 189 | + |
| 190 | + |
| 191 | +def scan_source(path: Path, source: str) -> list[Finding]: |
| 192 | + """Return every catch-all handler in *source* that hardcodes the verdict.""" |
| 193 | + try: |
| 194 | + tree = ast.parse(source, filename=str(path)) |
| 195 | + except SyntaxError: |
| 196 | + return [] |
| 197 | + |
| 198 | + function_of = _enclosing_functions(tree) |
| 199 | + findings: list[Finding] = [] |
| 200 | + |
| 201 | + for node in ast.walk(tree): |
| 202 | + if not isinstance(node, ast.Try): |
| 203 | + continue |
| 204 | + peels_not_found = any(_handler_catches_not_found(h) for h in node.handlers) |
| 205 | + if peels_not_found: |
| 206 | + continue |
| 207 | + for handler in node.handlers: |
| 208 | + if not _handler_is_catch_all(handler): |
| 209 | + continue |
| 210 | + spelling = _unreachable_spelling(handler) |
| 211 | + if spelling is None: |
| 212 | + continue |
| 213 | + findings.append( |
| 214 | + Finding( |
| 215 | + path=path, |
| 216 | + lineno=handler.lineno, |
| 217 | + spelling=spelling, |
| 218 | + function=function_of.get(handler.lineno, "<module>"), |
| 219 | + ) |
| 220 | + ) |
| 221 | + return findings |
| 222 | + |
| 223 | + |
| 224 | +def scan_file(path: Path) -> list[Finding]: |
| 225 | + try: |
| 226 | + source = path.read_text(encoding="utf-8") |
| 227 | + except OSError: |
| 228 | + return [] |
| 229 | + return scan_source(path, source) |
| 230 | + |
| 231 | + |
| 232 | +def collect_findings(services_dir: Path = SERVICES_DIR) -> list[Finding]: |
| 233 | + """Walk *services_dir* and return every hardcoded-verdict finding.""" |
| 234 | + findings: list[Finding] = [] |
| 235 | + if not services_dir.is_dir(): |
| 236 | + return findings |
| 237 | + for path in sorted(services_dir.rglob("*.py")): |
| 238 | + if str(path.relative_to(services_dir)) in EXEMPT: |
| 239 | + continue |
| 240 | + findings.extend(scan_file(path)) |
| 241 | + return findings |
| 242 | + |
| 243 | + |
| 244 | +def _print_report(findings: list[Finding]) -> None: |
| 245 | + print(f"=== CATCH-ALL HANDLERS HARDCODING server_unreachable ({len(findings)}) ===") |
| 246 | + for finding in findings: |
| 247 | + print(f" {finding.render()}") |
| 248 | + print() |
| 249 | + if EXEMPT: |
| 250 | + print("=== EXEMPT MODULES ===") |
| 251 | + for name, why in sorted(EXEMPT.items()): |
| 252 | + print(f" {name} — {why}") |
| 253 | + print() |
| 254 | + print("=== SUMMARY ===") |
| 255 | + print(f" TOTAL: {len(findings)}") |
| 256 | + |
| 257 | + |
| 258 | +def _print_violations(findings: list[Finding]) -> None: |
| 259 | + print(f"=== CATCH-ALL HANDLERS HARDCODING server_unreachable ({len(findings)}) ===") |
| 260 | + for finding in findings: |
| 261 | + print(f" {finding.render()}") |
| 262 | + print() |
| 263 | + print( |
| 264 | + "ERROR: a catch-all 'except Exception' in py_modules/services/ must not hardcode " |
| 265 | + "ErrorCode.SERVER_UNREACHABLE — a definitive 404 is the server ANSWERING, and must " |
| 266 | + "reach the frontend as 'not_found'. Route the exception through classify_error() " |
| 267 | + "(lib/errors.py), or peel the 404 off with a sibling 'except RommNotFoundError' " |
| 268 | + "clause when the verdict is a partial-success flag rather than a reason slug " |
| 269 | + "(CLAUDE.md → invariant register)." |
| 270 | + ) |
| 271 | + |
| 272 | + |
| 273 | +def main(argv: list[str]) -> int: |
| 274 | + if any(a in {"-h", "--help"} for a in argv): |
| 275 | + print(__doc__) |
| 276 | + return 0 |
| 277 | + |
| 278 | + enforce = "--check" in argv |
| 279 | + findings = collect_findings(SERVICES_DIR) |
| 280 | + |
| 281 | + if enforce: |
| 282 | + if findings: |
| 283 | + _print_violations(findings) |
| 284 | + return 1 |
| 285 | + print(f"OK: no catch-all handler hardcodes server_unreachable in {SERVICES_DIR.relative_to(REPO_ROOT)}.") |
| 286 | + return 0 |
| 287 | + |
| 288 | + _print_report(findings) |
| 289 | + return 0 |
| 290 | + |
| 291 | + |
| 292 | +if __name__ == "__main__": |
| 293 | + sys.exit(main(sys.argv[1:])) |
0 commit comments