ai-code-review-agent / app /agents /bug_detection_agent.py
Padmanav's picture
feat: add security depth — secret scanning, dependency vulns, report surface
d860535
Raw
History Blame Contribute Delete
4.72 kB
from app.core.logger import get_logger
from app.tools.static_analyzer import run_full_analysis
from app.tools.pytest_runner import run_tests
from app.tools.file_scanner import get_python_files, sanitize_for_prompt
from app.core.llm import call_llm_for_json
from app.core.prompts import BUG_DETECTION_PROMPT
from app.models.issue import IssueReport, Bug, Warning, Suggestion
from app.tools.ast_parser import scan_repository_security, scan_secrets_regex
from app.tools.dependency_scanner import scan_dependencies
logger = get_logger(__name__)
async def run(local_path: str) -> IssueReport:
"""
Detect bugs, warnings and suggestions in a repository.
"""
logger.info("Starting bug detection agent", extra={"path": local_path})
# Step 1 - Run static analysis
static_results = run_full_analysis(local_path)
# Step 1.5 - AST-based security scan (no LLM, zero cost, deterministic)
ast_security_findings = scan_repository_security(local_path)
# Regex-based secret scan (covers non-Python files: .env, YAML, JS, etc.)
secret_findings = scan_secrets_regex(local_path)
logger.info("Secret scan complete", extra={"findings": len(secret_findings)})
# Dependency vulnerability scan
dep_vulnerabilities = scan_dependencies(local_path)
logger.info("Dependency scan complete", extra={"vulnerabilities": len(dep_vulnerabilities)})
logger.info(
"AST security scan complete",
extra={"findings": len(ast_security_findings)},
)
# Step 2 - Run tests
test_results = run_tests(local_path)
# Step 3 - Read sample of code for LLM
python_files = get_python_files(local_path)
code_samples = []
for file_path in python_files[:5]: # Limit to first 5 files
try:
with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
content = f.read()[:2000] # Limit to 2000 chars per file
content = sanitize_for_prompt(content)
code_samples.append(f"### {file_path}\n{content}")
except Exception as e: # nosec B110
logger.warning(
"Failed to read file", extra={"file": file_path, "error": str(e)}
)
# Step 4 - Build context for LLM
code_info = f"""
Static Analysis Results:
Ruff Issues: {len(static_results["ruff"]["issues"])}
Bandit Issues: {len(static_results["bandit"]["issues"])}
Ruff Output:
{chr(10).join(static_results["ruff"]["issues"][:20])}
Bandit Output:
{chr(10).join(static_results["bandit"]["issues"][:20])}
AST Security Findings:
{chr(10).join(f"[{f['severity'].upper()}] {f['rule']} at line {f['line']} in {f['file']}: {f['detail']}" for f in ast_security_findings[:20])}
Dependency Vulnerabilities:
{chr(10).join(f"[VULN] {v['package']}=={v['version']} {v['id']}: {v['description'][:100]}" for v in dep_vulnerabilities[:10]) or "None found"}
Secret Scan (non-Python files):
{chr(10).join(f"[{f['severity'].upper()}] {f['rule']} at line {f['line']} in {f['file']}" for f in secret_findings[:10]) or "None found"}
Test Results:
Passed: {test_results["passed"]}
Failed: {test_results["failed"]}
Errors: {test_results["errors"]}
Code Samples:
{chr(10).join(code_samples)}
"""
# Step 5 - Skip LLM if static analysis found nothing and no code to review
total_static = (
len(static_results["ruff"]["issues"])
+ len(static_results["bandit"]["issues"])
+ len(ast_security_findings)
)
if total_static == 0 and not code_samples:
logger.info("No static issues and no code samples — skipping LLM call")
return IssueReport()
# Call LLM only when there's something meaningful to analyse
prompt = BUG_DETECTION_PROMPT.format(code_info=code_info)
llm_data = await call_llm_for_json(prompt, task="bug_detection")
# call_llm_for_json handles JSON parsing and retry internally; llm_data is already a dict
# Step 6 - Build IssueReport
critical = [
Bug(**bug)
if isinstance(bug, dict)
else Bug(description=str(bug), file="unknown")
for bug in llm_data.get("critical", [])
]
warnings = [
Warning(**w)
if isinstance(w, dict)
else Warning(description=str(w), file="unknown")
for w in llm_data.get("warnings", [])
]
suggestions = [
Suggestion(**s)
if isinstance(s, dict)
else Suggestion(description=str(s), file="unknown")
for s in llm_data.get("suggestions", [])
]
report = IssueReport(critical=critical, warnings=warnings, suggestions=suggestions)
logger.info(
"Bug detection complete",
extra={"critical": report.total_critical, "warnings": report.total_warnings},
)
return report