Coverage for agentos/security/auditor.py: 29%
205 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 10:59 +0800
1"""AgentOS Security Auditor — automated vulnerability scanning and code analysis.
3Audits dependencies and source patterns for common security issues.
4"""
6from __future__ import annotations
8import ast
9import json
10import re
11from dataclasses import dataclass, field
12from enum import Enum
13from pathlib import Path
14from typing import Optional
16# ── Severity ──────────────────────────────────────────────────────────────────
19class AuditSeverity(Enum):
20 """Severity level for security audit findings."""
21 CRITICAL = "critical"
22 HIGH = "high"
23 MEDIUM = "medium"
24 LOW = "low"
25 INFO = "info"
28# ── Data classes ──────────────────────────────────────────────────────────────
31@dataclass
32class AuditFinding:
33 """A single security finding from an audit scan.
35 Attributes:
36 id: Unique finding identifier.
37 category: Finding category (e.g., injection, hardcoded_secret).
38 severity: Severity level.
39 message: Human-readable description.
40 location: File path and line reference.
41 recommendation: Suggested remediation.
42 cve: Optional CVE identifier if known.
43 """
44 id: str
45 category: str
46 severity: AuditSeverity
47 message: str
48 location: str = ""
49 recommendation: str = ""
50 cve: Optional[str] = None
52 def to_dict(self) -> dict:
53 return {
54 "id": self.id,
55 "category": self.category,
56 "severity": self.severity.value,
57 "message": self.message,
58 "location": self.location,
59 "recommendation": self.recommendation,
60 "cve": self.cve,
61 }
64@dataclass
65class AuditReport:
66 """Aggregated report of all audit findings across scanned resources.
68 Attributes:
69 findings: List of individual findings.
70 scanned_files: Number of files scanned.
71 scanned_deps: Number of dependencies checked.
72 """
73 findings: list[AuditFinding] = field(default_factory=list)
74 scanned_files: int = 0
75 scanned_deps: int = 0
77 @property
78 def critical(self) -> int:
79 return sum(1 for f in self.findings if f.severity == AuditSeverity.CRITICAL)
81 @property
82 def high(self) -> int:
83 return sum(1 for f in self.findings if f.severity == AuditSeverity.HIGH)
85 @property
86 def medium(self) -> int:
87 return sum(1 for f in self.findings if f.severity == AuditSeverity.MEDIUM)
89 @property
90 def low(self) -> int:
91 return sum(1 for f in self.findings if f.severity == AuditSeverity.LOW)
93 def passed(self) -> bool:
94 return self.critical == 0 and self.high == 0
96 def summary(self) -> str:
97 return (
98 f"Audit: {self.critical}C / {self.high}H / {self.medium}M / {self.low}L "
99 f"across {self.scanned_files} files, {self.scanned_deps} deps — "
100 f"{'PASSED' if self.passed() else 'FAILED'}"
101 )
103 def to_dict(self) -> dict:
104 return {
105 "findings": [f.to_dict() for f in self.findings],
106 "summary": {
107 "critical": self.critical,
108 "high": self.high,
109 "medium": self.medium,
110 "low": self.low,
111 "passed": self.passed(),
112 "scanned_files": self.scanned_files,
113 "scanned_deps": self.scanned_deps,
114 },
115 }
117 def to_json(self) -> str:
118 import json
119 return json.dumps(self.to_dict(), indent=2, default=str)
121 def to_markdown(self) -> str:
122 lines = [
123 "# Security Audit Report",
124 "",
125 f"- **Scanned files**: {self.scanned_files}",
126 f"- **Scanned dependencies**: {self.scanned_deps}",
127 f"- **Result**: {'PASSED' if self.passed() else 'FAILED'}",
128 "",
129 "| Severity | Count |",
130 "|----------|-------|",
131 f"| CRITICAL | {self.critical} |",
132 f"| HIGH | {self.high} |",
133 f"| MEDIUM | {self.medium} |",
134 f"| LOW | {self.low} |",
135 "",
136 ]
137 if self.findings:
138 lines.append("## Findings")
139 lines.append("")
140 for f in self.findings:
141 lines.append(f"- **[{f.severity.value.upper()}]** `{f.id}` — {f.message}")
142 if f.recommendation:
143 lines.append(f" → {f.recommendation}")
144 return "\n".join(lines)
147# ── Built‑in checkers ────────────────────────────────────────────────────────
149# Known-vulnerable version patterns (illustrative)
150_VULN_PATTERNS: list[dict] = [
151 {"pkg": "django", "range": "<4.2.15", "cve": "CVE-2024-45230", "severity": "HIGH"},
152 {"pkg": "requests", "range": "<2.32.0", "cve": "CVE-2024-35195", "severity": "MEDIUM"},
153 {"pkg": "cryptography", "range": "<42.0.0", "cve": "CVE-2024-26130", "severity": "HIGH"},
154 {"pkg": "jinja2", "range": "<3.1.4", "cve": "CVE-2024-34064", "severity": "MEDIUM"},
155 {"pkg": "aiohttp", "range": "<3.9.4", "cve": "CVE-2024-30251", "severity": "HIGH"},
156]
158# Dangerous AST patterns
159_DANGEROUS_PATTERNS: list[dict] = [
160 {"name": "eval-use", "node": "Call", "attr": "func.id", "match": "eval", "severity": "CRITICAL",
161 "msg": "eval() detected — arbitrary code execution risk"},
162 {"name": "exec-use", "node": "Call", "attr": "func.id", "match": "exec", "severity": "CRITICAL",
163 "msg": "exec() detected — arbitrary code execution risk"},
164 {"name": "pickle-load", "node": "Call", "attr": "func.attr", "match": "loads",
165 "parent_attr": "func.value.id", "parent_match": "pickle", "severity": "HIGH",
166 "msg": "pickle.loads() on untrusted data may execute arbitrary code"},
167 {"name": "hardcoded-secret", "node": "Assign", "attr": "targets[0].id",
168 "match_re": r"(?i)(password|secret|api_key|token|access_key)\s*$", "severity": "HIGH",
169 "msg": "Potential hard-coded secret"},
170 {"name": "shell-true", "node": "Call", "attr": "keywords",
171 "match_expr": "subprocess.Popen(… shell=True) or os.system() — command injection risk",
172 "severity": "HIGH",
173 "msg": "shell=True detected — command injection risk when input is untrusted"},
174 {"name": "insecure-deserialization", "node": "Call", "attr": "func.attr", "match": "loads",
175 "parent_attr": "func.value.id", "parent_match": "yaml", "severity": "HIGH",
176 "msg": "yaml.load() without SafeLoader — arbitrary code execution risk"},
177 {"name": "md5-hash", "node": "Call", "attr": "func.attr", "match": "md5",
178 "parent_attr": "func.value.id", "parent_match": "hashlib", "severity": "LOW",
179 "msg": "MD5 is cryptographically broken; use SHA-256"},
180]
183# ── Dependency scanner ───────────────────────────────────────────────────────
186def _parse_requirements(content: str) -> list[tuple[str, str]]:
187 """Parse requirements.txt into (pkg, version_spec) pairs."""
188 deps: list[tuple[str, str]] = []
189 for line in content.splitlines():
190 line = line.strip()
191 if not line or line.startswith("#") or line.startswith("--"):
192 continue
193 # Normalise: requests==2.31.0 -> ('requests', '2.31.0')
194 m = re.match(r"^([a-zA-Z0-9_.-]+)\s*([><=!~]+\s*[\d.*]+(?:,\s*[><=!~]+\s*[\d.*]+)*)?", line)
195 if m:
196 pkg = m.group(1).lower()
197 ver = (m.group(2) or "").strip()
198 deps.append((pkg, ver))
199 return deps
202def _check_vuln_db(pkg: str, version_spec: str) -> list[AuditFinding]:
203 findings: list[AuditFinding] = []
204 for entry in _VULN_PATTERNS:
205 if entry["pkg"] != pkg:
206 continue
207 findings.append(
208 AuditFinding(
209 id=f"VULN-{entry['cve']}",
210 category="dependency",
211 severity=AuditSeverity(entry["severity"].lower()),
212 message=f"{pkg}{version_spec and ' ' + version_spec} is vulnerable — {entry['cve']}",
213 recommendation=f"Upgrade to {entry['range'].lstrip('<')}+",
214 cve=entry["cve"],
215 )
216 )
217 return findings
220def scan_dependencies(req_path: str | Path) -> AuditReport:
221 """Scan a requirements.txt or pyproject.toml for known-vulnerable dependencies."""
222 req_path = Path(req_path)
223 report = AuditReport()
225 if not req_path.exists():
226 report.findings.append(
227 AuditFinding(
228 id="DEP-001",
229 category="dependency",
230 severity=AuditSeverity.INFO,
231 message=f"Dependency file not found: {req_path}",
232 )
233 )
234 return report
236 content = req_path.read_text()
237 deps = _parse_requirements(content)
238 report.scanned_deps = len(deps)
240 for pkg, ver in deps:
241 report.findings.extend(_check_vuln_db(pkg, ver))
243 return report
246# ── Source scanner ────────────────────────────────────────────────────────────
249class _DangerousVisitor(ast.NodeVisitor):
250 """AST visitor that flags dangerous code patterns (exec, eval, subprocess, etc.)."""
251 def __init__(self) -> None:
252 self.findings: list[AuditFinding] = []
254 def _match(self, node: ast.AST, pattern: dict, lineno: int) -> Optional[AuditFinding]:
255 name = pattern["name"]
256 severity = AuditSeverity(pattern["severity"].lower())
258 if "match_re" in pattern:
259 attr_path = pattern["attr"]
260 try:
261 val = eval(f"node.{attr_path}", {"node": node})
262 except Exception:
263 return None
264 if isinstance(val, str) and re.search(pattern["match_re"], val):
265 return AuditFinding(
266 id=f"SRC-{name.upper()}",
267 category="source",
268 severity=severity,
269 message=pattern["msg"],
270 location=f"line {lineno}",
271 recommendation="Remove or replace with a safe alternative",
272 )
273 return None
275 if "match_expr" in pattern:
276 # Special-case shell=True
277 for kw in getattr(node, "keywords", []):
278 if kw.arg == "shell" and getattr(kw.value, "value", None) is True:
279 return AuditFinding(
280 id=f"SRC-{name.upper()}",
281 category="source",
282 severity=severity,
283 message=pattern["msg"],
284 location=f"line {lineno}",
285 recommendation="Avoid shell=True; use list args",
286 )
287 return None
289 # Standard attr match
290 attr_path = pattern["attr"]
291 match_val = pattern["match"]
292 parent_attr = pattern.get("parent_attr")
293 parent_match = pattern.get("parent_match")
295 try:
296 val = eval(f"node.{attr_path}", {"node": node})
297 except Exception:
298 return None
300 if parent_attr is not None:
301 try:
302 pval = eval(f"node.{parent_attr}", {"node": node})
303 except Exception:
304 return None
305 if pval == parent_match and val == match_val:
306 return AuditFinding(
307 id=f"SRC-{name.upper()}",
308 category="source",
309 severity=severity,
310 message=pattern["msg"],
311 location=f"line {lineno}",
312 recommendation="Remove or replace with a safe alternative",
313 )
314 elif isinstance(val, str) and val == match_val:
315 return AuditFinding(
316 id=f"SRC-{name.upper()}",
317 category="source",
318 severity=severity,
319 message=pattern["msg"],
320 location=f"line {lineno}",
321 recommendation="Remove or replace with a safe alternative",
322 )
323 return None
325 def visit_Call(self, node: ast.Call) -> None: # noqa: N802
326 for pat in _DANGEROUS_PATTERNS:
327 if pat["node"] == "Call":
328 finding = self._match(node, pat, node.lineno)
329 if finding:
330 self.findings.append(finding)
331 self.generic_visit(node)
333 def visit_Assign(self, node: ast.Assign) -> None: # noqa: N802
334 for pat in _DANGEROUS_PATTERNS:
335 if pat["node"] == "Assign":
336 finding = self._match(node, pat, node.lineno)
337 if finding:
338 self.findings.append(finding)
339 self.generic_visit(node)
342def scan_source(source_dir: str | Path) -> AuditReport:
343 """AST-based source code security scan."""
344 source_dir = Path(source_dir)
345 report = AuditReport()
346 py_files = list(source_dir.rglob("*.py"))
348 for fpath in py_files:
349 try:
350 tree = ast.parse(fpath.read_text())
351 except SyntaxError:
352 continue
353 visitor = _DangerousVisitor()
354 visitor.visit(tree)
355 report.findings.extend(visitor.findings)
356 report.scanned_files += 1
358 return report
361# ── Security Auditor class ────────────────────────────────────────────────────
364class SecurityAuditor:
365 """High-level security auditor that orchestrates dependency and source scanning."""
367 def __init__(self, req_path: Optional[str | Path] = None, source_dir: Optional[str | Path] = None):
368 self.req_path: Optional[Path] = Path(req_path) if req_path else None
369 self.source_dir: Optional[Path] = Path(source_dir) if source_dir else None
371 def scan_dependencies(self, req_path: Optional[str | Path] = None) -> AuditReport:
372 """Scan dependencies for known vulnerabilities."""
373 path = Path(req_path) if req_path else self.req_path
374 if not path or not path.exists():
375 return AuditReport()
376 return scan_dependencies(path)
378 def scan_source(self, paths: Optional[list[str | Path]] = None) -> AuditReport:
379 """AST-based source code security scan."""
380 if paths:
381 report = AuditReport()
382 for p in paths:
383 r = scan_source(p)
384 report.findings.extend(r.findings)
385 report.scanned_files += r.scanned_files
386 return report
387 if not self.source_dir:
388 return AuditReport()
389 return scan_source(self.source_dir)
391 def full_audit(self, source_dir: Optional[str | Path] = None, req_path: Optional[str | Path] = None) -> AuditReport:
392 """Run dependency + source audit and merge results."""
393 sd = source_dir or self.source_dir
394 rp = req_path or self.req_path
395 if not sd or not rp:
396 return AuditReport()
397 return full_audit(Path(sd), Path(rp))
400# ── Full audit (module-level) ─────────────────────────────────────────────────
403def full_audit(
404 source_dir: str | Path,
405 req_path: str | Path,
406) -> AuditReport:
407 """Run dependency + source audit and merge results."""
408 dep_report = scan_dependencies(req_path)
409 src_report = scan_source(source_dir)
411 merged = AuditReport(
412 findings=dep_report.findings + src_report.findings,
413 scanned_files=src_report.scanned_files,
414 scanned_deps=dep_report.scanned_deps,
415 )
416 return merged
419# ── Report export ─────────────────────────────────────────────────────────────
422def export_report(report: AuditReport, fmt: str = "json") -> str:
423 """Export audit report to JSON or Markdown."""
424 if fmt == "json":
425 return json.dumps(report.to_dict(), indent=2)
426 # Markdown
427 lines = [
428 "# Security Audit Report",
429 "",
430 f"**Summary**: {report.summary()}",
431 "",
432 "| Severity | Count |",
433 "|----------|-------|",
434 f"| Critical | {report.critical} |",
435 f"| High | {report.high} |",
436 f"| Medium | {report.medium} |",
437 f"| Low | {report.low} |",
438 "",
439 "## Findings",
440 "",
441 ]
442 for f in sorted(report.findings, key=lambda x: (4 - list(AuditSeverity).index(x.severity))):
443 lines.append(f"- **[{f.severity.value.upper()}]** {f.message} ")
444 if f.location:
445 lines.append(f" *Location*: {f.location}")
446 if f.recommendation:
447 lines.append(f" *Fix*: {f.recommendation}")
448 if f.cve:
449 lines.append(f" *CVE*: {f.cve}")
450 lines.append("")
452 return "\n".join(lines)