From ad117a1adf702fbc3f33b86f4731d4ab60104517 Mon Sep 17 00:00:00 2001 From: Adrian Ehrsam Date: Tue, 29 Sep 2026 20:07:01 +0200 Subject: [PATCH 1/3] fix(find-injection): allow shell=True arg lists, bare SQL(), skip csp-weakened in tests; bump to 0.28.0 Co-Authored-By: Claude Code --- bmsdna/devtools/find_injection.py | 3 ++- bmsdna/devtools/injection_python.py | 7 ++++++- bmsdna/devtools/lint_sql.py | 3 ++- pyproject.toml | 2 +- tests/test_find_injection.py | 32 +++++++++++++++++++++++++++++ uv.lock | 2 +- 6 files changed, 44 insertions(+), 5 deletions(-) diff --git a/bmsdna/devtools/find_injection.py b/bmsdna/devtools/find_injection.py index fbccde0..fd2cacf 100644 --- a/bmsdna/devtools/find_injection.py +++ b/bmsdna/devtools/find_injection.py @@ -265,7 +265,8 @@ def check_csp(root: Path, scanned: list[Path], scope: _ScanFilter, only: set[Pat continue if has_csp(text): found = True - weakened.extend(find_csp_weakening(candidate, text)) + if not _is_test_file(candidate, root): + weakened.extend(find_csp_weakening(candidate, text)) if found: return [f for f in weakened if only is None or f.path.resolve() in only] return [ diff --git a/bmsdna/devtools/injection_python.py b/bmsdna/devtools/injection_python.py index 5e3c7f2..c72de6b 100644 --- a/bmsdna/devtools/injection_python.py +++ b/bmsdna/devtools/injection_python.py @@ -50,6 +50,11 @@ def _is_literal(node: ast.expr) -> bool: return isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add) and _is_literal(node.left) and _is_literal(node.right) +def _is_arg_list(node: ast.expr | None) -> bool: + """A list/tuple command whose program is a literal: extra (even dynamic) items are arguments, not shell code.""" + return isinstance(node, (ast.List, ast.Tuple)) and bool(node.elts) and _is_literal(node.elts[0]) + + def _shell_true(call: ast.Call) -> bool: return any(kw.arg == "shell" and isinstance(kw.value, ast.Constant) and kw.value.value is True for kw in call.keywords) @@ -84,7 +89,7 @@ def _check_call(path: Path, node: ast.Call, aliases: dict[str, str]) -> list[Fin "error", ) ] - if name.startswith("subprocess.") and _shell_true(node): + if name.startswith("subprocess.") and _shell_true(node) and not _is_arg_list(first_arg): if dynamic or isinstance(first_arg, (ast.JoinedStr, ast.BinOp)): return [ _finding( diff --git a/bmsdna/devtools/lint_sql.py b/bmsdna/devtools/lint_sql.py index a22b295..412e9e3 100644 --- a/bmsdna/devtools/lint_sql.py +++ b/bmsdna/devtools/lint_sql.py @@ -109,7 +109,8 @@ def _is_sql_composed_call(node: ast.AST) -> bool: if not isinstance(node, ast.Call): return False func = node.func - return isinstance(func, ast.Attribute) and func.attr in ("SQL", "Identifier", "Composed") + name = func.attr if isinstance(func, ast.Attribute) else func.id if isinstance(func, ast.Name) else "" + return name in ("SQL", "Identifier", "Composed") _UNWRAP_METHODS = frozenset({"strip", "lstrip", "rstrip"}) diff --git a/pyproject.toml b/pyproject.toml index 04e32f9..2d2eede 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,7 +10,7 @@ packages = ["bmsdna"] [project] name = "bmsdna-devtools" -version = "0.27.0" +version = "0.28.0" description = "Shared Azure DevOps / GitHub / git / Azure Monitor developer tooling for BMS projects" readme = "README.md" requires-python = ">=3.14" # pgdevkit>=0.7.1 requires 3.14; was >=3.11 before adding it as a dependency diff --git a/tests/test_find_injection.py b/tests/test_find_injection.py index 5652b74..b608fb3 100644 --- a/tests/test_find_injection.py +++ b/tests/test_find_injection.py @@ -298,3 +298,35 @@ def run(conn, t): """ ) assert find_injection.run([], root=tmp_path).findings == [] + + +def test_subprocess_arg_list_with_shell_true_and_bare_sql(tmp_path: Path) -> None: + (tmp_path / "ok.py").write_text( + """import subprocess +from psycopg.sql import SQL + + +def run(conn, target): + subprocess.check_call(["bun", "run", "build"], shell=True) + subprocess.check_call(["bun", "run", target], cwd=target, shell=True) + conn.execute(SQL("UPDATE t SET a = 1 WHERE id = %s"), (1,)) +""" + ) + (tmp_path / "bad.py").write_text( + """import subprocess + + +def run(cmd, target): + subprocess.check_call([cmd, "x"], shell=True) + subprocess.check_call(f"bun run {target}", shell=True) +""" + ) + result = find_injection.run([], root=tmp_path) + assert [(f.path.name, f.line, f.rule) for f in result.findings] == [("bad.py", 5, "py-shell-command"), ("bad.py", 6, "py-shell-command")] + + +def test_csp_weakening_in_test_files_is_ignored(tmp_path: Path) -> None: + (tmp_path / "app.js").write_text("run();\n") + (tmp_path / "web.config").write_text("Content-Security-Policy: default-src 'self'") + (tmp_path / "test_headers.py").write_text("H = \"Content-Security-Policy: script-src 'unsafe-eval'\"\n") + assert find_injection.run([], root=tmp_path).findings == [] diff --git a/uv.lock b/uv.lock index c1703b5..c4c35b9 100644 --- a/uv.lock +++ b/uv.lock @@ -22,7 +22,7 @@ wheels = [ [[package]] name = "bmsdna-devtools" -version = "0.27.0" +version = "0.28.0" source = { editable = "." } dependencies = [ { name = "pgdevkit", extra = ["db"] }, From 4225e84199df4f2fed22c32ec0c4743a871fa553 Mon Sep 17 00:00:00 2001 From: Adrian Ehrsam Date: Tue, 29 Sep 2026 20:12:21 +0200 Subject: [PATCH 2/3] feat(find-injection): treat names assigned only constants as constant in Python sinks Co-Authored-By: Claude Code --- bmsdna/devtools/injection_python.py | 201 ++++++++++++++++++++++++++-- tests/test_find_injection.py | 62 +++++++++ 2 files changed, 252 insertions(+), 11 deletions(-) diff --git a/bmsdna/devtools/injection_python.py b/bmsdna/devtools/injection_python.py index c72de6b..a073cc3 100644 --- a/bmsdna/devtools/injection_python.py +++ b/bmsdna/devtools/injection_python.py @@ -44,15 +44,195 @@ def _dotted(func: ast.expr, aliases: dict[str, str]) -> str: return ".".join(reversed(parts)) -def _is_literal(node: ast.expr) -> bool: +class _Scope: + """Every binding of each name in one function/module scope. `None` marks a binding whose value + isn't statically known (parameter, loop over a variable, `+=`, import, ...).""" + + def __init__(self, parent: _Scope | None, is_class: bool = False) -> None: + self.parent = parent + self.is_class = is_class + self.bindings: dict[str, list[ast.expr | None]] = {} + self.global_names: set[str] = set() + + +def _target_names(target: ast.AST) -> list[str]: + return [n.id for n in ast.walk(target) if isinstance(n, ast.Name)] + + +class _ScopeBuilder(ast.NodeVisitor): + def __init__(self, tree: ast.Module) -> None: + self.module = _Scope(None) + self.scope = self.module + self.calls: list[tuple[ast.Call, _Scope]] = [] + self._body(tree) + + def _body(self, node: ast.AST) -> None: + for child in ast.iter_child_nodes(node): + self.visit(child) + + def _bind(self, name: str, value: ast.expr | None) -> None: + scope = self.module if name in self.scope.global_names else self.scope + scope.bindings.setdefault(name, []).append(value) + + def _unknown(self, target: ast.AST) -> None: + for name in _target_names(target): + self._bind(name, None) + + def _function(self, node: ast.FunctionDef | ast.AsyncFunctionDef | ast.Lambda) -> None: + if not isinstance(node, ast.Lambda): + self._bind(node.name, None) + for deco in node.decorator_list: + self.visit(deco) + for default in [*node.args.defaults, *(d for d in node.args.kw_defaults if d is not None)]: + self.visit(default) + outer = self.scope + parent = outer + while parent.is_class and parent.parent is not None: + parent = parent.parent + self.scope = _Scope(parent) + args = node.args + for a in [*args.posonlyargs, *args.args, *args.kwonlyargs, *(x for x in (args.vararg, args.kwarg) if x)]: + self._bind(a.arg, None) + for stmt in node.body if isinstance(node.body, list) else [node.body]: + self.visit(stmt) + self.scope = outer + + visit_FunctionDef = visit_AsyncFunctionDef = visit_Lambda = _function + + def visit_ClassDef(self, node: ast.ClassDef) -> None: + self._bind(node.name, None) + for expr in [*node.bases, *(k.value for k in node.keywords), *node.decorator_list]: + self.visit(expr) + outer = self.scope + self.scope = _Scope(outer, is_class=True) + for stmt in node.body: + self.visit(stmt) + self.scope = outer + + def visit_Global(self, node: ast.Global) -> None: + self.scope.global_names.update(node.names) + for name in node.names: + self.module.bindings.setdefault(name, []) + + def visit_Nonlocal(self, node: ast.Nonlocal) -> None: + scope = self.scope.parent + while scope is not None and scope is not self.module: + for name in node.names: + scope.bindings.setdefault(name, []).append(None) + scope = scope.parent + + def visit_Assign(self, node: ast.Assign) -> None: + self.visit(node.value) + for target in node.targets: + if isinstance(target, ast.Name): + self._bind(target.id, node.value) + else: + self._unknown(target) + + def visit_AnnAssign(self, node: ast.AnnAssign) -> None: + if node.value is not None: + self.visit(node.value) + if isinstance(node.target, ast.Name): + self._bind(node.target.id, node.value) + else: + self._unknown(node.target) + + def visit_AugAssign(self, node: ast.AugAssign) -> None: + self.visit(node.value) + self._unknown(node.target) + + def visit_NamedExpr(self, node: ast.NamedExpr) -> None: + self.visit(node.value) + self._bind(node.target.id, None) + + def visit_For(self, node: ast.For | ast.AsyncFor) -> None: + self.visit(node.iter) + if isinstance(node.target, ast.Name) and isinstance(node.iter, (ast.List, ast.Tuple, ast.Set)): + for element in node.iter.elts: + self._bind(node.target.id, element) + else: + self._unknown(node.target) + for stmt in [*node.body, *node.orelse]: + self.visit(stmt) + + visit_AsyncFor = visit_For + + def visit_comprehension(self, node: ast.comprehension) -> None: + self._unknown(node.target) + self._body(node) + + def visit_withitem(self, node: ast.withitem) -> None: + if node.optional_vars is not None: + self._unknown(node.optional_vars) + self._body(node) + + def visit_ExceptHandler(self, node: ast.ExceptHandler) -> None: + if node.name: + self._bind(node.name, None) + self._body(node) + + def visit_Import(self, node: ast.Import) -> None: + for a in node.names: + self._bind((a.asname or a.name).split(".")[0], None) + + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + for a in node.names: + self._bind(a.asname or a.name, None) + + def visit_Delete(self, node: ast.Delete) -> None: + for target in node.targets: + self._unknown(target) + + def visit_MatchAs(self, node: ast.MatchAs) -> None: + if node.name: + self._bind(node.name, None) + self._body(node) + + def visit_MatchStar(self, node: ast.MatchStar) -> None: + if node.name: + self._bind(node.name, None) + + def visit_MatchMapping(self, node: ast.MatchMapping) -> None: + if node.rest: + self._bind(node.rest, None) + self._body(node) + + def visit_Call(self, node: ast.Call) -> None: + self.calls.append((node, self.scope)) + self._body(node) + + +def _lookup(scope: _Scope, name: str) -> list[ast.expr | None] | None: + current: _Scope | None = scope + while current is not None: + if name in current.bindings: + return current.bindings[name] + current = current.parent + return None + + +def _is_const(node: ast.expr, scope: _Scope, seen: frozenset[str] = frozenset()) -> bool: + """True when every value `node` can take is a string built only from literals -- including names + whose every assignment in scope is such a constant.""" if isinstance(node, ast.Constant): return isinstance(node.value, str) - return isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add) and _is_literal(node.left) and _is_literal(node.right) + if isinstance(node, ast.BinOp): + return isinstance(node.op, ast.Add) and _is_const(node.left, scope, seen) and _is_const(node.right, scope, seen) + if isinstance(node, ast.JoinedStr): + return all(_is_const(v.value if isinstance(v, ast.FormattedValue) else v, scope, seen) for v in node.values) + if isinstance(node, ast.IfExp): + return _is_const(node.body, scope, seen) and _is_const(node.orelse, scope, seen) + if isinstance(node, ast.Name): + if node.id in seen: + return False + bound = _lookup(scope, node.id) + return bool(bound) and all(b is not None and _is_const(b, scope, seen | {node.id}) for b in bound) + return False -def _is_arg_list(node: ast.expr | None) -> bool: +def _is_arg_list(node: ast.expr | None, scope: _Scope) -> bool: """A list/tuple command whose program is a literal: extra (even dynamic) items are arguments, not shell code.""" - return isinstance(node, (ast.List, ast.Tuple)) and bool(node.elts) and _is_literal(node.elts[0]) + return isinstance(node, (ast.List, ast.Tuple)) and bool(node.elts) and _is_const(node.elts[0], scope) def _shell_true(call: ast.Call) -> bool: @@ -63,11 +243,11 @@ def _finding(path: Path, node: ast.Call, rule: str, message: str, severity: str) return Finding(path, node.lineno, rule, message, severity=severity) -def _check_call(path: Path, node: ast.Call, aliases: dict[str, str]) -> list[Finding]: +def _check_call(path: Path, node: ast.Call, aliases: dict[str, str], scope: _Scope) -> list[Finding]: name = _dotted(node.func, aliases) last = name.rsplit(".", 1)[-1] first_arg = node.args[0] if node.args else next((kw.value for kw in node.keywords if kw.arg in _FIRST_ARG_KEYWORDS), None) - dynamic = first_arg is not None and not _is_literal(first_arg) + dynamic = first_arg is not None and not _is_const(first_arg, scope) if name in ("eval", "exec") and dynamic: return [ @@ -89,8 +269,8 @@ def _check_call(path: Path, node: ast.Call, aliases: dict[str, str]) -> list[Fin "error", ) ] - if name.startswith("subprocess.") and _shell_true(node) and not _is_arg_list(first_arg): - if dynamic or isinstance(first_arg, (ast.JoinedStr, ast.BinOp)): + if name.startswith("subprocess.") and _shell_true(node) and not _is_arg_list(first_arg, scope): + if dynamic: return [ _finding( path, @@ -188,7 +368,6 @@ def check_python_sinks(path: Path, source: str | None = None) -> list[Finding]: return [] aliases = _import_aliases(tree) findings: list[Finding] = [] - for node in ast.walk(tree): - if isinstance(node, ast.Call): - findings.extend(_check_call(path, node, aliases)) + for node, scope in _ScopeBuilder(tree).calls: + findings.extend(_check_call(path, node, aliases, scope)) return findings diff --git a/tests/test_find_injection.py b/tests/test_find_injection.py index b608fb3..b77c3fe 100644 --- a/tests/test_find_injection.py +++ b/tests/test_find_injection.py @@ -330,3 +330,65 @@ def test_csp_weakening_in_test_files_is_ignored(tmp_path: Path) -> None: (tmp_path / "web.config").write_text("Content-Security-Policy: default-src 'self'") (tmp_path / "test_headers.py").write_text("H = \"Content-Security-Policy: script-src 'unsafe-eval'\"\n") assert find_injection.run([], root=tmp_path).findings == [] + + +def _lines(tmp_path: Path, src: str) -> list[int]: + path = tmp_path / "m.py" + path.write_text(src) + return [f.line for f in check_python_sinks(path)] + + +def test_names_assigned_only_constants_are_constant(tmp_path: Path) -> None: + src = """import os, subprocess +GREETING = "echo hi" +CMD = GREETING + " there" + + +def ok(flag): + os.system(CMD) + cmd = "ls" if flag else "pwd" + os.system(cmd) + for c in ("a", "b"): + os.system(c) + os.system(f"{GREETING} now") + prog = "bun" + subprocess.check_call([prog, "x"], shell=True) + + +def bad(arg, flag): + cmd = "ls" + cmd += arg + os.system(cmd) + other = "ls" + if flag: + other = arg + os.system(other) + os.system(arg) + for c in arg: + os.system(c) + os.system(undefined_name) +""" + assert _lines(tmp_path, src) == [20, 24, 25, 27, 28] + + +def test_global_rebinding_and_class_scope_defeat_constness(tmp_path: Path) -> None: + src = """import os +GREETING = "echo hi" + + +def rebind(): + global GREETING + GREETING = input() + + +def use(): + os.system(GREETING) + + +class K: + x = "ls" + + def m(self): + os.system(x) +""" + assert _lines(tmp_path, src) == [11, 18] From f1b1f330e733c4d2947835e452539832e0afad3d Mon Sep 17 00:00:00 2001 From: Adrian Ehrsam Date: Tue, 29 Sep 2026 21:40:29 +0200 Subject: [PATCH 3/3] feat(lint): sql f-string error points to t-strings ({value} bound, {name:i} identifier) Co-Authored-By: Claude Code --- README.md | 3 ++- bmsdna/devtools/lint_sql.py | 18 +++++++++--------- 2 files changed, 11 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 407df4d..68d9584 100644 --- a/README.md +++ b/README.md @@ -499,7 +499,8 @@ it can't see through): - **`sql-fstring-injection`** / **`sql-concat-injection`** / **`sql-percent-format-injection`** / **`sql-format-injection`** — SQL built with an f-string, `+` concatenation, the `%` operator, or `str.format()` - instead of a psycopg t-string (3.14+), `psycopg.sql`, or bound params. + instead of a psycopg t-string (3.14+), `psycopg.sql`, or bound params. For an f-string the fix is + usually just `f"..."` -> `t"..."` (`{value}` is bound, `{name:i}` quotes an identifier). - **`sql-positional-param`** — positional `%s` instead of named `%(name)s`. - **`sql-forbidden-join`** — `RIGHT JOIN`/`LATERAL JOIN`/`CROSS APPLY` (same patterns the `prek` skill's `check_files.py` forbids in `.sql` files). diff --git a/bmsdna/devtools/lint_sql.py b/bmsdna/devtools/lint_sql.py index 412e9e3..986a4ed 100644 --- a/bmsdna/devtools/lint_sql.py +++ b/bmsdna/devtools/lint_sql.py @@ -266,18 +266,18 @@ def visit(operand: ast.expr) -> None: return "".join(parts) if saw_literal else None +_FSTRING_FIX = ( + "Change the `f` prefix to `t` (psycopg t-string, Python 3.14+): `{value}` becomes a bound parameter and " + "`{name:i}` quotes a table/column identifier. Or use load_sql()/a .sql file for static SQL." +) +_GENERIC_FIX = "Use a psycopg t-string (Python 3.14+), psycopg.sql for dynamic SQL, or load_sql()/a .sql file for static SQL." + + def _injection_finding(text: str, path: Path, lineno: int, rule: str, how: str) -> list[Finding]: if parse_sql_text(text) is None: return [] - return [ - Finding( - path, - lineno, - rule, - f"SQL built with {how} -- injection risk. Use a psycopg t-string (Python 3.14+), " - "psycopg.sql for dynamic SQL, or load_sql()/a .sql file for static SQL.", - ) - ] + fix = _FSTRING_FIX if rule == "sql-fstring-injection" else _GENERIC_FIX + return [Finding(path, lineno, rule, f"SQL built with {how} -- injection risk. {fix}")] def _literal_findings(text: str, path: Path, lineno: int) -> list[Finding]: