Clover/tools/security_tools.py
Jarian Cottingham 392bcfa2ef feat: Implement AI agent with multi-turn conversation and tool calling
Major enhancements to Clover CLI:

 New Features:
- AI agent with multi-turn conversation capabilities
- Tool calling system with 11+ tools for file operations, Git, linting, etc.
- Step-by-step AI assistance with play-by-play commentary
- Enhanced interactive mode with better UX

🔧 Core Components Added:
- ai_agent.py: Main AI agent with conversation management
- models/: API client and model management system
- Comprehensive tool system for development tasks

🛠️ Tools Available:
- File operations (create, read, update, delete)
- Command execution with safety checks
- Git operations (status, diff, commit, push)
- Code linting and formatting
- Project structure analysis
- Security scanning and dependency management

💡 User Experience:
- Real-time tool execution summaries
- File creation with full path visibility
- Error handling and retry mechanisms
- Clean conversation flow until task completion

🧹 Repository Cleanup:
- Added comprehensive .gitignore
- Removed __pycache__ directories and build artifacts
- Organized project structure

The AI can now actually create files, run commands, and work through complex
development tasks step-by-step with full transparency.
2026-01-15 01:23:40 -06:00

811 lines
29 KiB
Python

"""
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', {})}")