diff --git a/CHANGELOG.md b/CHANGELOG.md index fc94b97..62a2932 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 only the identifier name appears in output, so a real secret never lands in CI logs or PR comments. The name match is singular-only (`tokens_in` token counts and `_FIX_TOKENS` word lists don't fire). +- **Lint rule `db-call-in-loop` + `loop_depth` on call records.** + Both parsers now thread loop depth through their call-collection DFS; + CALLS edge metadata gains `loop_depth` / `in_loop`. TS additionally + treats function arguments to `.map/.forEach/.filter/.reduce/.flatMap` + as implicit loop bodies. The new rule (high) flags DB-API calls + (`db.select|insert|update|delete|execute|query` in TS; + `session|cursor|db . query|execute|add|commit` in Python) executing + inside a loop — the classic N+1. +- **Risk-ranked untested + `new_untested_hotspot` review rule.** + `rank_untested()` joins untested × hotspot score × blast radius so + reports lead with untested *and* high-fan-in functions; surfaced in + `codegraph query untested` and the MCP `untested` tool. New default + review rule (high, threshold 5): a newly-added function with ≥5 + callers and no test-module caller now gates PRs. **Note:** this new + default rule can change `codegraph review` exit codes for existing + users; disable by shipping a custom `.codegraph/rules.yml`. +- **Lint rules `any-on-boundary` (low) and `any-into-db-write` (med).** + The TS parser records `exported`, `any_params`, and `any_return` + metadata; the rules flag exported functions with `any` in their + signature and `as any` casts passed into DB-write calls. Surface + syntax only — this is not type inference. ## [0.1.2] — 2026-05-31 diff --git a/codegraph/analysis/lint.py b/codegraph/analysis/lint.py index 8eccc2e..68c4b1c 100644 --- a/codegraph/analysis/lint.py +++ b/codegraph/analysis/lint.py @@ -24,7 +24,7 @@ from codegraph.parsers.base import load_parser, node_text from codegraph.parsers.python import _is_test_file as _is_py_test_file -from codegraph.parsers.typescript import EXT_TO_LANG +from codegraph.parsers.typescript import _ANY_WORD_RE, EXT_TO_LANG from codegraph.parsers.typescript import _is_test_file as _is_ts_test_file if TYPE_CHECKING: @@ -391,11 +391,155 @@ def _check_db_call_in_loop( ) +_DEFAULT_DB_WRITE_PATTERN = ( + r"\b(db|database)\.(insert|update|execute)\b|\.values\(|\.set\(" +) + +_MAX_ANCESTOR_LEVELS = 6 + + +def _has_export_ancestor(node: tree_sitter.Node) -> bool: + """Return True if any ancestor (up to _MAX_ANCESTOR_LEVELS) is an export_statement.""" + cur = node.parent + for _ in range(_MAX_ANCESTOR_LEVELS): + if cur is None: + return False + if cur.type == "export_statement": + return True + cur = cur.parent + return False + + +def _as_any_in_args(args_node: tree_sitter.Node, src: bytes) -> bool: + """Return True if any argument inside *args_node* is an ``as any`` cast.""" + for child in args_node.children: + if child.type == "as_expression": + right = child.child_by_field_name("type") + # tree-sitter represents the rhs of `as` as a 'predefined_type' child + # or via child_by_field_name("type"). Walk children to be safe. + for c in child.children: + if c.type == "predefined_type" and node_text(c, src) == "any": + return True + if right is not None and node_text(right, src) == "any": + return True + return False + + +def _check_any_on_boundary( + node: tree_sitter.Node, ctx: _LintContext +) -> LintFinding | None: + """Exported function / arrow with ``any`` in a param type or return type.""" + if ctx.language not in _TS_LANGS or ctx.is_test_file: + return None + + # Match function declarations directly or inside export_statement + if node.type == "function_declaration": + if not _has_export_ancestor(node): + return None + elif node.type in ("arrow_function", "function_expression", "function"): + # Must be inside an export_statement (via lexical_declaration) + if not _has_export_ancestor(node): + return None + else: + return None + + # Check params for any + params_node = node.child_by_field_name("parameters") + if params_node is None: + for c in node.children: + if c.type == "formal_parameters": + params_node = c + break + + any_param_names: list[str] = [] + if params_node is not None: + for param in params_node.children: + if param.type not in ("required_parameter", "optional_parameter"): + continue + ann = param.child_by_field_name("type") + if ann is not None: + type_text = node_text(ann, ctx.src) + if _ANY_WORD_RE.search(type_text): + # Grab param name + for c in param.children: + if c.type == "identifier": + any_param_names.append(node_text(c, ctx.src)) + break + + # Check return type annotation + has_any_return = False + for c in node.children: + if c.type == "type_annotation": + if _ANY_WORD_RE.search(node_text(c, ctx.src)): + has_any_return = True + break + + if not any_param_names and not has_any_return: + return None + + parts: list[str] = [] + if any_param_names: + parts.append(f"param(s) {', '.join(any_param_names)}") + if has_any_return: + parts.append("return type") + detail = " and ".join(parts) + + # Find a name for the function (best-effort) + name_node = node.child_by_field_name("name") + func_name = node_text(name_node, ctx.src) if name_node is not None else "" + + return LintFinding( + rule_id=ctx.rule.id, + severity=ctx.rule.severity, + message=_format_message(ctx.rule, name=func_name, detail=detail), + file=ctx.rel_path, + line=node.start_point[0] + 1, + snippet=_snippet(node, ctx.src), + ) + + +def _check_any_into_db_write( + node: tree_sitter.Node, ctx: _LintContext +) -> LintFinding | None: + """``as any`` cast passed as argument to a DB-write call.""" + if ctx.language not in _TS_LANGS or ctx.is_test_file: + return None + if node.type != "call_expression" or not _is_chain_root(node): + return None + + chain = _member_chain(node, ctx.src) + if chain is None: + return None + base, methods = chain + chain_text = ".".join([base, *methods]) + pattern = str( + ctx.rule.options.get("pattern") or _DEFAULT_DB_WRITE_PATTERN + ) + if not re.search(pattern, chain_text): + return None + + # Check whether any argument in the outermost call contains `as any` + args_node = node.child_by_field_name("arguments") + if args_node is None or not _as_any_in_args(args_node, ctx.src): + return None + + return LintFinding( + rule_id=ctx.rule.id, + severity=ctx.rule.severity, + message=_format_message(ctx.rule, chain=chain_text), + file=ctx.rel_path, + line=node.start_point[0] + 1, + snippet=_snippet(node, ctx.src), + ) + + _CHECK_REGISTRY: dict[str, _Checker] = { "console-in-prod": _check_console_in_prod, "unfiltered-query": _check_unfiltered_query, "sensitive-literal": _check_sensitive_literal, "db-call-in-loop": _check_db_call_in_loop, + "any-on-boundary": _check_any_on_boundary, + "any-into-db-write": _check_any_into_db_write, } @@ -424,6 +568,18 @@ def _check_db_call_in_loop( severity="high", message="DB call `{chain}` inside a loop (N+1 risk)", ), + LintRule( + id="any-on-boundary", + check="any-on-boundary", + severity="low", + message="Exported function `{name}` uses `any` in {detail}", + ), + LintRule( + id="any-into-db-write", + check="any-into-db-write", + severity="med", + message="DB-write call `{chain}` receives an `as any` cast argument", + ), ] diff --git a/codegraph/parsers/typescript.py b/codegraph/parsers/typescript.py index 7182bed..8ebc187 100644 --- a/codegraph/parsers/typescript.py +++ b/codegraph/parsers/typescript.py @@ -257,6 +257,20 @@ def _strip_type_annotation(text: str) -> str: return s.strip() +_ANY_WORD_RE = re.compile(r"\bany\b") + + +def _type_mentions_any(t: str | None) -> bool: + """Return True if *t* contains the bare type ``any``. + + Matches ``any``, ``any[]``, ``Promise``, ``Record`` + but NOT ``Company``, ``anything``, or similar sub-word occurrences. + """ + if t is None: + return False + return bool(_ANY_WORD_RE.search(t)) + + def _extract_param( p: tree_sitter.Node, src: bytes ) -> dict[str, str | None] | None: @@ -682,6 +696,7 @@ def _visit( self._handle_function_decl( sub, rel, parent_qualname, parent_id, lang, src, nodes, edges, + exported=True, ) elif sub.type in ( "lexical_declaration", "variable_declaration" @@ -689,6 +704,7 @@ def _visit( self._handle_lexical_decl( sub, rel, parent_qualname, parent_id, lang, src, nodes, edges, + exported=True, ) def _handle_import( @@ -899,6 +915,8 @@ def _handle_function_decl( src: bytes, nodes: list[Node], edges: list[Edge], + *, + exported: bool = False, ) -> None: name_node = node.child_by_field_name("name") if name_node is None: @@ -921,6 +939,17 @@ def _handle_function_decl( "params": params_list, "returns": return_type, } + if exported: + func_md["exported"] = True + any_params = [ + p["name"] + for p in params_list + if isinstance(p, dict) and _type_mentions_any(p.get("type")) + ] + if any_params: + func_md["any_params"] = any_params + if _type_mentions_any(return_type): + func_md["any_return"] = True if _has_public_api_pragma_ts(node, src): func_md["public_api"] = True func_node = Node( @@ -957,6 +986,8 @@ def _handle_lexical_decl( src: bytes, nodes: list[Node], edges: list[Edge], + *, + exported: bool = False, ) -> None: for child in node.children: if child.type != "variable_declarator": @@ -990,6 +1021,23 @@ def _handle_lexical_decl( value_node, arrow_params, src ) + arrow_md: dict[str, Any] = { + "arrow": True, + "params": params_list, + "returns": return_type, + } + if exported: + arrow_md["exported"] = True + any_params = [ + p["name"] + for p in params_list + if isinstance(p, dict) and _type_mentions_any(p.get("type")) + ] + if any_params: + arrow_md["any_params"] = any_params + if _type_mentions_any(return_type): + arrow_md["any_return"] = True + func_node = Node( id=func_id, kind=NodeKind.FUNCTION, @@ -999,11 +1047,7 @@ def _handle_lexical_decl( line_start=node.start_point[0] + 1, line_end=node.end_point[0] + 1, language=lang, - metadata={ - "arrow": True, - "params": params_list, - "returns": return_type, - }, + metadata=arrow_md, ) nodes.append(func_node) diff --git a/tests/fixtures/lint_sample/src/api.ts b/tests/fixtures/lint_sample/src/api.ts new file mode 100644 index 0000000..4dfddce --- /dev/null +++ b/tests/fixtures/lint_sample/src/api.ts @@ -0,0 +1,34 @@ +import { db, jobs } from "./schema"; + +// Case A — exported function with any param and any return: SHOULD fire any-on-boundary +export function processPayload(x: any): any { + return x; +} + +// Case A — exported arrow with any param: SHOULD fire any-on-boundary +export const handleInput = (data: any): string => { + return String(data); +}; + +// Typed exported counterexample — SHOULD NOT fire any-on-boundary +export function typedHandler(x: string): string { + return x.trim(); +} + +// Non-exported function with any — SHOULD NOT fire any-on-boundary (case A) +function internalProcess(x: any): any { + return x; +} + +// Case B — as any into db.insert().values(): SHOULD fire any-into-db-write +export async function submitJob(payload: unknown): Promise { + await db.insert(jobs).values(payload as any); +} + +// Typed call — SHOULD NOT fire any-into-db-write +export async function submitTypedJob(payload: { name: string }): Promise { + await db.insert(jobs).values(payload); +} + +// Keep internalProcess reachable so tree-shaking doesn't flag it +export { internalProcess }; diff --git a/tests/test_extractor_typescript.py b/tests/test_extractor_typescript.py index ac91caf..50bdefc 100644 --- a/tests/test_extractor_typescript.py +++ b/tests/test_extractor_typescript.py @@ -93,3 +93,61 @@ def test_parse_arrow_function(extractor: TypeScriptExtractor) -> None: nodes, _ = extractor.parse_file(FIXTURE_DIR / "utils.ts", FIXTURE_DIR) names = {n.name for n in nodes if n.kind == NodeKind.FUNCTION} assert "multiply" in names + + +# --- exported + any metadata ----------------------------------------------- + +LINT_SAMPLE_DIR = Path(__file__).parent / "fixtures" / "lint_sample" / "src" + + +def test_exported_flag_set_on_exported_function( + extractor: TypeScriptExtractor, +) -> None: + nodes, _ = extractor.parse_file(LINT_SAMPLE_DIR / "api.ts", LINT_SAMPLE_DIR) + funcs = {n.name: n for n in nodes if n.kind == NodeKind.FUNCTION} + assert funcs["processPayload"].metadata.get("exported") is True + assert funcs["typedHandler"].metadata.get("exported") is True + assert funcs["handleInput"].metadata.get("exported") is True + + +def test_exported_flag_absent_on_non_exported_function( + extractor: TypeScriptExtractor, +) -> None: + nodes, _ = extractor.parse_file(LINT_SAMPLE_DIR / "api.ts", LINT_SAMPLE_DIR) + funcs = {n.name: n for n in nodes if n.kind == NodeKind.FUNCTION} + assert not funcs["internalProcess"].metadata.get("exported") + + +def test_any_params_recorded_for_any_param_function( + extractor: TypeScriptExtractor, +) -> None: + nodes, _ = extractor.parse_file(LINT_SAMPLE_DIR / "api.ts", LINT_SAMPLE_DIR) + funcs = {n.name: n for n in nodes if n.kind == NodeKind.FUNCTION} + assert funcs["processPayload"].metadata.get("any_params") == ["x"] + assert funcs["handleInput"].metadata.get("any_params") == ["data"] + + +def test_any_params_absent_for_typed_function( + extractor: TypeScriptExtractor, +) -> None: + nodes, _ = extractor.parse_file(LINT_SAMPLE_DIR / "api.ts", LINT_SAMPLE_DIR) + funcs = {n.name: n for n in nodes if n.kind == NodeKind.FUNCTION} + assert not funcs["typedHandler"].metadata.get("any_params") + + +def test_any_return_set_when_return_type_is_any( + extractor: TypeScriptExtractor, +) -> None: + nodes, _ = extractor.parse_file(LINT_SAMPLE_DIR / "api.ts", LINT_SAMPLE_DIR) + funcs = {n.name: n for n in nodes if n.kind == NodeKind.FUNCTION} + assert funcs["processPayload"].metadata.get("any_return") is True + + +def test_any_return_absent_for_typed_return( + extractor: TypeScriptExtractor, +) -> None: + nodes, _ = extractor.parse_file(LINT_SAMPLE_DIR / "api.ts", LINT_SAMPLE_DIR) + funcs = {n.name: n for n in nodes if n.kind == NodeKind.FUNCTION} + assert not funcs["typedHandler"].metadata.get("any_return") + # arrow with any param but typed return should NOT set any_return + assert not funcs["handleInput"].metadata.get("any_return") diff --git a/tests/test_lint.py b/tests/test_lint.py index af3f8fd..d3a29ed 100644 --- a/tests/test_lint.py +++ b/tests/test_lint.py @@ -258,3 +258,110 @@ def test_db_call_in_loop_message_contains_chain(tmp_path: Path) -> None: assert findings for f in findings: assert "db." in f.message or "session." in f.message + + +# --- any-on-boundary ------------------------------------------------------- + + +def test_any_on_boundary_fires_for_exported_any_param(tmp_path: Path) -> None: + repo = _sample_repo(tmp_path) + findings = [ + f for f in run_lint(repo) + if f.rule_id == "any-on-boundary" and f.file == "src/api.ts" + ] + names_in_msgs = " ".join(f.message for f in findings) + assert "processPayload" in names_in_msgs + + +def test_any_on_boundary_fires_for_exported_any_return(tmp_path: Path) -> None: + repo = _sample_repo(tmp_path) + findings = [ + f for f in run_lint(repo) + if f.rule_id == "any-on-boundary" and f.file == "src/api.ts" + ] + # processPayload has both any param and any return + process_findings = [f for f in findings if "processPayload" in f.message] + assert process_findings + assert "return type" in process_findings[0].message + + +def test_any_on_boundary_fires_for_exported_arrow_any_param( + tmp_path: Path, +) -> None: + repo = _sample_repo(tmp_path) + findings = [ + f for f in run_lint(repo) + if f.rule_id == "any-on-boundary" and f.file == "src/api.ts" + and f.line == 9 + ] + assert findings, "expected finding on line 9 (exported arrow with any param)" + + +def test_any_on_boundary_does_not_fire_for_non_exported( + tmp_path: Path, +) -> None: + repo = _sample_repo(tmp_path) + findings = [ + f for f in run_lint(repo) + if f.rule_id == "any-on-boundary" and f.file == "src/api.ts" + ] + # internalProcess is not exported; must not appear + assert not any("internalProcess" in f.message for f in findings) + + +def test_any_on_boundary_does_not_fire_for_typed_export( + tmp_path: Path, +) -> None: + repo = _sample_repo(tmp_path) + findings = [ + f for f in run_lint(repo) + if f.rule_id == "any-on-boundary" and f.file == "src/api.ts" + ] + assert not any("typedHandler" in f.message for f in findings) + + +def test_any_on_boundary_severity_low(tmp_path: Path) -> None: + repo = _sample_repo(tmp_path) + findings = [ + f for f in run_lint(repo) if f.rule_id == "any-on-boundary" + ] + assert findings + assert all(f.severity == "low" for f in findings) + + +# --- any-into-db-write ------------------------------------------------------- + + +def test_any_into_db_write_fires_for_as_any_in_values( + tmp_path: Path, +) -> None: + repo = _sample_repo(tmp_path) + findings = [ + f for f in run_lint(repo) + if f.rule_id == "any-into-db-write" and f.file == "src/api.ts" + ] + assert findings + assert findings[0].line == 25 + + +def test_any_into_db_write_severity_med(tmp_path: Path) -> None: + repo = _sample_repo(tmp_path) + findings = [ + f for f in run_lint(repo) if f.rule_id == "any-into-db-write" + ] + assert findings + assert all(f.severity == "med" for f in findings) + + +def test_any_into_db_write_does_not_fire_for_typed_arg( + tmp_path: Path, +) -> None: + repo = _sample_repo(tmp_path) + findings = [ + f for f in run_lint(repo) + if f.rule_id == "any-into-db-write" and f.file == "src/api.ts" + ] + # submitTypedJob passes a typed payload — should not fire + # Only submitJob (line 25) should fire; line 30 should not + assert len(findings) == 1 + assert findings[0].line == 25