eas-oss/bugclass-detectors
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

bugclass-detectors v0.1.0 — five silent-failure bug-class checks (MIT)

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Elite Agentic Solutions 2026-06-12 01:29:50 -07:00
commit 7ee5300d43
11 changed files with 1000 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
.github/workflows/ci.yml vendored Normal file
View file

@ -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

7
.gitignore vendored Normal file
View file

@ -0,0 +1,7 @@
__pycache__/
*.pyc
*.egg-info/
build/
dist/
.venv/
.pytest_cache/

21
LICENSE Normal file
View file

@ -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.

102
README.md Normal file
View file

@ -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/<org>/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).

16
bugclass_detectors/__init__.py Normal file
View file

@ -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__"]

5
bugclass_detectors/__main__.py Normal file
View file

@ -0,0 +1,5 @@
import sys
from .core import main
if __name__ == "__main__":
sys.exit(main())

665
bugclass_detectors/core.py Normal file
View file

@ -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 "<module>"
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: <swallow>")
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,}[\'"]', '"<masked>"', 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<name>[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())

42
examples/buggy_sample.py Normal file
View file

@ -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

42
examples/clean_sample.py Normal file
View file

@ -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):
...

31
pyproject.toml Normal file
View file

@ -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"]

49
tests/test_detectors.py Normal file
View file

@ -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")