diff --git a/scripts/ast_taint_engine.py b/scripts/ast_taint_engine.py index 7d689883..85dc9d5a 100644 --- a/scripts/ast_taint_engine.py +++ b/scripts/ast_taint_engine.py @@ -2710,14 +2710,14 @@ def _analyze_function_body(self, func_def: FunctionDef, def _extract_return_taint(self, func_def: FunctionDef, final_state: TaintState, language: str) -> Optional[TaintState]: - """Inspect `return` statements in a function body for tainted data. + """Inspect return statements in a function body for tainted data. - Walks the function body AST looking for `return` nodes. For each + Walks the function body AST looking for return nodes. For each return, resolves the returned expression to a variable name and checks if that variable is tainted in *final_state*. - Returns a TaintState containing the taint info of any tainted - return expressions, or None if no return carries taint. + Yields a TaintState containing the taint info of any tainted + return-expressions, or None if no return carries taint. """ if not func_def.body_node or not self._detector: return None diff --git a/scripts/autofix_engine.py b/scripts/autofix_engine.py index d790c3fb..4438fe64 100644 --- a/scripts/autofix_engine.py +++ b/scripts/autofix_engine.py @@ -197,29 +197,14 @@ def generate_fix(self, finding: Dict, file_content: str, file_lines: List[str]) # Check it's a simple console.log (not in a catch block) if "catch" in file_lines[max(0, line_idx - 3):line_idx]: return None # Don't remove console.error in catch blocks - return { - "replacement_lines": None, # Delete the line - "confidence": 0.95, - "risk": RISK_SAFE, - "description": f"Remove debug statement: {stripped[:60]}", - } + return {"replacement_lines": None, "confidence": 0.95, "risk": RISK_SAFE, "description": f"Remove debug statement: {stripped[:60]}"} elif categories == "print_statement": # Don't remove print in __main__ blocks if '__main__' in ''.join(file_lines[max(0, line_idx - 10):line_idx]): return None - return { - "replacement_lines": None, - "confidence": 0.90, - "risk": RISK_SAFE, - "description": f"Remove print statement: {stripped[:60]}", - } + return {"replacement_lines": None, "confidence": 0.90, "risk": RISK_SAFE, "description": f"Remove print statement: {stripped[:60]}"} elif categories in ("debugger", "pdb", "breakpoint"): - return { - "replacement_lines": None, - "confidence": 0.98, - "risk": RISK_SAFE, - "description": f"Remove debugger statement: {stripped[:60]}", - } + return {"replacement_lines": None, "confidence": 0.98, "risk": RISK_SAFE, "description": f"Remove debugger statement: {stripped[:60]}"} return None @@ -302,12 +287,7 @@ def generate_fix(self, finding: Dict, file_content: str, file_lines: List[str]) if line_idx < 0 or line_idx >= len(file_lines): return None - return { - "replacement_lines": None, - "confidence": 0.85, - "risk": RISK_MODERATE, - "description": f"Remove unused import on line {line_idx + 1}", - } + return {"replacement_lines": None, "confidence": 0.85, "risk": RISK_MODERATE, "description": f"Remove unused import on line {line_idx + 1}"} # ─── Auto-Fix Engine ──────────────────────────────────────── diff --git a/scripts/callgraph_engine.py b/scripts/callgraph_engine.py index 50813f3b..06bdb123 100644 --- a/scripts/callgraph_engine.py +++ b/scripts/callgraph_engine.py @@ -176,10 +176,13 @@ def __hash__(self): return hash((self.caller, self.callee, self.file_path, self.line)) def __eq__(self, other): - if not isinstance(other, CallEdge): - return False - return (self.caller == other.caller and self.callee == other.callee - and self.file_path == other.file_path and self.line == other.line) + return ( + isinstance(other, CallEdge) + and self.caller == other.caller + and self.callee == other.callee + and self.file_path == other.file_path + and self.line == other.line + ) @dataclass diff --git a/scripts/circular_engine.py b/scripts/circular_engine.py index bc0e7aaf..195e930a 100755 --- a/scripts/circular_engine.py +++ b/scripts/circular_engine.py @@ -420,8 +420,10 @@ def _classify_cycle_source(chain_files: List[str]) -> str: # Check if all files are in core directories (src/, lib/) def _is_core_path(f: str) -> bool: norm = '/' + f if not f.startswith('/') else f - return any(d in norm for d in ('/src/', '/lib/')) or \ - any(f.startswith(d) for d in ('src/', 'lib/')) + return ( + any(d in norm for d in ('/src/', '/lib/')) + or any(f.startswith(d) for d in ('src/', 'lib/')) + ) all_core = all(_is_core_path(f) for f in chain_files) diff --git a/scripts/commands/analyze.py b/scripts/commands/analyze.py index bfaf9607..c47f2527 100644 --- a/scripts/commands/analyze.py +++ b/scripts/commands/analyze.py @@ -454,16 +454,17 @@ def _detect_vulns(workspace: str, max_items: int) -> Optional[Dict]: total = vuln.get("stats", {}).get("total_vulnerabilities", 0) if total == 0: return None - 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", - } + 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", + } def _detect_dataflow(workspace: str, max_items: int) -> Optional[Dict]: @@ -472,15 +473,16 @@ def _detect_dataflow(workspace: str, max_items: int) -> Optional[Dict]: violations = df.get("stats", {}).get("violations", 0) if violations == 0: return None - 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", - } + 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", + } def _detect_env(workspace: str, max_items: int) -> Optional[Dict]: @@ -493,19 +495,20 @@ 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 - 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", - } + 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", + } def _detect_smells(workspace: str, severity_filter: set, max_items: int) -> Optional[Dict]: @@ -558,16 +561,17 @@ def _detect_complexity(workspace: str, max_items: int) -> Optional[Dict]: hotspots = comp.get("hotspots", []) if not hotspots: return None - 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", - } + 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", + } def _detect_dead_code(workspace: str, max_items: int) -> Optional[Dict]: @@ -576,16 +580,17 @@ 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 - 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", - } + 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", + } def _detect_circular(workspace: str, max_items: int) -> Optional[Dict]: @@ -618,16 +623,17 @@ def _detect_perf(workspace: str, max_items: int) -> Optional[Dict]: total = perf.get("stats", {}).get("total_hints", 0) if total == 0: return None - 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", - } + 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", + } def _detect_config_drift(workspace: str, max_items: int) -> Optional[Dict]: @@ -636,15 +642,16 @@ 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 - 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", - } + 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", + } def _detect_binaries(workspace: str, max_items: int) -> Optional[Dict]: @@ -653,17 +660,18 @@ def _detect_binaries(workspace: str, max_items: int) -> Optional[Dict]: total = bins.get("stats", {}).get("total_artifacts", 0) if total == 0: return None - 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", - } + 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", + } # ─── Helper Functions ────────────────────────────────────── diff --git a/scripts/dashboard_engine.py b/scripts/dashboard_engine.py index 80993421..c5eb2f77 100644 --- a/scripts/dashboard_engine.py +++ b/scripts/dashboard_engine.py @@ -73,7 +73,10 @@ def generate_dashboard( with open(output_path, 'w', encoding='utf-8') as f: f.write(html) except IOError as e: - return {"status": "error", "error": f"Failed to write dashboard: {e}"} + return { + "status": "error", + "error": f"Failed to write dashboard: {e}", + } return { "status": "ok", @@ -1462,7 +1465,7 @@ def _render_comparison_section(comparison: Optional[Dict]) -> str: overall = summary.get("overall", "unchanged") overall_color = "#22c55e" if overall == "improved" else "#ef4444" if overall == "degraded" else "#64748b" - return f"""
+ html = f"""
Snapshot Comparison
Overall: {overall.upper()} @@ -1475,3 +1478,4 @@ def _render_comparison_section(comparison: Optional[Dict]) -> str:
{''.join(rows)}
""" + return html diff --git a/scripts/mcp_server.py b/scripts/mcp_server.py index 817c93ad..9bd18a92 100644 --- a/scripts/mcp_server.py +++ b/scripts/mcp_server.py @@ -1264,11 +1264,12 @@ def _handle_message(self, raw: str) -> Optional[Dict[str, Any]]: return self._dispatch_request(method, params, msg_id) # No method — invalid request - return { + error_response = { "jsonrpc": "2.0", "id": msg_id, - "error": {"code": -32600, "message": "Invalid Request: missing method"} + "error": {"code": -32600, "message": "Invalid Request: missing method"}, } + return error_response def _dispatch_notification(self, method: str, params: Any) -> None: """Handle JSON-RPC notifications (no response expected).""" diff --git a/scripts/plugin_system.py b/scripts/plugin_system.py index 67edd866..c04c2f3f 100644 --- a/scripts/plugin_system.py +++ b/scripts/plugin_system.py @@ -520,7 +520,7 @@ def _parse_single_rule(data: Dict, plugin_name: str, file_path: str) -> Optional logger.debug(f"Skipping rule without id in {file_path}") return None - return PluginRule( + rule = PluginRule( id=str(rule_id), name=str(data.get("name", rule_id)), severity=str(data.get("severity", "info")), @@ -536,6 +536,7 @@ def _parse_single_rule(data: Dict, plugin_name: str, file_path: str) -> Optional plugin_name=plugin_name, file_path=file_path, ) + return rule # ─── Plugin Manager ──────────────────────────────────────────