cascade/diagnostics/bug_detector.py

"""
CASCADE Bug Detector - Automatic bug detection using pattern matching.

Uses cascade-lattice's forensic capabilities:
- GhostLog for inferring missing execution (what *should* have run)
- Artifact patterns for detecting anomalies
- SymbioticAdapter for interpreting signals
"""

import time
import hashlib
from typing import Any, Dict, List, Optional, Set, Callable, Tuple
from dataclasses import dataclass, field
from pathlib import Path
import ast
import re

from cascade.core.adapter import SymbioticAdapter
from cascade.forensics.artifacts import ArtifactDetector, TimestampArtifacts


@dataclass
class BugPattern:
    """A detectable bug pattern."""
    name: str
    description: str
    severity: str  # "critical", "error", "warning", "info"
    detector: Callable[[str, ast.AST], List[Dict[str, Any]]]
    category: str = "general"
    
    
@dataclass 
class DetectedIssue:
    """A detected code issue."""
    issue_id: str
    pattern_name: str
    severity: str
    file_path: str
    line_number: int
    column: int
    code_snippet: str
    message: str
    suggestion: Optional[str] = None
    confidence: float = 1.0
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "id": self.issue_id,
            "pattern": self.pattern_name,
            "severity": self.severity,
            "location": {
                "file": self.file_path,
                "line": self.line_number,
                "column": self.column,
            },
            "snippet": self.code_snippet,
            "message": self.message,
            "suggestion": self.suggestion,
            "confidence": self.confidence,
        }


class BugDetector:
    """
    Static analysis bug detector using AST patterns.
    
    Usage:
        detector = BugDetector()
        issues = detector.scan_file("path/to/file.py")
        issues = detector.scan_directory("path/to/project")
    """
    
    def __init__(self):
        self.patterns: List[BugPattern] = []
        self._detected_issues: List[DetectedIssue] = []
        self._scanned_files: Set[str] = set()
        
        # For signal interpretation
        self.adapter = SymbioticAdapter()
        
        # Register built-in patterns
        self._register_builtin_patterns()
    
    def _register_builtin_patterns(self):
        """Register built-in bug detection patterns."""
        patterns = [
            # Null checks
            BugPattern(
                name="potential_none_access",
                description="Accessing attribute on potentially None value",
                severity="warning",
                detector=self._detect_none_access,
                category="null_safety",
            ),
            
            # Exception handling
            BugPattern(
                name="bare_except",
                description="Bare except clause catches all exceptions",
                severity="warning",
                detector=self._detect_bare_except,
                category="exception_handling",
            ),
            BugPattern(
                name="empty_except",
                description="Empty except block silently swallows exceptions",
                severity="error",
                detector=self._detect_empty_except,
                category="exception_handling",
            ),
            
            # Resource management
            BugPattern(
                name="unclosed_resource",
                description="File/resource opened but not closed",
                severity="warning",
                detector=self._detect_unclosed_resource,
                category="resource_management",
            ),
            
            # Common mistakes
            BugPattern(
                name="mutable_default_arg",
                description="Mutable default argument in function",
                severity="warning",
                detector=self._detect_mutable_default,
                category="common_mistakes",
            ),
            BugPattern(
                name="comparison_to_none",
                description="Using == instead of 'is' for None comparison",
                severity="info",
                detector=self._detect_none_comparison,
                category="common_mistakes",
            ),
            BugPattern(
                name="unreachable_code",
                description="Code that can never be executed",
                severity="warning",
                detector=self._detect_unreachable_code,
                category="common_mistakes",
            ),
            
            # Security
            BugPattern(
                name="hardcoded_secret",
                description="Potential hardcoded secret or password",
                severity="error",
                detector=self._detect_hardcoded_secret,
                category="security",
            ),
            BugPattern(
                name="sql_injection_risk",
                description="Potential SQL injection vulnerability",
                severity="critical",
                detector=self._detect_sql_injection,
                category="security",
            ),
            
            # Performance
            BugPattern(
                name="loop_invariant",
                description="Computation inside loop that could be moved outside",
                severity="info",
                detector=self._detect_loop_invariant,
                category="performance",
            ),
        ]
        
        self.patterns.extend(patterns)
    
    def register_pattern(self, pattern: BugPattern):
        """Register a custom bug pattern."""
        self.patterns.append(pattern)
    
    def scan_file(self, file_path: str) -> List[DetectedIssue]:
        """Scan a Python file for bugs."""
        issues = []
        
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                source = f.read()
        except Exception as e:
            return [DetectedIssue(
                issue_id=self._generate_id(file_path, 0, "read_error"),
                pattern_name="file_read_error",
                severity="error",
                file_path=file_path,
                line_number=0,
                column=0,
                code_snippet="",
                message=f"Could not read file: {e}",
            )]
        
        try:
            tree = ast.parse(source)
        except SyntaxError as e:
            return [DetectedIssue(
                issue_id=self._generate_id(file_path, e.lineno or 0, "syntax_error"),
                pattern_name="syntax_error",
                severity="critical",
                file_path=file_path,
                line_number=e.lineno or 0,
                column=e.offset or 0,
                code_snippet=e.text or "",
                message=f"Syntax error: {e.msg}",
            )]
        
        # Run all patterns
        lines = source.splitlines()
        for pattern in self.patterns:
            try:
                matches = pattern.detector(source, tree)
                for match in matches:
                    line_num = match.get("line", 0)
                    snippet = lines[line_num - 1] if 0 < line_num <= len(lines) else ""
                    
                    issues.append(DetectedIssue(
                        issue_id=self._generate_id(file_path, line_num, pattern.name),
                        pattern_name=pattern.name,
                        severity=pattern.severity,
                        file_path=file_path,
                        line_number=line_num,
                        column=match.get("column", 0),
                        code_snippet=snippet.strip(),
                        message=match.get("message", pattern.description),
                        suggestion=match.get("suggestion"),
                        confidence=match.get("confidence", 1.0),
                    ))
            except Exception as e:
                print(f"[DIAG] Pattern {pattern.name} failed on {file_path}: {e}")
        
        self._scanned_files.add(file_path)
        self._detected_issues.extend(issues)
        
        return issues
    
    def scan_directory(self, dir_path: str, recursive: bool = True) -> List[DetectedIssue]:
        """Scan a directory for Python files and detect bugs."""
        path = Path(dir_path)
        issues = []
        
        pattern = "**/*.py" if recursive else "*.py"
        for py_file in path.glob(pattern):
            # Skip __pycache__
            if "__pycache__" in str(py_file):
                continue
            issues.extend(self.scan_file(str(py_file)))
        
        return issues
    
    def _generate_id(self, file_path: str, line: int, pattern: str) -> str:
        """Generate a unique issue ID."""
        content = f"{file_path}:{line}:{pattern}"
        return hashlib.sha256(content.encode()).hexdigest()[:16]
    
    # =========================================================================
    # PATTERN DETECTORS
    # =========================================================================
    
    def _detect_none_access(self, source: str, tree: ast.AST) -> List[Dict]:
        """Detect potential None access."""
        matches = []
        
        class Visitor(ast.NodeVisitor):
            def visit_Attribute(self, node):
                # Look for patterns like: x.y where x might be None
                # This is heuristic - check if there's no None check before
                if isinstance(node.value, ast.Name):
                    # Simple heuristic: flag if variable name suggests nullable
                    name = node.value.id
                    if any(word in name.lower() for word in ["result", "maybe", "optional", "response"]):
                        matches.append({
                            "line": node.lineno,
                            "column": node.col_offset,
                            "message": f"'{name}' may be None - consider adding a null check",
                            "confidence": 0.6,
                        })
                self.generic_visit(node)
        
        Visitor().visit(tree)
        return matches
    
    def _detect_bare_except(self, source: str, tree: ast.AST) -> List[Dict]:
        """Detect bare except clauses."""
        matches = []
        
        class Visitor(ast.NodeVisitor):
            def visit_ExceptHandler(self, node):
                if node.type is None:
                    matches.append({
                        "line": node.lineno,
                        "column": node.col_offset,
                        "message": "Bare 'except:' catches all exceptions including KeyboardInterrupt",
                        "suggestion": "Use 'except Exception:' instead",
                    })
                self.generic_visit(node)
        
        Visitor().visit(tree)
        return matches
    
    def _detect_empty_except(self, source: str, tree: ast.AST) -> List[Dict]:
        """Detect empty except blocks."""
        matches = []
        lines = source.splitlines()
        
        class Visitor(ast.NodeVisitor):
            def visit_ExceptHandler(self, node):
                # Check if body is just 'pass' or empty
                if len(node.body) == 1:
                    stmt = node.body[0]
                    if isinstance(stmt, ast.Pass):
                        # Check if there's a comment explaining the pass
                        line_idx = stmt.lineno - 1
                        if line_idx < len(lines):
                            line = lines[line_idx]
                            if '#' in line:
                                # Has a comment - don't flag as issue
                                self.generic_visit(node)
                                return
                        matches.append({
                            "line": node.lineno,
                            "column": node.col_offset,
                            "message": "Empty except block silently ignores exception",
                            "suggestion": "At minimum, log the exception",
                        })
                self.generic_visit(node)
        
        Visitor().visit(tree)
        return matches
    
    def _detect_unclosed_resource(self, source: str, tree: ast.AST) -> List[Dict]:
        """Detect files opened without context manager."""
        matches = []
        
        class Visitor(ast.NodeVisitor):
            def __init__(self):
                self.in_with = False
            
            def visit_With(self, node):
                old = self.in_with
                self.in_with = True
                self.generic_visit(node)
                self.in_with = old
            
            def visit_Call(self, node):
                if isinstance(node.func, ast.Name) and node.func.id == 'open':
                    if not self.in_with:
                        matches.append({
                            "line": node.lineno,
                            "column": node.col_offset,
                            "message": "File opened without 'with' context manager",
                            "suggestion": "Use 'with open(...) as f:' to ensure file is closed",
                        })
                self.generic_visit(node)
        
        Visitor().visit(tree)
        return matches
    
    def _detect_mutable_default(self, source: str, tree: ast.AST) -> List[Dict]:
        """Detect mutable default arguments."""
        matches = []
        
        class Visitor(ast.NodeVisitor):
            def visit_FunctionDef(self, node):
                for default in node.args.defaults + node.args.kw_defaults:
                    if default and isinstance(default, (ast.List, ast.Dict, ast.Set)):
                        matches.append({
                            "line": node.lineno,
                            "column": node.col_offset,
                            "message": f"Mutable default argument in function '{node.name}'",
                            "suggestion": "Use None as default and create mutable object inside function",
                        })
                        break
                self.generic_visit(node)
            
            def visit_AsyncFunctionDef(self, node):
                self.visit_FunctionDef(node)
        
        Visitor().visit(tree)
        return matches
    
    def _detect_none_comparison(self, source: str, tree: ast.AST) -> List[Dict]:
        """Detect == None instead of 'is None'."""
        matches = []
        
        class Visitor(ast.NodeVisitor):
            def visit_Compare(self, node):
                for i, (op, comparator) in enumerate(zip(node.ops, node.comparators)):
                    if isinstance(op, (ast.Eq, ast.NotEq)):
                        if isinstance(comparator, ast.Constant) and comparator.value is None:
                            matches.append({
                                "line": node.lineno,
                                "column": node.col_offset,
                                "message": "Use 'is None' instead of '== None'",
                                "suggestion": "Replace with 'is None' or 'is not None'",
                            })
                self.generic_visit(node)
        
        Visitor().visit(tree)
        return matches
    
    def _detect_unreachable_code(self, source: str, tree: ast.AST) -> List[Dict]:
        """Detect code after return/raise/break/continue."""
        matches = []
        
        class Visitor(ast.NodeVisitor):
            def check_body(self, body):
                for i, stmt in enumerate(body):
                    if isinstance(stmt, (ast.Return, ast.Raise, ast.Break, ast.Continue)):
                        # Check if there's code after this
                        if i + 1 < len(body):
                            next_stmt = body[i + 1]
                            matches.append({
                                "line": next_stmt.lineno,
                                "column": next_stmt.col_offset,
                                "message": "Unreachable code after return/raise/break/continue",
                            })
                    self.visit(stmt)
            
            def visit_FunctionDef(self, node):
                self.check_body(node.body)
            
            def visit_AsyncFunctionDef(self, node):
                self.check_body(node.body)
            
            def visit_If(self, node):
                self.check_body(node.body)
                self.check_body(node.orelse)
            
            def visit_For(self, node):
                self.check_body(node.body)
            
            def visit_While(self, node):
                self.check_body(node.body)
        
        Visitor().visit(tree)
        return matches
    
    def _detect_hardcoded_secret(self, source: str, tree: ast.AST) -> List[Dict]:
        """Detect potential hardcoded secrets."""
        matches = []
        
        # Pattern for secret-like variable names
        secret_patterns = re.compile(
            r'\b(password|passwd|pwd|secret|api_key|apikey|token|auth|credential)\s*=\s*["\'][^"\']+["\']',
            re.IGNORECASE
        )
        
        for i, line in enumerate(source.splitlines(), 1):
            if secret_patterns.search(line):
                matches.append({
                    "line": i,
                    "column": 0,
                    "message": "Potential hardcoded secret detected",
                    "suggestion": "Use environment variables or a secrets manager",
                })
        
        return matches
    
    def _detect_sql_injection(self, source: str, tree: ast.AST) -> List[Dict]:
        """Detect potential SQL injection vulnerabilities."""
        matches = []
        
        class Visitor(ast.NodeVisitor):
            def visit_Call(self, node):
                # Look for .execute() calls with string formatting
                if isinstance(node.func, ast.Attribute) and node.func.attr == 'execute':
                    for arg in node.args:
                        if isinstance(arg, ast.BinOp) and isinstance(arg.op, ast.Mod):
                            matches.append({
                                "line": node.lineno,
                                "column": node.col_offset,
                                "message": "Potential SQL injection - using string formatting in query",
                                "suggestion": "Use parameterized queries instead",
                            })
                        elif isinstance(arg, ast.JoinedStr):  # f-string
                            matches.append({
                                "line": node.lineno,
                                "column": node.col_offset,
                                "message": "Potential SQL injection - using f-string in query",
                                "suggestion": "Use parameterized queries instead",
                            })
                self.generic_visit(node)
        
        Visitor().visit(tree)
        return matches
    
    def _detect_loop_invariant(self, source: str, tree: ast.AST) -> List[Dict]:
        """Detect computations that could be moved outside loops."""
        matches = []
        
        class Visitor(ast.NodeVisitor):
            def visit_For(self, node):
                # Get loop variable name
                if isinstance(node.target, ast.Name):
                    loop_var = node.target.id
                    
                    # Look for calls that don't use the loop variable
                    for stmt in node.body:
                        if isinstance(stmt, ast.Assign):
                            for target in stmt.targets:
                                if isinstance(target, ast.Name):
                                    # Check if value doesn't depend on loop var
                                    value_vars = self._get_names(stmt.value)
                                    if loop_var not in value_vars and self._is_expensive(stmt.value):
                                        matches.append({
                                            "line": stmt.lineno,
                                            "column": stmt.col_offset,
                                            "message": "Computation inside loop may be loop-invariant",
                                            "suggestion": "Consider moving this outside the loop",
                                            "confidence": 0.5,
                                        })
                self.generic_visit(node)
            
            def _get_names(self, node) -> Set[str]:
                names = set()
                for child in ast.walk(node):
                    if isinstance(child, ast.Name):
                        names.add(child.id)
                return names
            
            def _is_expensive(self, node) -> bool:
                # Heuristic: calls are potentially expensive
                for child in ast.walk(node):
                    if isinstance(child, ast.Call):
                        return True
                return False
        
        Visitor().visit(tree)
        return matches
    
    # =========================================================================
    # REPORTING
    # =========================================================================
    
    def get_summary(self) -> Dict[str, Any]:
        """Get detection summary."""
        by_severity = {}
        by_category = {}
        
        for issue in self._detected_issues:
            by_severity[issue.severity] = by_severity.get(issue.severity, 0) + 1
            
            pattern = next((p for p in self.patterns if p.name == issue.pattern_name), None)
            if pattern:
                by_category[pattern.category] = by_category.get(pattern.category, 0) + 1
        
        return {
            "files_scanned": len(self._scanned_files),
            "total_issues": len(self._detected_issues),
            "by_severity": by_severity,
            "by_category": by_category,
        }
    
    def get_report(self) -> str:
        """Generate a human-readable report."""
        lines = [
            "BUG DETECTION REPORT",
            "=" * 60,
            f"Files scanned: {len(self._scanned_files)}",
            f"Issues found: {len(self._detected_issues)}",
            "",
        ]
        
        # Group by severity
        by_severity: Dict[str, List[DetectedIssue]] = {}
        for issue in self._detected_issues:
            if issue.severity not in by_severity:
                by_severity[issue.severity] = []
            by_severity[issue.severity].append(issue)
        
        severity_order = ["critical", "error", "warning", "info"]
        severity_icons = {"critical": "🔴", "error": "❌", "warning": "⚠️", "info": "ℹ️"}
        
        for severity in severity_order:
            if severity in by_severity:
                issues = by_severity[severity]
                icon = severity_icons.get(severity, "•")
                lines.append(f"\n{icon} {severity.upper()} ({len(issues)})")
                lines.append("-" * 40)
                
                for issue in issues:
                    lines.append(f"  {issue.file_path}:{issue.line_number}")
                    lines.append(f"    {issue.message}")
                    if issue.suggestion:
                        lines.append(f"    💡 {issue.suggestion}")
                    lines.append("")
        
        return "\n".join(lines)