""" Security scanning tools for Clover - A terminal assistant for AI-powered project management """ import json import os import re import subprocess import sys from pathlib import Path from typing import Any, Dict, List, Optional, Tuple # Add the current directory to Python path for imports sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from config.settings import load_config from models.api_client import APIClient from tools.file_tools import read_file class SecurityScanner: """Handle security scanning operations""" def __init__(self): self.config = load_config() self.api_client = APIClient() self.security_tools = { "python": { "bandit": ["python", "-m", "bandit", "-r", "-f", "json"], "safety": ["safety", "check", "--json"], "semgrep": ["semgrep", "--config=auto", "--json"], }, "javascript": { "npm_audit": ["npm", "audit", "--json"], "eslint_security": [ "eslint", "--format=json", "-c", ".eslintrc.security.js", ], "semgrep": ["semgrep", "--config=auto", "--json"], }, "general": { "git_secrets": ["git-secrets", "--scan"], "trufflehog": ["trufflehog", "--json"], }, } # Common security patterns to check for self.security_patterns = { "hardcoded_secrets": [ r"password\s*=\s*['\"][^'\"]+['\"]", r"api_key\s*=\s*['\"][^'\"]+['\"]", r"secret\s*=\s*['\"][^'\"]+['\"]", r"token\s*=\s*['\"][^'\"]+['\"]", r"['\"]sk-[a-zA-Z0-9]{20,}['\"]", # OpenAI API keys r"['\"]xoxb-[0-9]{11,12}-[0-9]{11,12}-[a-zA-Z0-9]{24}['\"]", # Slack bot tokens ], "sql_injection": [ r"SELECT\s+\*\s+FROM\s+\w+\s+WHERE\s+.*\+.*", r"execute\s*\(\s*['\"].*\%.*['\"]", r"cursor\.execute\s*\(\s*f['\"].*\{.*\}.*['\"]", ], "path_traversal": [ r"open\s*\(\s*.*\+.*\.\./", r"file\s*=\s*.*\+.*\.\./", ], "weak_crypto": [ r"md5\s*\(", r"sha1\s*\(", r"DES\s*\(", ], } def _run_tool(self, cmd: List[str], cwd: str = ".") -> Dict[str, Any]: """ Execute a security tool and return structured result Args: cmd (List[str]): Command to execute cwd (str): Working directory Returns: Dict containing tool execution result """ try: result = subprocess.run( cmd, cwd=cwd, capture_output=True, text=True, timeout=300, # 5 minute timeout ) return { "success": True, "stdout": result.stdout.strip(), "stderr": result.stderr.strip(), "return_code": result.returncode, } except subprocess.TimeoutExpired: return { "success": False, "error": f"Tool timed out: {' '.join(cmd)}", "return_code": -1, } except FileNotFoundError: return { "success": False, "error": f"Tool not found: {cmd[0]}", "return_code": -1, } except Exception as e: return { "success": False, "error": f"Error executing tool: {str(e)}", "return_code": -1, } def _pattern_scan(self, filepath: str) -> List[Dict[str, Any]]: """ Scan file for security patterns Args: filepath (str): Path to file to scan Returns: List of security issues found """ try: content = read_file(filepath) issues = [] for category, patterns in self.security_patterns.items(): for pattern in patterns: matches = re.finditer( pattern, content, re.IGNORECASE | re.MULTILINE ) for match in matches: # Find line number line_num = content[: match.start()].count("\n") + 1 issues.append( { "category": category, "pattern": pattern, "match": match.group(), "line": line_num, "severity": self._get_pattern_severity(category), "filepath": filepath, } ) return issues except Exception as e: return [ { "error": f"Error scanning {filepath}: {str(e)}", "filepath": filepath, } ] def _get_pattern_severity(self, category: str) -> str: """Get severity level for security category""" severity_map = { "hardcoded_secrets": "high", "sql_injection": "high", "path_traversal": "medium", "weak_crypto": "medium", } return severity_map.get(category, "low") def _analyze_with_llm(self, security_findings: List[Dict]) -> str: """ Use LLM to analyze security findings and provide recommendations Args: security_findings (List[Dict]): List of security issues Returns: str: Analysis and recommendations """ try: if not security_findings: return "No security issues detected in the analysis." # Prepare summary of findings findings_summary = "" for finding in security_findings[:10]: # Limit to first 10 for prompt size findings_summary += f"- {finding.get('category', 'unknown')}: {finding.get('description', finding.get('match', 'No description'))}\n" prompt = f""" Analyze the following security findings and provide recommendations: Security Issues Found: {findings_summary} Please provide: 1. Risk assessment (High/Medium/Low) for each category 2. Specific remediation steps 3. General security best practices for this codebase 4. Priority order for fixing issues Keep the analysis concise and actionable. """ response = self.api_client.generate_text( prompt=prompt, model=self.config.get("model", "qwen2.5-coder:7b") ) if "error" in response: return f"Error generating security analysis: {response['error']}" if "choices" in response and len(response["choices"]) > 0: return response["choices"][0]["message"]["content"].strip() else: return "Security analysis completed. Please review findings manually." except Exception as e: return f"Error in LLM security analysis: {str(e)}" def security_scan(project_path: str = ".", tools: List[str] = None) -> Dict[str, Any]: """ Run security audit (bandit, npm audit, etc.) on project Args: project_path (str): Path to project directory tools (List[str]): Specific tools to run (None for auto-detect) Returns: Dict containing security scan results """ try: scanner = SecurityScanner() # Detect project type project_types = _detect_project_languages(project_path) results = { "project_path": project_path, "project_types": project_types, "tool_results": {}, "pattern_scan": {}, "summary": {}, } # Run appropriate security tools for project_type in project_types: if project_type in scanner.security_tools: type_tools = scanner.security_tools[project_type] for tool_name, cmd in type_tools.items(): if tools is None or tool_name in tools: print(f"Running {tool_name} for {project_type}...") # Customize command for specific tools if tool_name == "bandit": cmd_with_path = cmd + [project_path] elif tool_name == "npm_audit": cmd_with_path = cmd else: cmd_with_path = cmd + [project_path] result = scanner._run_tool(cmd_with_path, project_path) if result["success"]: # Parse tool output parsed_result = _parse_tool_output( tool_name, result["stdout"] ) results["tool_results"][tool_name] = parsed_result else: results["tool_results"][tool_name] = { "error": result.get("error", "Tool execution failed"), "available": False, } # Run pattern-based scanning on source files print("Running pattern-based security scan...") pattern_issues = [] for root, dirs, files in os.walk(project_path): # Skip common non-source directories dirs[:] = [ d for d in dirs if d not in {".git", "__pycache__", "node_modules", "venv", "env"} ] for file in files: if _is_source_file(file): filepath = os.path.join(root, file) file_issues = scanner._pattern_scan(filepath) pattern_issues.extend(file_issues) results["pattern_scan"] = { "issues": pattern_issues, "total_issues": len(pattern_issues), "files_scanned": len( [ f for root, dirs, files in os.walk(project_path) for f in files if _is_source_file(f) ] ), } # Generate summary total_issues = len(pattern_issues) high_severity = len([i for i in pattern_issues if i.get("severity") == "high"]) medium_severity = len( [i for i in pattern_issues if i.get("severity") == "medium"] ) # Add tool-based issue counts for tool_result in results["tool_results"].values(): if isinstance(tool_result, dict) and "issues" in tool_result: total_issues += len(tool_result["issues"]) results["summary"] = { "total_issues": total_issues, "high_severity": high_severity, "medium_severity": medium_severity, "tools_run": len(results["tool_results"]), "risk_level": "high" if high_severity > 0 else "medium" if medium_severity > 0 else "low", } # Get LLM analysis all_issues = pattern_issues.copy() for tool_result in results["tool_results"].values(): if isinstance(tool_result, dict) and "issues" in tool_result: all_issues.extend(tool_result["issues"]) results["llm_analysis"] = scanner._analyze_with_llm(all_issues) return results except Exception as e: return {"error": f"Error running security scan: {str(e)}"} def _detect_project_languages(project_path: str) -> List[str]: """Detect programming languages used in project""" languages = [] for root, dirs, files in os.walk(project_path): dirs[:] = [d for d in dirs if d not in {".git", "__pycache__", "node_modules"}] for file in files: ext = Path(file).suffix.lower() if ext == ".py": languages.append("python") elif ext in [".js", ".ts"]: languages.append("javascript") elif ext in [".java"]: languages.append("java") elif ext in [".cs"]: languages.append("csharp") elif ext in [".go"]: languages.append("go") return list(set(languages)) # Remove duplicates def _is_source_file(filename: str) -> bool: """Check if file is a source code file""" source_extensions = { ".py", ".js", ".ts", ".java", ".cs", ".go", ".rb", ".php", ".cpp", ".c", ".h", } return Path(filename).suffix.lower() in source_extensions def _parse_tool_output(tool_name: str, output: str) -> Dict[str, Any]: """Parse security tool output into structured format""" try: if tool_name == "bandit": if output.strip(): data = json.loads(output) return { "tool": "bandit", "issues": data.get("results", []), "metrics": data.get("metrics", {}), "total_issues": len(data.get("results", [])), } else: return {"tool": "bandit", "issues": [], "total_issues": 0} elif tool_name == "safety": if output.strip(): data = json.loads(output) return { "tool": "safety", "vulnerabilities": data, "total_issues": len(data) if isinstance(data, list) else 0, } else: return {"tool": "safety", "vulnerabilities": [], "total_issues": 0} elif tool_name == "npm_audit": if output.strip(): data = json.loads(output) vulnerabilities = data.get("vulnerabilities", {}) return { "tool": "npm_audit", "vulnerabilities": vulnerabilities, "total_issues": len(vulnerabilities), "summary": data.get("metadata", {}), } else: return {"tool": "npm_audit", "vulnerabilities": {}, "total_issues": 0} else: # Generic JSON parsing try: data = json.loads(output) return {"tool": tool_name, "data": data} except json.JSONDecodeError: return {"tool": tool_name, "raw_output": output} except Exception as e: return { "error": f"Error parsing {tool_name} output: {str(e)}", "raw_output": output, } def vulnerability_report(project_path: str = ".") -> Dict[str, Any]: """ Return structured security findings Args: project_path (str): Path to project directory Returns: Dict containing comprehensive vulnerability report """ try: # Run comprehensive security scan scan_results = security_scan(project_path) if "error" in scan_results: return scan_results # Compile comprehensive vulnerability report report = { "project_path": project_path, "scan_timestamp": str(Path().absolute()), # Simple timestamp "executive_summary": { "total_vulnerabilities": scan_results["summary"]["total_issues"], "high_risk": scan_results["summary"]["high_severity"], "medium_risk": scan_results["summary"]["medium_severity"], "overall_risk": scan_results["summary"]["risk_level"], }, "detailed_findings": [], "recommendations": scan_results.get( "llm_analysis", "No analysis available" ), "tools_used": list(scan_results["tool_results"].keys()), } # Compile detailed findings from all sources # Add pattern scan findings for issue in scan_results["pattern_scan"]["issues"]: if "error" not in issue: report["detailed_findings"].append( { "source": "pattern_scan", "category": issue.get("category", "unknown"), "severity": issue.get("severity", "low"), "description": f"Pattern match: {issue.get('match', 'No details')}", "file": issue.get("filepath", "unknown"), "line": issue.get("line", 0), } ) # Add tool scan findings for tool_name, tool_result in scan_results["tool_results"].items(): if isinstance(tool_result, dict) and not tool_result.get("error"): if tool_name == "bandit" and "issues" in tool_result: for issue in tool_result["issues"]: report["detailed_findings"].append( { "source": "bandit", "category": issue.get("test_name", "unknown"), "severity": issue.get("issue_severity", "low").lower(), "description": issue.get( "issue_text", "No description" ), "file": issue.get("filename", "unknown"), "line": issue.get("line_number", 0), } ) elif tool_name == "safety" and "vulnerabilities" in tool_result: for vuln in tool_result["vulnerabilities"]: report["detailed_findings"].append( { "source": "safety", "category": "dependency_vulnerability", "severity": "high", # Safety issues are typically high severity "description": vuln.get( "advisory", "Dependency vulnerability" ), "package": vuln.get("package_name", "unknown"), } ) elif tool_name == "npm_audit" and "vulnerabilities" in tool_result: for pkg_name, vuln_info in tool_result["vulnerabilities"].items(): if isinstance(vuln_info, dict): report["detailed_findings"].append( { "source": "npm_audit", "category": "dependency_vulnerability", "severity": vuln_info.get("severity", "medium"), "description": vuln_info.get( "title", "NPM package vulnerability" ), "package": pkg_name, } ) # Sort findings by severity severity_order = {"high": 0, "medium": 1, "low": 2} report["detailed_findings"].sort( key=lambda x: severity_order.get(x.get("severity", "low"), 2) ) return report except Exception as e: return {"error": f"Error generating vulnerability report: {str(e)}"} def check_secrets(project_path: str = ".") -> Dict[str, Any]: """ Scan for hardcoded secrets and sensitive information Args: project_path (str): Path to project directory Returns: Dict containing secrets analysis """ try: scanner = SecurityScanner() secrets_found = [] # Enhanced patterns for secrets detection secret_patterns = { "api_keys": [ r"['\"]?[Aa][Pp][Ii]_?[Kk][Ee][Yy]['\"]?\s*[:=]\s*['\"][a-zA-Z0-9_\-]{20,}['\"]", r"['\"]sk-[a-zA-Z0-9]{48}['\"]", # OpenAI API key r"['\"]xoxb-[0-9]{11,12}-[0-9]{11,12}-[a-zA-Z0-9]{24}['\"]", # Slack bot token ], "passwords": [ r"['\"]?[Pp][Aa][Ss][Ss][Ww][Oo][Rr][Dd]['\"]?\s*[:=]\s*['\"][^'\"]{6,}['\"]", ], "tokens": [ r"['\"]?[Tt][Oo][Kk][Ee][Nn]['\"]?\s*[:=]\s*['\"][a-zA-Z0-9_\-]{16,}['\"]", r"Bearer\s+[a-zA-Z0-9\-_.]{16,}", ], "database_urls": [ r"['\"]?[Dd][Aa][Tt][Aa][Bb][Aa][Ss][Ee]_?[Uu][Rr][Ll]['\"]?\s*[:=]\s*['\"][^'\"]+://[^'\"]+['\"]", r"mongodb://[^'\"\s]+", r"postgres://[^'\"\s]+", ], } # Scan all source files for root, dirs, files in os.walk(project_path): dirs[:] = [ d for d in dirs if d not in {".git", "__pycache__", "node_modules", "venv"} ] for file in files: if _is_source_file(file) or file.endswith((".env", ".config", ".ini")): filepath = os.path.join(root, file) try: content = read_file(filepath) for category, patterns in secret_patterns.items(): for pattern in patterns: matches = re.finditer( pattern, content, re.IGNORECASE | re.MULTILINE ) for match in matches: line_num = content[: match.start()].count("\n") + 1 secrets_found.append( { "category": category, "file": filepath, "line": line_num, "match": match.group()[:50] + "..." if len(match.group()) > 50 else match.group(), "severity": "high", "recommendation": f"Remove hardcoded {category.replace('_', ' ')} and use environment variables", } ) except Exception as e: continue # Skip files that can't be read return { "project_path": project_path, "secrets_found": secrets_found, "total_secrets": len(secrets_found), "categories": list(set([s["category"] for s in secrets_found])), "risk_level": "high" if secrets_found else "low", } except Exception as e: return {"error": f"Error checking for secrets: {str(e)}"} def security_best_practices_check(project_path: str = ".") -> Dict[str, Any]: """ Check for adherence to security best practices Args: project_path (str): Path to project directory Returns: Dict containing best practices analysis """ try: checks = [] # Check 1: Presence of security-related files security_files = { ".gitignore": "Prevents sensitive files from being committed", "requirements.txt": "Dependency management for Python projects", "package-lock.json": "Dependency locking for Node.js projects", ".env.example": "Template for environment variables", "SECURITY.md": "Security policy documentation", } for filename, purpose in security_files.items(): filepath = os.path.join(project_path, filename) checks.append( { "check": f"Security file: {filename}", "status": "pass" if os.path.exists(filepath) else "fail", "description": purpose, "severity": "medium" if filename in [".gitignore", "requirements.txt"] else "low", } ) # Check 2: .env files not in git (check .gitignore) gitignore_path = os.path.join(project_path, ".gitignore") if os.path.exists(gitignore_path): gitignore_content = read_file(gitignore_path) env_ignored = any( pattern in gitignore_content for pattern in [".env", "*.env"] ) checks.append( { "check": "Environment files ignored in git", "status": "pass" if env_ignored else "fail", "description": "Prevents accidental commit of sensitive environment variables", "severity": "high", } ) # Check 3: Requirements pinning (Python) req_path = os.path.join(project_path, "requirements.txt") if os.path.exists(req_path): req_content = read_file(req_path) pinned_deps = len(re.findall(r"==\d+\.\d+", req_content)) total_deps = len( [ line for line in req_content.splitlines() if line.strip() and not line.strip().startswith("#") ] ) if total_deps > 0: pin_ratio = pinned_deps / total_deps checks.append( { "check": "Dependency version pinning", "status": "pass" if pin_ratio > 0.8 else "warn" if pin_ratio > 0.5 else "fail", "description": f"{pinned_deps}/{total_deps} dependencies are version-pinned", "severity": "medium", } ) # Check 4: Secure HTTP headers (look for Flask/Django security configs) security_headers_found = False for root, dirs, files in os.walk(project_path): for file in files: if file.endswith(".py"): filepath = os.path.join(root, file) try: content = read_file(filepath) if any( header in content.lower() for header in [ "x-frame-options", "x-content-type-options", "strict-transport-security", ] ): security_headers_found = True break except: continue if security_headers_found: break checks.append( { "check": "Security headers configuration", "status": "pass" if security_headers_found else "warn", "description": "Web applications should implement security headers", "severity": "medium", } ) # Calculate overall score passed = len([c for c in checks if c["status"] == "pass"]) total = len(checks) score = (passed / total * 100) if total > 0 else 0 return { "project_path": project_path, "checks": checks, "summary": { "total_checks": total, "passed": passed, "failed": len([c for c in checks if c["status"] == "fail"]), "warnings": len([c for c in checks if c["status"] == "warn"]), "score": round(score, 1), }, "recommendations": _generate_security_recommendations(checks), } except Exception as e: return {"error": f"Error checking security best practices: {str(e)}"} def _generate_security_recommendations(checks: List[Dict]) -> List[str]: """Generate security recommendations based on failed checks""" recommendations = [] for check in checks: if check["status"] == "fail": if "gitignore" in check["check"].lower(): recommendations.append( "Create a .gitignore file to prevent sensitive files from being committed" ) elif "environment" in check["check"].lower(): recommendations.append( "Add .env files to .gitignore to prevent credential exposure" ) elif "pinning" in check["check"].lower(): recommendations.append( "Pin dependency versions to specific versions for security and reproducibility" ) elif "security headers" in check["check"].lower(): recommendations.append( "Implement security headers (X-Frame-Options, X-Content-Type-Options, etc.)" ) return recommendations # Example usage if __name__ == "__main__": # Run security scan scan_result = security_scan(".") print(f"Security scan completed: {scan_result['summary']}") # Generate vulnerability report vuln_report = vulnerability_report(".") print(f"Vulnerabilities found: {vuln_report.get('executive_summary', {})}")