From 7ee5300d43b88cc9e2d12e9086c14f0b2442169e Mon Sep 17 00:00:00 2001 From: Elite Agentic Solutions Date: Fri, 12 Jun 2026 01:29:50 -0700 Subject: [PATCH] =?UTF-8?q?bugclass-detectors=20v0.1.0=20=E2=80=94=20five?= =?UTF-8?q?=20silent-failure=20bug-class=20checks=20(MIT)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AST-aware, stdlib-only static checks: swallowed-exception-around-write (C1), undefined-name (C2), hardcoded-secret-literal (C3), fail-open-flag (C4), rotated-secret-captured-once (C5). Tests + CI + examples included. Co-Authored-By: Claude Fable 5 --- .github/workflows/ci.yml | 20 + .gitignore | 7 + LICENSE | 21 ++ README.md | 102 +++++ bugclass_detectors/__init__.py | 16 + bugclass_detectors/__main__.py | 5 + bugclass_detectors/core.py | 665 +++++++++++++++++++++++++++++++++ examples/buggy_sample.py | 42 +++ examples/clean_sample.py | 42 +++ pyproject.toml | 31 ++ tests/test_detectors.py | 49 +++ 11 files changed, 1000 insertions(+) create mode 100644 .github/workflows/ci.yml create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 README.md create mode 100644 bugclass_detectors/__init__.py create mode 100644 bugclass_detectors/__main__.py create mode 100644 bugclass_detectors/core.py create mode 100644 examples/buggy_sample.py create mode 100644 examples/clean_sample.py create mode 100644 pyproject.toml create mode 100644 tests/test_detectors.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..71e329f --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,20 @@ +name: ci +on: + push: + pull_request: + +jobs: + test: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.8", "3.11", "3.13"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: smoke tests + run: python tests/test_detectors.py + - name: dogfood (self-scan must be clean of HIGH) + run: python -m bugclass_detectors bugclass_detectors/ --lint --min-sev HIGH diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..54940d5 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +__pycache__/ +*.pyc +*.egg-info/ +build/ +dist/ +.venv/ +.pytest_cache/ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..55b819c --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Elite Agentic Solutions + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..01e746e --- /dev/null +++ b/README.md @@ -0,0 +1,102 @@ +# bugclass-detectors + +**AST-aware static checks for the silent-failure bug classes that AI-generated and fast-moving code keeps reintroducing.** + +These are not style nits. They are the bugs that pass code review, pass your tests, keep the process *green* — and silently lose data, fail open, or crash on the one unlucky path. They are exactly the classes an AI coding assistant reintroduces because each instance looks locally reasonable. + +`bugclass-detectors` finds them with a real parser (not grep), in pure-stdlib Python, with zero config. + +```bash +pip install bugclass-detectors # or: pipx run bugclass-detectors . +bugclass-detectors src/ # human report +bugclass-detectors src/ --lint # exit 1 on any HIGH finding (CI gate) +bugclass-detectors src/ --json # machine-readable +``` + +--- + +## The five bug classes + +| ID | Class | What it catches | Why it bites | +|----|-------|-----------------|--------------| +| **C1** | `SWALLOWED_EXCEPTION_AROUND_WRITE` | A bare/catch-all `except` that swallows an exception wrapping a **write** (DB / cache / file) with no log and no re-raise. | The producer stays alive, the heartbeat is green, the data is *gone*, and nothing logs. The single most dangerous silent-failure shape there is. | +| **C2** | `UNDEFINED_NAME` | Use of a name never bound in its scope and not a builtin/import/param/closure var. | A `NameError` that only fires on the code path you didn't test. (pyflakes F821 class, no pyflakes dependency.) | +| **C3** | `HARDCODED_SECRET_LITERAL` | A password/token/key assigned to an inline string literal (in code or a systemd `Environment=`). | Diverges silently the moment the real secret rotates → an outage with no obvious cause. Secrets are masked in output. | +| **C4** | `FAIL_OPEN_FLAG` | A safety / kill / arming / feature gate read from a store repointable by one env var, with no presence assertion. | If the flag is **absent** on the repointed store it reads falsy → the gate fails **open** instead of closed. | +| **C5** | `ROTATED_SECRET_CAPTURED_ONCE` | A long-lived consumer (`while True`, a daemon) that captures a secret **once** at module scope. | When the secret rotates the process keeps the stale value and silently 401/403s forever. | + +Every check is **AST/tokenizer-aware** (so it doesn't fire on the word in a comment or a string), **read-only**, and emits a uniform finding: + +```json +{ "check": "C1_SWALLOWED_WRITE", "severity": "HIGH", + "file": "src/recorder.py", "lineno": 88, + "snippet": "except Exception:", + "why": "catch-all/bare except SILENTLY SWALLOWS a db/cache write `hset` — ..." } +``` + +Disable any check with `BUGCLASS_DISABLE=C1,C4` or `--checks C2,C3`. + +--- + +## Quick start + +```bash +git clone https://github.com//bugclass-detectors +cd bugclass-detectors +python -m bugclass_detectors examples/buggy_sample.py # see all five fire +python tests/test_detectors.py # smoke tests +``` + +### As a CI gate (GitHub Actions) + +```yaml +- name: bug-class lint + run: pipx run bugclass-detectors . --lint --min-sev HIGH +``` + +`--lint` returns a non-zero exit on any HIGH finding, so a swallowed write or an undefined name fails the build. + +### As a library + +```python +from bugclass_detectors import run +for f in run(["./src"], checks=["C1", "C4"], min_sev="HIGH"): + print(f["file"], f["lineno"], f["why"]) +``` + +--- + +## Why these five (and why a parser, not grep) + +This pack started as the in-house lint that runs against a continuously-operating multi-service production estate. Each class on this list earned its place by causing a **real** silent outage in production — a recorder whose only output was swallowed for weeks behind a bare `except`, a feature gate that failed open when its flag-store was repointed, a credential captured once that 403'd silently after a rotation. The fix in every case was one line; the cost of *not* catching it was days. + +Grep can't catch these without drowning you in false positives — `except` and `password` and `kill` appear constantly in comments, strings, and legitimate code. These checks parse the AST and tokenize the source, so they fire on the *shape* of the bug, not the word. + +--- + +## Limitations (honest) + +- Python only (today). The classes are language-agnostic; ports welcome. +- C1/C4/C5 are heuristic — tuned to favor precision over recall (few false positives, will miss exotic phrasings). C2 is conservative (only flags provably-unbound names). +- This is a *focused* tool, not a replacement for a full linter, type-checker, or SAST suite. Run it alongside them. + +--- + +## Want this run continuously, across every repo, with triage and auto-fix? + +This open-source pack is the **engine**. Running it well at scale — across many repos and languages, with severity triage, suppression management, auto-fix PRs, dashboards, and a human in the loop — is an operations problem. + +- **[Elite Agentic Solutions](https://eliteagenticsolutions.com)** runs this and a larger battery of production-hardened checks as a **continuous code-health / audit-as-a-service** for teams shipping AI-assisted code. We operate the gate so your pipeline doesn't regress. +- **[CloudHostAI](https://cloudhostai.com)** offers it as a one-click hosted scanner tile for your repos. + +The library is, and will stay, free under MIT. The hosted operation is the paid product. + +--- + +## Contributing + +Issues and PRs welcome — especially new bug classes that have caused you a real silent outage, and ports to other languages. Keep each check AST-aware, read-only, and dependency-free. + +## License + +MIT © Elite Agentic Solutions. See [LICENSE](LICENSE). diff --git a/bugclass_detectors/__init__.py b/bugclass_detectors/__init__.py new file mode 100644 index 0000000..88f6c3c --- /dev/null +++ b/bugclass_detectors/__init__.py @@ -0,0 +1,16 @@ +"""bugclass-detectors — AST-aware static checks for the silent-failure bug classes +that AI-generated and fast-moving code keeps reintroducing. + +Public API: + from bugclass_detectors import run, CHECKS + findings = run(["./src"], checks=["C1", "C2", "C3", "C4", "C5"]) + +Each finding is a uniform dict: + {"check", "severity", "file", "lineno", "snippet", "why"} + +Every check is READ-ONLY (reads files only), AST/tokenizer-aware (avoids the +false positives of blind grep), and dependency-free (Python stdlib only). +""" +from .core import run, CHECKS, F, __version__ + +__all__ = ["run", "CHECKS", "F", "__version__"] diff --git a/bugclass_detectors/__main__.py b/bugclass_detectors/__main__.py new file mode 100644 index 0000000..757a73c --- /dev/null +++ b/bugclass_detectors/__main__.py @@ -0,0 +1,5 @@ +import sys +from .core import main + +if __name__ == "__main__": + sys.exit(main()) diff --git a/bugclass_detectors/core.py b/bugclass_detectors/core.py new file mode 100644 index 0000000..1463dbc --- /dev/null +++ b/bugclass_detectors/core.py @@ -0,0 +1,665 @@ +#!/usr/bin/env python3 +"""bugclass-detectors — the five generic silent-failure bug-class checks. + +These are the bug classes that look fine in review, pass tests, and keep a +process "green" while its real job silently fails. They are the classes a +fast-moving (and especially an AI-assisted) codebase reintroduces over and over. + + C1 SWALLOWED_EXCEPTION_AROUND_WRITE + A bare/catch-all `except` that swallows an exception wrapping a WRITE + (DB/cache/file/network). The producer stays alive, the data is lost, + nothing logs. The most dangerous silent-failure shape there is. + + C2 UNDEFINED_NAME + Use of a name that is never bound in its scope and is not a builtin, + import, parameter, global, comprehension var, or closure var -> a + NameError that only fires on the unlucky code path. (pyflakes F821 class, + no pyflakes dependency.) + + C3 HARDCODED_SECRET_LITERAL + A password/token/key assigned to an inline string literal (in code or a + systemd `Environment=` line) instead of being read from the environment / + a secrets store. Diverges silently the moment the real secret rotates. + + C4 FAIL_OPEN_FLAG + A safety / kill / arming / feature gate read from a backing store that is + repointable by a single env var (HOST/PORT/URL), with no independent + presence assertion. If the flag is ABSENT on the repointed store it reads + as falsy -> the gate fails OPEN instead of closed. + + C5 ROTATED_SECRET_CAPTURED_ONCE + A long-lived consumer (a daemon / `while True` loop) that captures a + secret ONCE at module/global scope. When that secret rotates, the + long-lived process keeps using the stale value and silently 401/403s. + +Every check is AST/tokenizer-aware, READ-ONLY, stdlib-only, and returns a uniform +finding dict: {check, severity, file, lineno, snippet, why}. + +Disable any check with an env flag, e.g. BUGCLASS_DISABLE=C1,C4 (no-op for those). +""" +from __future__ import annotations + +import ast +import io +import os +import re +import sys +import json +import glob +import argparse +import builtins +import tokenize + +__version__ = "0.1.0" + +SKIP_SUBSTR = ( + "/__pycache__/", "/.git/", "/node_modules/", "/.venv/", "/venv/", + "/site-packages/", "/dist-packages/", ".bak", ".pyc", +) + + +# --------------------------------------------------------------------------- # +# Shared helpers +# --------------------------------------------------------------------------- # +def F(check, severity, file, lineno, snippet, why): + return {"check": check, "severity": severity, "file": file, + "lineno": lineno, "snippet": snippet, "why": why} + + +def _read(path): + try: + with open(path, "r", encoding="utf-8", errors="replace") as fh: + return fh.read() + except Exception: + return "" + + +def _snippet(src, lineno, width=160): + lines = src.splitlines() + if 1 <= lineno <= len(lines): + return lines[lineno - 1].strip()[:width] + return "" + + +def _code_line_set(src): + """1-based line numbers that are EXECUTABLE code (not inside a comment or a + string/docstring). Returns None on tokenizer failure (treat all lines as code, + the conservative choice).""" + code_lines, str_only = set(), set() + try: + for tok in tokenize.generate_tokens(io.StringIO(src).readline): + t, (srow, _), (erow, _) = tok.type, tok.start, tok.end + if t == tokenize.STRING: + # a string spanning multiple lines is "inside a string" on its + # interior lines (docstring body); the start line may still be code. + str_only.update(range(srow + 1, erow + 1)) + elif t == tokenize.COMMENT: + pass # a comment never makes a line non-code by itself + elif t not in (tokenize.NL, tokenize.NEWLINE, tokenize.INDENT, + tokenize.DEDENT, tokenize.ENCODING, tokenize.ENDMARKER): + code_lines.add(srow) + except Exception: + return None + # a line is code if it carries any real token, even if it also has a string. + return code_lines - (str_only - code_lines) + + +def _iter_files(roots, exts): + for root in roots: + if os.path.isfile(root): + if root.endswith(exts): + yield root + continue + for dp, dns, fns in os.walk(root): + dns[:] = [d for d in dns if d not in ( + ".git", "__pycache__", "node_modules", ".venv", "venv")] + for fn in fns: + if not fn.endswith(exts): + continue + p = os.path.join(dp, fn) + if any(s in p for s in SKIP_SUBSTR): + continue + yield p + + +# ========================================================================== # +# C1 — SWALLOWED_EXCEPTION_AROUND_WRITE +# ========================================================================== # +DB_CACHE_WRITE_METHODS = { + # cache / kv + "hset", "hmset", "set", "setex", "setnx", "psetex", "mset", "msetnx", + "lpush", "rpush", "lset", "linsert", "xadd", "zadd", "zincrby", "sadd", + "incr", "incrby", "decr", "hincrby", "hincrbyfloat", "expire", "pexpire", + "expireat", "append", "getset", "hdel", "delete", "ltrim", "publish", + # sql / db / orm + "insert", "executemany", "commit", "insert_df", "insert_dataframe", + "bulk_create", +} +FILE_WRITE_METHODS = {"write", "writelines", "writerow", "writerows", "flush"} +MODULE_WRITERS = { + ("json", "dump"), ("pickle", "dump"), ("os", "replace"), ("os", "rename"), + ("shutil", "copy"), ("shutil", "copy2"), ("shutil", "move"), +} +PRIMARY_NAME_HINTS = ( + "emit", "record", "write", "snapshot", "flush", "persist", "save", + "publish", "store", "commit", "dump", "push", "send", "report", + "upload", "ingest", "sink", "export", +) +BEST_EFFORT_HINTS = ("expire", "pexpire", "ttl", "heartbeat", "close", + "cleanup", "metric") + + +def _is_catch_all(handler): + if handler.type is None: + return True + t = handler.type + if isinstance(t, ast.Name) and t.id in ("Exception", "BaseException"): + return True + if isinstance(t, ast.Tuple): + return any(isinstance(e, ast.Name) and e.id in ("Exception", "BaseException") + for e in t.elts) + return False + + +def _handler_swallow_kind(handler): + """'continue' (per-item skip), 'abandon' (pass/return/break/...), or None + (the handler logs / re-raises / does real work).""" + kind, saw = "abandon", False + for node in handler.body: + if isinstance(node, ast.Pass): + saw = True + elif isinstance(node, ast.Continue): + kind, saw = "continue", True + elif isinstance(node, ast.Break): + saw = True + elif isinstance(node, ast.Return): + if node.value is None or isinstance(node.value, ast.Constant): + saw = True + else: + return None + elif (isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant) + and node.value.value is Ellipsis): + saw = True + else: + return None + return kind if saw else None + + +def _handler_logs_or_reraises(handler): + for node in ast.walk(handler): + if isinstance(node, ast.Raise): + return True + if isinstance(node, ast.Call): + fn = node.func + name = fn.attr.lower() if isinstance(fn, ast.Attribute) else ( + fn.id.lower() if isinstance(fn, ast.Name) else "") + if any(k in name for k in ("log", "print", "warn", "error", "alert", + "exception", "critical", "capture", + "notify", "sentry", "report_error")): + return True + return False + + +def _writes_in_body(body): + hits = [] + for stmt in body: + for node in ast.walk(stmt): + if not isinstance(node, ast.Call): + continue + fn = node.func + if isinstance(fn, ast.Attribute): + ml = fn.attr.lower() + if ml in DB_CACHE_WRITE_METHODS: + hits.append((node.lineno, fn.attr, "db/cache", ml)) + elif ml in FILE_WRITE_METHODS: + hits.append((node.lineno, fn.attr, "file", ml)) + elif (isinstance(fn.value, ast.Name) + and (fn.value.id, fn.attr) in MODULE_WRITERS): + hits.append((node.lineno, f"{fn.value.id}.{fn.attr}", "file", ml)) + elif isinstance(fn, ast.Name) and fn.id == "open": + for a in node.args[1:]: + if (isinstance(a, ast.Constant) and isinstance(a.value, str) + and any(c in a.value for c in ("w", "a", "x", "+"))): + hits.append((node.lineno, "open(...,w)", "file", "open")) + return hits + + +def _enclosing_func(tree, target): + best = None + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + if getattr(node, "lineno", 0) <= target.lineno <= getattr(node, "end_lineno", 0): + if best is None or node.lineno > best.lineno: + best = node + return best + + +def _func_has_other_effects(func, try_node): + if func is None: + return True + other = 0 + for node in ast.walk(func): + if isinstance(node, ast.Return) and node.value is not None \ + and not isinstance(node.value, ast.Constant): + other += 1 + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute): + nm = node.func.attr.lower() + if (nm in DB_CACHE_WRITE_METHODS or nm in FILE_WRITE_METHODS + or "log" in nm or "print" in nm): + if not (getattr(try_node, "lineno", 0) <= node.lineno + <= getattr(try_node, "end_lineno", 1 << 30)): + other += 1 + return other > 0 + + +def check_C1_swallowed_write(path): + src = _read(path) + try: + tree = ast.parse(src) + except Exception: + return [] + lines, out = src.splitlines(), [] + for node in ast.walk(tree): + if not isinstance(node, ast.Try): + continue + writes = _writes_in_body(node.body) + if not writes: + continue + for handler in node.handlers: + if not _is_catch_all(handler) or _handler_logs_or_reraises(handler): + continue + kind = _handler_swallow_kind(handler) + if kind is None: + continue + w_line, w_name, w_kind, w_method = writes[0] + func = _enclosing_func(tree, node) + fname = func.name if func else "" + in_loop = any( + isinstance(n, (ast.For, ast.While, ast.AsyncFor)) + and getattr(n, "lineno", 0) <= node.lineno <= getattr(n, "end_lineno", 0) + for n in ast.walk(tree)) + primary = any(h in fname.lower() for h in PRIMARY_NAME_HINTS) + best_effort = (any(h in w_method for h in BEST_EFFORT_HINTS) + or any(h in w_name.lower() for h in BEST_EFFORT_HINTS)) + has_other = _func_has_other_effects(func, node) + snip = (lines[handler.lineno - 1].strip() + if 0 < handler.lineno <= len(lines) else "except: ") + wsnip = lines[w_line - 1].strip() if 0 < w_line <= len(lines) else w_name + + if best_effort and not primary: + sev = "LOW" + why = (f"catch-all swallow around a best-effort {w_kind} write " + f"`{w_method}` in {fname}() — failure is invisible but the write " + f"is non-primary (ttl/heartbeat/cleanup class).") + elif kind == "continue" and in_loop and not primary: + sev = "MED" + why = (f"`except: continue` skips one item's {w_kind} write `{w_method}` " + f"in a loop in {fname}() — fine per item, but a SYSTEMATIC failure " + f"drops the whole batch with no log. Count failures and alert.") + elif primary or not has_other: + sev = "HIGH" + reason = ("function name marks it a primary producer" + if primary else "the write is the function's ONLY output") + kn = " and the handler ABANDONS it (pass/return)" if kind == "abandon" else "" + why = (f"catch-all/bare except SILENTLY SWALLOWS a {w_kind} write " + f"`{w_method}` ({wsnip[:60]}) — {reason}{kn}. The producer looks " + f"alive, the data is lost, nothing logs. Add logging + re-raise/alert.") + else: + sev = "MED" + why = (f"catch-all swallow around a {w_kind} write `{w_method}` in " + f"{fname}() — partially visible (other effects exist) but the " + f"write can still vanish silently.") + out.append(F("C1_SWALLOWED_WRITE", sev, path, handler.lineno, snip, why)) + return out + + +# ========================================================================== # +# C2 — UNDEFINED_NAME +# ========================================================================== # +_BUILTINS = set(dir(builtins)) | { + "__file__", "__name__", "__doc__", "__package__", "__spec__", "__loader__", + "__builtins__", "__class__", "self", "cls", "_"} + + +class _ScopeCollector(ast.NodeVisitor): + def __init__(self): + self.bound = set() + + def _t(self, target): + for n in ast.walk(target): + if isinstance(n, ast.Name): + self.bound.add(n.id) + + def visit_FunctionDef(self, node): + self.bound.add(node.name) + visit_AsyncFunctionDef = visit_FunctionDef + + def visit_ClassDef(self, node): + self.bound.add(node.name) + + def visit_Assign(self, node): + for t in node.targets: + self._t(t) + self.generic_visit(node) + + def visit_AnnAssign(self, node): + if node.target: + self._t(node.target) + self.generic_visit(node) + + def visit_AugAssign(self, node): + self._t(node.target) + self.generic_visit(node) + + def visit_NamedExpr(self, node): + self._t(node.target) + self.generic_visit(node) + + def visit_For(self, node): + self._t(node.target) + self.generic_visit(node) + visit_AsyncFor = visit_For + + def visit_With(self, node): + for item in node.items: + if item.optional_vars: + self._t(item.optional_vars) + self.generic_visit(node) + visit_AsyncWith = visit_With + + def visit_Import(self, node): + for a in node.names: + self.bound.add((a.asname or a.name).split(".")[0]) + + def visit_ImportFrom(self, node): + for a in node.names: + self.bound.add(a.asname or a.name) + + def visit_Global(self, node): + self.bound.update(node.names) + + def visit_Nonlocal(self, node): + self.bound.update(node.names) + + def visit_ExceptHandler(self, node): + if node.name: + self.bound.add(node.name) + self.generic_visit(node) + + def visit_comprehension(self, node): + self._t(node.target) + self.generic_visit(node) + + +def _scope_bound(node): + c = _ScopeCollector() + for child in ast.iter_child_nodes(node): + c.visit(child) + args = getattr(node, "args", None) + if args: + for a in (args.posonlyargs + args.args + args.kwonlyargs): + c.bound.add(a.arg) + if args.vararg: + c.bound.add(args.vararg.arg) + if args.kwarg: + c.bound.add(args.kwarg.arg) + return c.bound + + +def _collect_loads(node): + loads = [] + + def walk(n): + for child in ast.iter_child_nodes(n): + if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + continue + if isinstance(child, ast.Name) and isinstance(child.ctx, ast.Load): + loads.append(child) + walk(child) + walk(node) + return loads + + +def check_C2_undefined_name(path): + src = _read(path) + try: + tree = ast.parse(src) + except Exception: + return [] + mod_bound = _scope_bound(tree) | _BUILTINS + parents = {} + for parent in ast.walk(tree): + for child in ast.iter_child_nodes(parent): + parents[child] = parent + + def enclosing_func_bound(fn): + acc, node = set(), parents.get(fn) + while node is not None: + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + acc |= _scope_bound(node) + node = parents.get(node) + return acc + + out = [] + for fn in ast.walk(tree): + if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + avail = _scope_bound(fn) | enclosing_func_bound(fn) | mod_bound + for nm in _collect_loads(fn): + if nm.id not in avail: + out.append(F("C2_UNDEFINED_NAME", "HIGH", path, nm.lineno, + _snippet(src, nm.lineno), + f"name `{nm.id}` is used in {fn.name}() but never bound in " + f"this scope and is not a builtin/import/param/closure var " + f"-> NameError on this code path.")) + return out + + +# ========================================================================== # +# C3 — HARDCODED_SECRET_LITERAL +# ========================================================================== # +_INLINE_ENV_SECRET = re.compile( + r'(?im)^\s*Environment=.*(PASS(WORD)?|SECRET|TOKEN|AUTH|API_?KEY)\s*=') +_PY_INLINE_SECRET = re.compile( + r'(?i)(password|passwd|secret|api_?key|auth_?token|access_?token|' + r'private_?key|client_?secret)\s*=\s*' + r'[\'"][A-Za-z0-9+/=_-]{16,}[\'"]') +# avoid flagging obvious placeholders / reads-from-env +_PLACEHOLDER = re.compile( + r'(?i)(xxx|placeholder|your[_-]?|example|changeme|<.*>|os\.environ|getenv|' + r'\$\{|process\.env|settings\.|config\.|^[\'"]{2}$)') + + +def check_C3_hardcoded_secret(path): + src = _read(path) + out = [] + is_unit = path.endswith((".service", ".conf", ".env")) or "/systemd/" in path + code = _code_line_set(src) if path.endswith(".py") else None + for i, ln in enumerate(src.splitlines(), 1): + if is_unit and _INLINE_ENV_SECRET.search(ln): + out.append(F("C3_HARDCODED_SECRET", "HIGH", path, i, ln.strip()[:120], + "inline unit-file secret diverges from the rotated secret on the " + "next rotation — use EnvironmentFile= / a secrets store.")) + elif path.endswith(".py") and _PY_INLINE_SECRET.search(ln): + if code is not None and i not in code: + continue + if _PLACEHOLDER.search(ln): + continue + masked = re.sub(r'[\'"][A-Za-z0-9+/=_-]{16,}[\'"]', '""', ln.strip()) + out.append(F("C3_HARDCODED_SECRET", "HIGH", path, i, masked[:120], + "secret assigned to an inline literal — it diverges silently the " + "moment the real secret rotates. Read it from the environment / a " + "secrets store instead.")) + return out + + +# ========================================================================== # +# C4 — FAIL_OPEN_FLAG +# ========================================================================== # +_GATE_KEY = re.compile( + r"(?ix)(kill_switch|killswitch|live_kill|paper_mode|arming|armed|disabled|" + r"enabled|feature_flag|safety|maintenance_mode|read_only|allow_)") +_ENV_REPOINT_CONN = re.compile( + r"(?ix)(redis\.(Strict)?Redis|redis\.from_url|ConnectionPool|\.Redis\(|" + r"connect\(|Sentinel\(|create_client\()" + r".*?(os\.environ(?:\.get)?\(['\"][A-Z_]*(PORT|HOST|URL|REDIS|DSN)|" + r"getenv\(['\"][A-Z_]*(PORT|HOST|URL|REDIS|DSN))") +_PRESENCE_ASSERT = re.compile( + r"(?ix)(is None|== *None|!= *['\"]?true|== *['\"]?true|\bexists\b|key.*missing|" + r"absent|assert\b|raise\b|fail[_-]?closed|default.*true|or\s+['\"]?true|" + r"not\s+\w+.*:\s*(return|raise|sys\.exit)|\bif\s+not\b)") + + +def check_C4_fail_open_flag(path): + src = _read(path) + if not (_GATE_KEY.search(src) and _ENV_REPOINT_CONN.search(src)): + return [] + lines, out = src.splitlines(), [] + code = _code_line_set(src) + for i, ln in enumerate(lines, 1): + if code is not None and i not in code: + continue + if not _GATE_KEY.search(ln): + continue + if not re.search(r"(?i)(get|mget|hget|read|fetch)", ln): + continue + if ln.lstrip().startswith("#"): + continue + lo, hi = max(0, i - 6), min(len(lines), i + 6) + if _PRESENCE_ASSERT.search("\n".join(lines[lo:hi])): + continue + out.append(F("C4_FAIL_OPEN_FLAG", "HIGH", path, i, ln.strip()[:160], + "a safety / kill / feature gate is read from an env-repointable store " + "(HOST/PORT/URL) with no independent presence assert — if the flag is " + "ABSENT on the repointed store it reads falsy and the gate fails OPEN. " + "Re-assert the flag exists, or pin the flag-store separately.")) + return out + + +# ========================================================================== # +# C5 — ROTATED_SECRET_CAPTURED_ONCE +# ========================================================================== # +_LONGLIVED = re.compile( + r"(?ix)(while\s+True|asyncio\.run|\.run_forever|Consumer\(|consume\(|" + r"app\.run\(|serve_forever|schedule\.|\.start\(\))") +# A MODULE-LEVEL (column-0) capture of a secret-named variable from the env. +_MODULE_SECRET_CAPTURE = re.compile( + r"^(?P[A-Za-z_][A-Za-z0-9_]*)\s*=\s*(os\.environ|getenv|open\()") +_SECRET_NAME = re.compile(r"(?i)(pw|pass|passwd|secret|token|auth|api_?key)") + + +def _loop_line_ranges(tree): + """(start, end) line spans of every long-lived loop construct.""" + spans = [] + for n in ast.walk(tree): + if isinstance(n, (ast.While, ast.AsyncFor)): + spans.append((n.lineno, getattr(n, "end_lineno", n.lineno))) + return spans + + +def check_C5_rotated_secret(path): + src = _read(path) + if not _LONGLIVED.search(src): + return [] + try: + tree = ast.parse(src) + except Exception: + tree = None + # collect names referenced inside long-lived loops (so we only flag captures + # the loop actually USES — a module read never touched in the loop is fine). + used_in_loop = set() + loop_spans = _loop_line_ranges(tree) if tree else [] + if tree: + for node in ast.walk(tree): + if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Load): + if any(lo <= node.lineno <= hi for lo, hi in loop_spans): + used_in_loop.add(node.id) + out = [] + for i, ln in enumerate(src.splitlines(), 1): + if ln[:1] in (" ", "\t"): # module level only (column 0) + continue + m = _MODULE_SECRET_CAPTURE.match(ln) + if not m or not _SECRET_NAME.search(m.group("name")): + continue + # only a real risk if the long-lived loop actually consumes this name + if used_in_loop and m.group("name") not in used_in_loop: + continue + out.append(F("C5_ROTATED_SECRET", "MED", path, i, ln.strip()[:120], + "a long-lived consumer captures a secret ONCE at module/global " + "scope; when it rotates the process keeps the stale value and " + "silently 401/403s. Re-read the secret per (re)connection, or " + "ensure a rotation-aware reconnect covers this consumer.")) + return out + + +# --------------------------------------------------------------------------- # +# Registry + runner +# --------------------------------------------------------------------------- # +CHECKS = { + "C1": ("SWALLOWED_EXCEPTION_AROUND_WRITE", check_C1_swallowed_write, (".py",)), + "C2": ("UNDEFINED_NAME", check_C2_undefined_name, (".py",)), + "C3": ("HARDCODED_SECRET_LITERAL", check_C3_hardcoded_secret, + (".py", ".service", ".conf", ".env")), + "C4": ("FAIL_OPEN_FLAG", check_C4_fail_open_flag, (".py",)), + "C5": ("ROTATED_SECRET_CAPTURED_ONCE", check_C5_rotated_secret, (".py",)), +} +_SEV_RANK = {"HIGH": 0, "MED": 1, "LOW": 2} + + +def run(roots, checks=None, max_n=None, min_sev="LOW"): + """Run the selected checks over `roots`. Returns a list of finding dicts.""" + disabled = {c.strip() for c in os.environ.get("BUGCLASS_DISABLE", "").split(",") if c.strip()} + checks = [c for c in (checks or list(CHECKS)) if c in CHECKS and c not in disabled] + findings = [] + for cid in checks: + _name, fn, exts = CHECKS[cid] + for path in _iter_files(roots, exts): + try: + findings.extend(fn(path)) + except Exception: + continue + allow = {s for s in _SEV_RANK if _SEV_RANK[s] <= _SEV_RANK.get(min_sev, 2)} + findings = [f for f in findings if f["severity"] in allow] + findings.sort(key=lambda f: (_SEV_RANK.get(f["severity"], 9), f["check"], + f["file"], f["lineno"])) + return findings[:max_n] if max_n else findings + + +def main(argv=None): + ap = argparse.ArgumentParser( + prog="bugclass-detectors", + description="AST-aware static checks for silent-failure bug classes.") + ap.add_argument("roots", nargs="*", default=["."], + help="files or directories to scan (default: .)") + ap.add_argument("--checks", default=",".join(CHECKS), + help="comma list of check ids, e.g. C1,C3,C4") + ap.add_argument("--json", action="store_true", help="machine-readable output") + ap.add_argument("--lint", action="store_true", + help="exit 1 if any HIGH finding (CI gate mode)") + ap.add_argument("--max", type=int, default=None) + ap.add_argument("--min-sev", default="LOW", choices=["HIGH", "MED", "LOW"]) + args = ap.parse_args(argv) + + checks = [c.strip() for c in args.checks.split(",") if c.strip() in CHECKS] + findings = run(args.roots or ["."], checks, args.max, args.min_sev) + + if args.json: + print(json.dumps(findings, indent=2)) + else: + hi = sum(1 for f in findings if f["severity"] == "HIGH") + md = sum(1 for f in findings if f["severity"] == "MED") + lo = sum(1 for f in findings if f["severity"] == "LOW") + print(f"bugclass-detectors {__version__}: {len(findings)} findings " + f"(HIGH={hi} MED={md} LOW={lo}) roots={args.roots or ['.']}\n") + for f in findings: + print(f" {f['severity']:4} [{f['check']}] {f['file']}:{f['lineno']}") + print(f" {f['snippet'][:88]}") + print(f" -> {f['why'][:150]}") + + if args.lint and any(f["severity"] == "HIGH" for f in findings): + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/examples/buggy_sample.py b/examples/buggy_sample.py new file mode 100644 index 0000000..046f895 --- /dev/null +++ b/examples/buggy_sample.py @@ -0,0 +1,42 @@ +"""Intentionally-buggy fixture demonstrating each bug class. Run: + + python -m bugclass_detectors examples/buggy_sample.py + +Every flagged line is a real silent-failure shape this tool catches. +""" +import os +import redis + +r = redis.Redis(host=os.environ.get("CACHE_HOST", "localhost")) + +# --- C3: hardcoded secret literal ----------------------------------------- +DB_PASSWORD = "s3cr3t_inline_password_value" # noqa (should read from env) + + +# --- C1: swallowed exception around a primary write ----------------------- +def record_metric(name, value): + try: + r.hset("metrics", name, value) # the function's ONLY output + except Exception: + pass # producer looks alive, data lost, no log + + +# --- C2: undefined name --------------------------------------------------- +def compute_total(items): + for item in items: + running += item # `running` never initialized -> NameError + return running + + +# --- C4: fail-open feature/kill flag -------------------------------------- +def is_killed(): + client = redis.from_url(os.environ.get("FLAG_REDIS_URL", "redis://localhost")) + return client.get("kill_switch") # on a repointed store this reads None and fails OPEN + + +# --- C5: rotated secret captured once at module scope --------------------- +API_TOKEN = os.environ.get("API_TOKEN") # captured once + +def daemon(): + while True: + call_api(API_TOKEN) # stale after rotation; never re-read diff --git a/examples/clean_sample.py b/examples/clean_sample.py new file mode 100644 index 0000000..ea769cc --- /dev/null +++ b/examples/clean_sample.py @@ -0,0 +1,42 @@ +"""Clean fixture — the corrected version of every pattern in buggy_sample.py. +The detector should report ZERO findings here (false-positive guard).""" +import os +import logging +import redis + +log = logging.getLogger(__name__) +DB_PASSWORD = os.environ["DB_PASSWORD"] # read from env, not inline + + +def record_metric(name, value): + try: + r = redis.Redis(host=os.environ.get("CACHE_HOST", "localhost")) + r.hset("metrics", name, value) + except Exception: + log.exception("metric write failed for %s", name) # logs, not swallowed + raise + + +def compute_total(items): + running = 0 # initialized + for item in items: + running += item + return running + + +def is_killed(): + client = redis.from_url(os.environ.get("FLAG_REDIS_URL", "redis://localhost")) + val = client.get("kill_switch") + if val is None: # presence assert -> fail CLOSED + raise RuntimeError("kill flag missing; refusing to arm") + return val == b"1" + + +def daemon(): + while True: + token = os.environ.get("API_TOKEN") # re-read each loop, rotation-safe + call_api(token) + + +def call_api(token): + ... diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..fd6cd92 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,31 @@ +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "bugclass-detectors" +version = "0.1.0" +description = "AST-aware static checks for the silent-failure bug classes that AI-generated and fast-moving code keeps reintroducing." +readme = "README.md" +license = { text = "MIT" } +requires-python = ">=3.8" +authors = [{ name = "Elite Agentic Solutions" }] +keywords = ["static-analysis", "lint", "ast", "ci", "code-quality", "ai-code", "bug-detection"] +classifiers = [ + "Development Status :: 4 - Beta", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Topic :: Software Development :: Quality Assurance", +] +dependencies = [] + +[project.urls] +Homepage = "https://eliteagenticsolutions.com" +Repository = "https://github.com/elite-agentic-solutions/bugclass-detectors" + +[project.scripts] +bugclass-detectors = "bugclass_detectors.core:main" + +[tool.setuptools] +packages = ["bugclass_detectors"] diff --git a/tests/test_detectors.py b/tests/test_detectors.py new file mode 100644 index 0000000..3e1c953 --- /dev/null +++ b/tests/test_detectors.py @@ -0,0 +1,49 @@ +"""Smoke tests: the buggy fixture trips each check; the clean fixture trips none.""" +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from bugclass_detectors import run # noqa: E402 + +HERE = os.path.dirname(os.path.abspath(__file__)) +EX = os.path.join(HERE, "..", "examples") +BUGGY = os.path.join(EX, "buggy_sample.py") +CLEAN = os.path.join(EX, "clean_sample.py") + + +def _checks(findings): + return {f["check"].split("_")[0] for f in findings} + + +def test_buggy_trips_each_class(): + found = _checks(run([BUGGY])) + for cid in ("C1", "C2", "C3", "C4", "C5"): + assert cid in found, f"{cid} should have fired on the buggy fixture; got {found}" + + +def test_clean_has_no_findings(): + findings = run([CLEAN]) + assert findings == [], f"clean fixture should be silent, got: {findings}" + + +def test_disable_env_suppresses_check(): + os.environ["BUGCLASS_DISABLE"] = "C1,C2,C3,C4,C5" + try: + assert run([BUGGY]) == [] + finally: + del os.environ["BUGCLASS_DISABLE"] + + +def test_finding_shape(): + for f in run([BUGGY]): + assert set(f) == {"check", "severity", "file", "lineno", "snippet", "why"} + assert f["severity"] in ("HIGH", "MED", "LOW") + + +if __name__ == "__main__": + test_buggy_trips_each_class() + test_clean_has_no_findings() + test_disable_env_suppresses_check() + test_finding_shape() + print("all tests passed")