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)