"""erp-lint — static checks for the bug classes the wave audits proved, run over a brick's src/.
erp-lint <brick-dir> [--method-dir DIR] exit 1 on findings, file:line: id name — why
Every rule was a MUST in an audit. They are heuristics tuned for few false positives, not proofs: a finding is
"look here", and `# erp-lint: allow <id|name> <reason>` on (or the line above) the flagged line records a
reviewed exception. Rules (heuristic in brackets):
L1 utc-today date.today(), datetime.utcnow(), naive datetime.now(), or the .date() of a UTC now
[skips business_time*/clock files and tests; timestamps via now(tz) are fine]
L2 self-approval-one-identity a function with "approv" in its name that compares email but not sub, or sub but not email
L3 swallow-in-txn `except Exception`/bare except that does not re-raise, inside a `with <repository/session/
transaction>` block, with more work on the same handle after it
L4 int-on-shared-varchar int(<expr naming a VARCHAR key from modules/shared.yaml>) [needs --method-dir or ERP_METHOD_DIR]
L5 unbounded-money-product a * b of two money/qty/rate-named operands in a function that never calls a bound/validate/
quantize helper [name-based; a product of anything else is never flagged]
L6 serial-sequence SERIAL, AUTOINCREMENT, nextval, or max(...)+1 allocation (py and sql)
L7 fake-without-contract tests/ defines Fake* classes of a neighbour brick (not Odoo/Maps/Objects/Mail/Clock) but nothing there enables erp_platform.contract
(checked_seams or guard)
"""
from __future__ import annotations
import ast
import os
import re
import sys
from pathlib import Path
from typing import Iterable, Optional
RULES = {
"L1": ("utc-today", "a business date is computed in the business zone (business_time), never from the server's UTC clock"),
"L2": ("self-approval-one-identity", "an approver has an email AND a sub; refuse when ANY identity matches the requester"),
"L3": ("swallow-in-txn", "swallowing an error inside a transaction then writing more commits half of an operation"),
"L4": ("int-on-shared-varchar", "this key is a VARCHAR across modules (modules/shared.yaml); int() breaks on a migrated or customer key"),
"L5": ("unbounded-money-product", "a product of two amounts/rates can overflow NUMERIC; validate the bounds so it is a 422, not a 500"),
"L6": ("serial-sequence", "the engine has no sequences; number allocation must be a natural key or a claim row"),
"L7": ("fake-without-contract", "a fake that disagrees with the callee's spec hides broken seams; enable erp_platform.contract"),
}
_ALLOW = re.compile(r"#\s*erp-lint:\s*allow\s+(\S+)")
_MONEY = re.compile(r"(amount|price|rate|qty|quantity|cost|total|value|pct|percent|charge|fee|balance|discount)", re.I)
_BOUND = re.compile(r"(bound|validate|quantize|check_|clamp|finite|money|round_|ensure_|require_)", re.I)
_CLOCK_FILE = re.compile(r"(business_time|clock|democlock)", re.I)
_DB_HANDLE = re.compile(r"(repo|repository|session|conn|txn|tx|transaction|begin|engine)", re.I)
class Finding:
def __init__(self, path, line, rule, detail=""):
self.path, self.line, self.rule, self.detail = path, line, rule, detail
def __str__(self):
name, why = RULES[self.rule]
d = f" [{self.detail}]" if self.detail else ""
return f"{self.path}:{self.line}: {self.rule} {name}{d} — {why}"
def _name(n) -> str:
return n.attr if isinstance(n, ast.Attribute) else n.id if isinstance(n, ast.Name) else ""
def _text(n) -> str:
try:
return ast.unparse(n)
except Exception:
return ""
def _shared_varchars(method_dir: Optional[str]) -> set:
base = Path(method_dir or os.environ.get("ERP_METHOD_DIR", "") or ".")
for cand in (base / "modules" / "shared.yaml", base / "shared.yaml"):
if cand.is_file():
import yaml
keys = (yaml.safe_load(cand.read_text()) or {}).get("keys", {})
return {k for k, v in keys.items() if str(v.get("type", "")).upper().startswith("VARCHAR")}
return set()
def _l1(tree, rel):
if _CLOCK_FILE.search(Path(rel).name):
return
for n in ast.walk(tree):
if not isinstance(n, ast.Call):
continue
t = _text(n.func)
if t in ("date.today", "datetime.date.today") or t.endswith(".utcnow"):
yield Finding(rel, n.lineno, "L1", t + "()")
elif t.endswith("datetime.now") and not n.args and not n.keywords:
yield Finding(rel, n.lineno, "L1", "naive now()")
elif t.endswith(".date") and isinstance(n.func, ast.Attribute) and isinstance(n.func.value, ast.Call):
inner = n.func.value
it = _text(inner.func)
if it.endswith("datetime.now") or it.endswith(".utcnow") or it == "now":
arg = _text(inner)
if "utc" in arg.lower():
yield Finding(rel, n.lineno, "L1", ".date() of a UTC now")
def _l2(tree, rel):
for fn in ast.walk(tree):
if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)) or "approv" not in fn.name.lower():
continue
email = sub = False
for c in ast.walk(fn):
if isinstance(c, ast.Compare):
s = _text(c).lower()
email |= "email" in s
sub |= bool(re.search(r"(\bsub\b|_sub\b|\bsub_|['\"]sub['\"])", s))
if email != sub:
yield Finding(rel, fn.lineno, "L2", f"{fn.name} compares {'email' if email else 'sub'} only")
def _l3(tree, rel):
for w in ast.walk(tree):
if not isinstance(w, (ast.With, ast.AsyncWith)):
continue
handles = set()
txn = False
for it in w.items:
if _DB_HANDLE.search(_text(it.context_expr)):
txn = True
if isinstance(it.optional_vars, ast.Name):
handles.add(it.optional_vars.id)
if not txn:
continue
body = w.body
for i, st in enumerate(body):
if not isinstance(st, ast.Try):
continue
for h in st.handlers:
broad = h.type is None or _name(h.type) in ("Exception", "BaseException")
if not broad or any(isinstance(x, ast.Raise) for x in ast.walk(h)):
continue
later = body[i + 1:]
uses = any(
isinstance(c, ast.Call) and isinstance(c.func, ast.Attribute)
and isinstance(c.func.value, ast.Name) and c.func.value.id in handles
for s in later for c in ast.walk(s))
if uses or (not handles and later):
yield Finding(rel, h.lineno, "L3", "except Exception without re-raise, then more work")
def _direct_keys(n) -> set:
"""The names an int() argument IS (not merely mentions): x, self.x, row["x"], row.get("x"), str(x)."""
if isinstance(n, ast.Name):
return {n.id}
if isinstance(n, ast.Attribute):
return {n.attr}
if isinstance(n, ast.Subscript):
sl = n.slice
if isinstance(sl, ast.Constant) and isinstance(sl.value, str):
return {sl.value}
return set() # a slice of the value (period_id[:4]) is parsing it, not converting the key
if isinstance(n, ast.Call):
if _name(n.func) in ("get", "pop") and n.args and isinstance(n.args[0], ast.Constant) and isinstance(n.args[0].value, str):
return {n.args[0].value}
if _name(n.func) in ("str", "strip") and n.args:
return _direct_keys(n.args[0])
return set()
def _l4(tree, rel, varchars):
if not varchars:
return
for n in ast.walk(tree):
if isinstance(n, ast.Call) and _name(n.func) == "int" and n.args:
hit = _direct_keys(n.args[0]) & varchars
if hit:
yield Finding(rel, n.lineno, "L4", f"int({sorted(hit)[0]})")
def _moneyish(x) -> bool:
return isinstance(x, (ast.Name, ast.Attribute, ast.Subscript)) and bool(_MONEY.search(_text(x)))
def _l5(tree, rel):
for fn in ast.walk(tree):
if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
calls = [_name(c.func) for c in ast.walk(fn) if isinstance(c, ast.Call)]
if any(_BOUND.search(c) for c in calls) or _BOUND.search(fn.name):
continue
for b in ast.walk(fn):
if isinstance(b, ast.BinOp) and isinstance(b.op, ast.Mult):
if _moneyish(b.left) and _moneyish(b.right):
yield Finding(rel, b.lineno, "L5", _text(b)[:60])
# a column definition (`name SERIAL`, upper-case as DDL is written), AUTOINCREMENT, nextval(), or max(...)+1
# [the max() argument is skipped when it names a line/revision/step counter — numbering WITHIN one document]
_L6 = re.compile(r"((?<!never )(?<!no )(?<!not )\b(?:BIG|SMALL)?SERIAL\b|\bAUTOINCREMENT\b|\bautoincrement\s*=\s*(?!False)|\bnextval\b|\bmax\s*\(([^()]*(?:\([^()]*\))?[^()]*)\)\s*\)?\s*\+\s*1\b)", re.I)
_L6_INLINE = re.compile(r"(line_no|revision|level|step|seq_in|\bline\b|row)", re.I)
def _l6_lines(text, rel):
for i, ln in enumerate(text.splitlines(), 1):
s = ln.split("--")[0] if rel.endswith(".sql") else re.sub(r"#.*$", "", ln)
for m in _L6.finditer(s):
g = m.group(0)
if g == "autoincrement" or (g.upper().endswith("SERIAL") and "SERIAL" not in g):
continue # `serial` the noun (serial numbers), not the column type
if m.group(2) is not None and _L6_INLINE.search(m.group(2)):
continue
yield Finding(rel, i, "L6", g[:30])
break
def _l7(brick: Path):
tests = brick / "tests"
if not tests.is_dir():
return
files = [p for p in tests.rglob("*.py")]
first = None
enabled = False
for p in files:
t = p.read_text(errors="ignore")
if re.search(r"erp_platform(\.| import )contract|checked_seams|\bguard\(", t) and re.search(r"checked_seams|contract\.guard|import guard|\bguard\(", t):
enabled = True
if first is None:
# fakes of another BRICK's port only: external systems (Odoo, object store, maps, mail, clock) have no spec
m = re.search(r"^class (Fake(?!Odoo|Maps|Objects|Mail|Clock|Bank|Time)\w*)", t, re.M)
if m:
first = (p, t[:m.start()].count("\n") + 1)
if first and not enabled:
yield Finding(str(first[0].relative_to(brick)), first[1], "L7", "no checked_seams/guard anywhere in tests/")
def _suppressed(lines, f: Finding) -> bool:
name = RULES[f.rule][0]
for ln in (f.line, f.line - 1):
if 1 <= ln <= len(lines):
m = _ALLOW.search(lines[ln - 1])
if m and m.group(1) in (f.rule, name):
return True
return False
def lint(brick: str, method_dir: Optional[str] = None) -> list:
root = Path(brick)
src = root / "src"
varchars = _shared_varchars(method_dir)
out: list = []
for p in sorted(src.rglob("*")):
if p.suffix not in (".py", ".sql") or "__pycache__" in p.parts:
continue
rel = str(p.relative_to(root))
text = p.read_text(errors="ignore")
found: list = list(_l6_lines(text, rel))
if p.suffix == ".py":
try:
tree = ast.parse(text)
except SyntaxError:
continue
for fnc in (_l1, _l2, _l3, _l5):
found += list(fnc(tree, rel))
found += list(_l4(tree, rel, varchars))
lines = text.splitlines()
seen = set()
for f in found:
k = (f.line, f.rule, f.detail)
if k not in seen and not _suppressed(lines, f):
seen.add(k)
out.append(f)
for f in _l7(root):
p = root / f.path
if not _suppressed(p.read_text(errors="ignore").splitlines(), f):
out.append(f)
return out
def main(argv: Optional[Iterable[str]] = None) -> int:
a = list(sys.argv[1:] if argv is None else argv)
md = None
if "--method-dir" in a:
i = a.index("--method-dir")
md = a[i + 1]
del a[i:i + 2]
if len(a) != 1:
print("usage: erp-lint <brick-dir> [--method-dir DIR]", file=sys.stderr)
return 2
if not _shared_varchars(md) and not os.environ.get("ERP_METHOD_DIR") and not md:
print("erp-lint: L4 skipped (no --method-dir / ERP_METHOD_DIR)", file=sys.stderr)
fs = lint(a[0], md)
for f in fs:
print(f)
print(f"erp-lint: {len(fs)} finding(s)", file=sys.stderr)
return 1 if fs else 0
if __name__ == "__main__":
sys.exit(main())