From cc8ed0fa19beb6bc4d8f8b57bcaeefe382894bdd Mon Sep 17 00:00:00 2001 From: Wolfvin Date: Wed, 1 Jul 2026 07:44:13 +0000 Subject: [PATCH] fix(deadcode): false positive on early return + multi-line dict return (closes #105) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause: when a multi-line (open brace > close brace on the start line) was encountered, the scanner skipped the line via WITHOUT resetting . If a previous had set the terminal flag, the multi-line return body (e.g. dict literals) was falsely flagged as unreachable code. Example false positive (the exact pattern from analyze.py before this fix): def _detect_vulns(workspace, max_items): ... if total == 0: return None return { # <- multiline start, was skipped "category": "vulns", # <- falsely flagged as unreachable "total": total, } Fix: in the multi-line return skip path (Python), check the current line's indent vs the previous terminal's indent. If the current return is in an outer scope (lower indent), the previous terminal is no longer relevant — reset found_terminal before continuing. Also reverts the antipattern introduced in PR #96 across 8 functions in scripts/commands/analyze.py back to the PEP 8-friendly form. The antipattern was only there to satisfy the buggy scanner; with the fix the PEP 8 form scans cleanly. Tests: 5 new regression tests in tests/test_deadcode_engine.py covering simple/chained/nested early returns + multi-line dict return + a sanity check that genuinely unreachable code is still detected. All 14 deadcode tests pass. --- scripts/commands/analyze.py | 172 ++++++++++++++++------------------ scripts/deadcode_engine.py | 16 ++++ tests/test_deadcode_engine.py | 110 ++++++++++++++++++++++ 3 files changed, 208 insertions(+), 90 deletions(-) diff --git a/scripts/commands/analyze.py b/scripts/commands/analyze.py index c47f2527..bfaf9607 100644 --- a/scripts/commands/analyze.py +++ b/scripts/commands/analyze.py @@ -454,17 +454,16 @@ def _detect_vulns(workspace: str, max_items: int) -> Optional[Dict]: total = vuln.get("stats", {}).get("total_vulnerabilities", 0) if total == 0: return None - else: - return { - "category": "vulnerabilities", - "label": "Known CVEs", - "total": total, - "severity": "critical", - "by_severity": vuln.get("stats", {}).get("by_severity", {}), - "top_items": vuln.get("vulnerabilities", [])[:max_items], - "action": "Update vulnerable dependencies immediately — check npm audit, pip audit, cargo audit, or govulncheck", - "impact": "Known vulnerabilities can be exploited by attackers even without source code access", - } + return { + "category": "vulnerabilities", + "label": "Known CVEs", + "total": total, + "severity": "critical", + "by_severity": vuln.get("stats", {}).get("by_severity", {}), + "top_items": vuln.get("vulnerabilities", [])[:max_items], + "action": "Update vulnerable dependencies immediately — check npm audit, pip audit, cargo audit, or govulncheck", + "impact": "Known vulnerabilities can be exploited by attackers even without source code access", + } def _detect_dataflow(workspace: str, max_items: int) -> Optional[Dict]: @@ -473,16 +472,15 @@ def _detect_dataflow(workspace: str, max_items: int) -> Optional[Dict]: violations = df.get("stats", {}).get("violations", 0) if violations == 0: return None - else: - return { - "category": "dataflow_violations", - "label": "Unsafe Data Flows", - "total": violations, - "severity": "high", - "top_items": df.get("violations", [])[:max_items], - "action": "Add input sanitization and output encoding at every source→sink boundary", - "impact": "Untainted data flows can lead to SQL injection, XSS, and command injection attacks", - } + return { + "category": "dataflow_violations", + "label": "Unsafe Data Flows", + "total": violations, + "severity": "high", + "top_items": df.get("violations", [])[:max_items], + "action": "Add input sanitization and output encoding at every source→sink boundary", + "impact": "Untainted data flows can lead to SQL injection, XSS, and command injection attacks", + } def _detect_env(workspace: str, max_items: int) -> Optional[Dict]: @@ -495,20 +493,19 @@ def _detect_env(workspace: str, max_items: int) -> Optional[Dict]: issues = undocumented # Each undocumented var is an issue if issues == 0 and total_vars == 0: return None - else: - return { - "category": "env_issues", - "label": "Environment Issues", - "total": issues, - "severity": "medium", - "top_items": [{"name": v.get("name"), "is_required": v.get("is_required"), - "has_fallback": v.get("has_fallback"), - "documentation": v.get("documentation")} - for v in env.get("variables", [])[:max_items] - if not v.get("documentation")], - "action": "Review .env files, ensure secrets are not committed, add .env to .gitignore", - "impact": "Misconfigured environment variables can leak secrets or cause runtime failures", - } + return { + "category": "env_issues", + "label": "Environment Issues", + "total": issues, + "severity": "medium", + "top_items": [{"name": v.get("name"), "is_required": v.get("is_required"), + "has_fallback": v.get("has_fallback"), + "documentation": v.get("documentation")} + for v in env.get("variables", [])[:max_items] + if not v.get("documentation")], + "action": "Review .env files, ensure secrets are not committed, add .env to .gitignore", + "impact": "Misconfigured environment variables can leak secrets or cause runtime failures", + } def _detect_smells(workspace: str, severity_filter: set, max_items: int) -> Optional[Dict]: @@ -561,17 +558,16 @@ def _detect_complexity(workspace: str, max_items: int) -> Optional[Dict]: hotspots = comp.get("hotspots", []) if not hotspots: return None - else: - return { - "category": "complexity", - "label": "Complexity Hotspots", - "total": len(hotspots), - "severity": "high" if any(h.get("cyclomatic", 0) > 20 for h in hotspots) else "medium", - "avg_cyclomatic": comp.get("stats", {}).get("avg_cyclomatic", 0), - "top_items": hotspots[:max_items], - "action": "Refactor high-complexity functions by extracting helper methods, reducing branches, and simplifying conditionals", - "impact": "Complex functions are bug magnets — they're hard to test, understand, and maintain", - } + return { + "category": "complexity", + "label": "Complexity Hotspots", + "total": len(hotspots), + "severity": "high" if any(h.get("cyclomatic", 0) > 20 for h in hotspots) else "medium", + "avg_cyclomatic": comp.get("stats", {}).get("avg_cyclomatic", 0), + "top_items": hotspots[:max_items], + "action": "Refactor high-complexity functions by extracting helper methods, reducing branches, and simplifying conditionals", + "impact": "Complex functions are bug magnets — they're hard to test, understand, and maintain", + } def _detect_dead_code(workspace: str, max_items: int) -> Optional[Dict]: @@ -580,17 +576,16 @@ def _detect_dead_code(workspace: str, max_items: int) -> Optional[Dict]: total = dc.get("stats", {}).get("total_dead_code", 0) if total == 0: return None - else: - return { - "category": "dead_code", - "label": "Dead Code", - "total": total, - "severity": "medium", - "by_category": dc.get("stats", {}).get("by_category", {}), - "top_items": dc.get("results", {}).get("unreachable", [])[:max_items], - "action": "Remove dead code in batches with testing — start with unreachable code and unused exports", - "impact": "Dead code increases maintenance burden, confuses new developers, and bloats the codebase", - } + return { + "category": "dead_code", + "label": "Dead Code", + "total": total, + "severity": "medium", + "by_category": dc.get("stats", {}).get("by_category", {}), + "top_items": dc.get("results", {}).get("unreachable", [])[:max_items], + "action": "Remove dead code in batches with testing — start with unreachable code and unused exports", + "impact": "Dead code increases maintenance burden, confuses new developers, and bloats the codebase", + } def _detect_circular(workspace: str, max_items: int) -> Optional[Dict]: @@ -623,17 +618,16 @@ def _detect_perf(workspace: str, max_items: int) -> Optional[Dict]: total = perf.get("stats", {}).get("total_hints", 0) if total == 0: return None - else: - return { - "category": "perf_hints", - "label": "Performance Issues", - "total": total, - "severity": perf.get("risk", "low"), - "by_category": perf.get("stats", {}).get("by_category", {}), - "top_items": perf.get("hints", [])[:max_items], - "action": "Address N+1 queries first (critical), then sync blocking, then memory leaks", - "impact": "Performance issues compound — N+1 queries scale linearly with data size, blocking calls freeze the event loop", - } + return { + "category": "perf_hints", + "label": "Performance Issues", + "total": total, + "severity": perf.get("risk", "low"), + "by_category": perf.get("stats", {}).get("by_category", {}), + "top_items": perf.get("hints", [])[:max_items], + "action": "Address N+1 queries first (critical), then sync blocking, then memory leaks", + "impact": "Performance issues compound — N+1 queries scale linearly with data size, blocking calls freeze the event loop", + } def _detect_config_drift(workspace: str, max_items: int) -> Optional[Dict]: @@ -642,16 +636,15 @@ def _detect_config_drift(workspace: str, max_items: int) -> Optional[Dict]: total = drift.get("stats", {}).get("total_drift_items", 0) if total == 0: return None - else: - return { - "category": "config_drift", - "label": "Dependency Drift", - "total": total, - "severity": "low", - "top_items": drift.get("drift_items", [])[:max_items], - "action": "Update outdated dependencies to reduce security risk and get bug fixes", - "impact": "Outdated dependencies may contain unpatched security vulnerabilities", - } + return { + "category": "config_drift", + "label": "Dependency Drift", + "total": total, + "severity": "low", + "top_items": drift.get("drift_items", [])[:max_items], + "action": "Update outdated dependencies to reduce security risk and get bug fixes", + "impact": "Outdated dependencies may contain unpatched security vulnerabilities", + } def _detect_binaries(workspace: str, max_items: int) -> Optional[Dict]: @@ -660,18 +653,17 @@ def _detect_binaries(workspace: str, max_items: int) -> Optional[Dict]: total = bins.get("stats", {}).get("total_artifacts", 0) if total == 0: return None - else: - return { - "category": "binary_artifacts", - "label": "Binary/Compiled Files", - "total": total, - "severity": "low", - "by_category": bins.get("stats", {}).get("by_category", {}), - "top_items": bins.get("findings", [])[:max_items], - "recommendations": bins.get("recommendations", []), - "action": "Add binary files to .gitignore and use build pipelines instead", - "impact": "Binary files bloat the repository, make diffs meaningless, and may contain vulnerable code", - } + return { + "category": "binary_artifacts", + "label": "Binary/Compiled Files", + "total": total, + "severity": "low", + "by_category": bins.get("stats", {}).get("by_category", {}), + "top_items": bins.get("findings", [])[:max_items], + "recommendations": bins.get("recommendations", []), + "action": "Add binary files to .gitignore and use build pipelines instead", + "impact": "Binary files bloat the repository, make diffs meaningless, and may contain vulnerable code", + } # ─── Helper Functions ────────────────────────────────────── diff --git a/scripts/deadcode_engine.py b/scripts/deadcode_engine.py index d133c107..5155c5a2 100755 --- a/scripts/deadcode_engine.py +++ b/scripts/deadcode_engine.py @@ -461,6 +461,22 @@ def _detect_unreachable_code(content: str, ext: str, rel_path: str) -> List[Dict open_count = stripped.count('(') + stripped.count('[') + stripped.count('{') close_count = stripped.count(')') + stripped.count(']') + stripped.count('}') if open_count > close_count: + # v10 (issue #105): Before skipping, check if we've already + # exited the block that contained the previous terminal + # statement. The classic false-positive pattern is: + # if x: + # return None # terminal at indent 8 + # return { # indent 4 — multiline start + # "k": "v", # indent 8 — was flagged as + # } # unreachable (same indent + # # as the terminal inside if) + # The previous terminal was inside an `if` block; the + # current return is in the outer scope (lower indent), + # so the previous terminal is no longer relevant. + # Reset it so the multi-line return body is not flagged. + current_indent = len(line) - len(line.lstrip()) + if found_terminal and terminal_indent > 0 and current_indent < terminal_indent: + found_terminal = False continue # Return expression continues on the next line found_terminal = True terminal_line = i # 0-based: next line has i+1 > i = True diff --git a/tests/test_deadcode_engine.py b/tests/test_deadcode_engine.py index bc6cd8fc..ba9779e5 100644 --- a/tests/test_deadcode_engine.py +++ b/tests/test_deadcode_engine.py @@ -180,3 +180,113 @@ def test_empty_workspace(self): assert result["stats"]["total_dead_code"] == 0 finally: shutil.rmtree(ws, ignore_errors=True) + + # ─── Issue #105 regression tests ──────────────────────────────── + # These patterns must NOT be flagged as unreachable. They are the + # PEP 8-friendly early-return pattern that workers were previously + # forced to wrap in `else:` to satisfy the scanner. + + def test_issue_105_early_return_then_final_return(self): + """Early return inside `if` + final return after should NOT be flagged.""" + code = """def f(condition): + if condition: + return None + return {"key": "value"} +""" + ws = self._create_workspace(code, "app.py") + try: + result = detect_dead_code(ws) + assert result["status"] == "ok" + unreachable = result.get("results", {}).get("unreachable", []) + assert len(unreachable) == 0, \ + f"False positive: {unreachable}" + finally: + shutil.rmtree(ws, ignore_errors=True) + + def test_issue_105_multiline_return_after_early_return(self): + """Multi-line dict return after early return should NOT be flagged. + + This is the exact reproduction of issue #105. Before the fix, the + scanner reported line 5 (the dict body) as unreachable because the + multi-line return detection skipped the `return {` line without + resetting the terminal flag from the previous `return None` inside + the `if` block. + """ + code = """def _detect_vulns(workspace, max_items): + from vulnscan_engine import scan_vulnerabilities + vuln = scan_vulnerabilities(workspace) + total = vuln.get("stats", {}).get("total_vulnerabilities", 0) + if total == 0: + return None + return { + "category": "vulnerabilities", + "total": total, + "top_items": vuln.get("vulnerabilities", [])[:max_items], + } +""" + ws = self._create_workspace(code, "app.py") + try: + result = detect_dead_code(ws) + assert result["status"] == "ok" + unreachable = result.get("results", {}).get("unreachable", []) + assert len(unreachable) == 0, \ + f"False positive on multi-line dict return: {unreachable}" + finally: + shutil.rmtree(ws, ignore_errors=True) + + def test_issue_105_chained_early_returns(self): + """Multiple chained early returns + final return should NOT be flagged.""" + code = """def g(x): + if x is None: + return None + if x < 0: + return -1 + if x > 100: + return 100 + return x +""" + ws = self._create_workspace(code, "app.py") + try: + result = detect_dead_code(ws) + assert result["status"] == "ok" + unreachable = result.get("results", {}).get("unreachable", []) + assert len(unreachable) == 0, \ + f"False positive on chained early returns: {unreachable}" + finally: + shutil.rmtree(ws, ignore_errors=True) + + def test_issue_105_nested_if_early_return(self): + """Nested if/return + outer returns should NOT be flagged.""" + code = """def m(x, y): + if x: + if y: + return None + return 1 + return 2 +""" + ws = self._create_workspace(code, "app.py") + try: + result = detect_dead_code(ws) + assert result["status"] == "ok" + unreachable = result.get("results", {}).get("unreachable", []) + assert len(unreachable) == 0, \ + f"False positive on nested if early return: {unreachable}" + finally: + shutil.rmtree(ws, ignore_errors=True) + + def test_issue_105_genuinely_unreachable_still_detected(self): + """Sanity check: genuinely unreachable code after unconditional + return must still be detected after the fix.""" + code = """def f(): + return None + print("unreachable") +""" + ws = self._create_workspace(code, "app.py") + try: + result = detect_dead_code(ws) + assert result["status"] == "ok" + unreachable = result.get("results", {}).get("unreachable", []) + assert len(unreachable) >= 1, \ + f"Regression: genuinely unreachable code not detected: {unreachable}" + finally: + shutil.rmtree(ws, ignore_errors=True)