Coverage for agentos/security/auditor.py: 29%
204 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-08 01:44 +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
15# ── Severity ──────────────────────────────────────────────────────────────────
18class AuditSeverity(Enum):
19 """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 """
45 id: str
46 category: str
47 severity: AuditSeverity
48 message: str
49 location: str = ""
50 recommendation: str = ""
51 cve: str | None = None
53 def to_dict(self) -> dict:
54 return {
55 "id": self.id,
56 "category": self.category,
57 "severity": self.severity.value,
58 "message": self.message,
59 "location": self.location,
60 "recommendation": self.recommendation,
61 "cve": self.cve,
62 }
65@dataclass
66class AuditReport:
67 """Aggregated report of all audit findings across scanned resources.
69 Attributes:
70 findings: List of individual findings.
71 scanned_files: Number of files scanned.
72 scanned_deps: Number of dependencies checked.
73 """
75 findings: list[AuditFinding] = field(default_factory=list)
76 scanned_files: int = 0
77 scanned_deps: int = 0
79 @property
80 def critical(self) -> int:
81 return sum(1 for f in self.findings if f.severity == AuditSeverity.CRITICAL)
83 @property
84 def high(self) -> int:
85 return sum(1 for f in self.findings if f.severity == AuditSeverity.HIGH)
87 @property
88 def medium(self) -> int:
89 return sum(1 for f in self.findings if f.severity == AuditSeverity.MEDIUM)
91 @property
92 def low(self) -> int:
93 return sum(1 for f in self.findings if f.severity == AuditSeverity.LOW)
95 def passed(self) -> bool:
96 return self.critical == 0 and self.high == 0
98 def summary(self) -> str:
99 return (
100 f"Audit: {self.critical}C / {self.high}H / {self.medium}M / {self.low}L "
101 f"across {self.scanned_files} files, {self.scanned_deps} deps — "
102 f"{'PASSED' if self.passed() else 'FAILED'}"
103 )
105 def to_dict(self) -> dict:
106 return {
107 "findings": [f.to_dict() for f in self.findings],
108 "summary": {
109 "critical": self.critical,
110 "high": self.high,
111 "medium": self.medium,
112 "low": self.low,
113 "passed": self.passed(),
114 "scanned_files": self.scanned_files,
115 "scanned_deps": self.scanned_deps,
116 },
117 }
119 def to_json(self) -> str:
120 import json
122 return json.dumps(self.to_dict(), indent=2, default=str)
124 def to_markdown(self) -> str:
125 lines = [
126 "# Security Audit Report",
127 "",
128 f"- **Scanned files**: {self.scanned_files}",
129 f"- **Scanned dependencies**: {self.scanned_deps}",
130 f"- **Result**: {'PASSED' if self.passed() else 'FAILED'}",
131 "",
132 "| Severity | Count |",
133 "|----------|-------|",
134 f"| CRITICAL | {self.critical} |",
135 f"| HIGH | {self.high} |",
136 f"| MEDIUM | {self.medium} |",
137 f"| LOW | {self.low} |",
138 "",
139 ]
140 if self.findings:
141 lines.append("## Findings")
142 lines.append("")
143 for f in self.findings:
144 lines.append(f"- **[{f.severity.value.upper()}]** `{f.id}` — {f.message}")
145 if f.recommendation:
146 lines.append(f" → {f.recommendation}")
147 return "\n".join(lines)
150# ── Built‑in checkers ────────────────────────────────────────────────────────
152# Known-vulnerable version patterns (illustrative)
153_VULN_PATTERNS: list[dict] = [
154 {"pkg": "django", "range": "<4.2.15", "cve": "CVE-2024-45230", "severity": "HIGH"},
155 {"pkg": "requests", "range": "<2.32.0", "cve": "CVE-2024-35195", "severity": "MEDIUM"},
156 {"pkg": "cryptography", "range": "<42.0.0", "cve": "CVE-2024-26130", "severity": "HIGH"},
157 {"pkg": "jinja2", "range": "<3.1.4", "cve": "CVE-2024-34064", "severity": "MEDIUM"},
158 {"pkg": "aiohttp", "range": "<3.9.4", "cve": "CVE-2024-30251", "severity": "HIGH"},
159]
161# Dangerous AST patterns
162_DANGEROUS_PATTERNS: list[dict] = [
163 {
164 "name": "eval-use",
165 "node": "Call",
166 "attr": "func.id",
167 "match": "eval",
168 "severity": "CRITICAL",
169 "msg": "eval() detected — arbitrary code execution risk",
170 },
171 {
172 "name": "exec-use",
173 "node": "Call",
174 "attr": "func.id",
175 "match": "exec",
176 "severity": "CRITICAL",
177 "msg": "exec() detected — arbitrary code execution risk",
178 },
179 {
180 "name": "pickle-load",
181 "node": "Call",
182 "attr": "func.attr",
183 "match": "loads",
184 "parent_attr": "func.value.id",
185 "parent_match": "pickle",
186 "severity": "HIGH",
187 "msg": "pickle.loads() on untrusted data may execute arbitrary code",
188 },
189 {
190 "name": "hardcoded-secret",
191 "node": "Assign",
192 "attr": "targets[0].id",
193 "match_re": r"(?i)(password|secret|api_key|token|access_key)\s*$",
194 "severity": "HIGH",
195 "msg": "Potential hard-coded secret",
196 },
197 {
198 "name": "shell-true",
199 "node": "Call",
200 "attr": "keywords",
201 "match_expr": "subprocess.Popen(… shell=True) or os.system() — command injection risk",
202 "severity": "HIGH",
203 "msg": "shell=True detected — command injection risk when input is untrusted",
204 },
205 {
206 "name": "insecure-deserialization",
207 "node": "Call",
208 "attr": "func.attr",
209 "match": "loads",
210 "parent_attr": "func.value.id",
211 "parent_match": "yaml",
212 "severity": "HIGH",
213 "msg": "yaml.load() without SafeLoader — arbitrary code execution risk",
214 },
215 {
216 "name": "md5-hash",
217 "node": "Call",
218 "attr": "func.attr",
219 "match": "md5",
220 "parent_attr": "func.value.id",
221 "parent_match": "hashlib",
222 "severity": "LOW",
223 "msg": "MD5 is cryptographically broken; use SHA-256",
224 },
225]
228# ── Dependency scanner ───────────────────────────────────────────────────────
231def _parse_requirements(content: str) -> list[tuple[str, str]]:
232 """Parse requirements.txt into (pkg, version_spec) pairs."""
233 deps: list[tuple[str, str]] = []
234 for line in content.splitlines():
235 line = line.strip()
236 if not line or line.startswith("#") or line.startswith("--"):
237 continue
238 # Normalise: requests==2.31.0 -> ('requests', '2.31.0')
239 m = re.match(r"^([a-zA-Z0-9_.-]+)\s*([><=!~]+\s*[\d.*]+(?:,\s*[><=!~]+\s*[\d.*]+)*)?", line)
240 if m:
241 pkg = m.group(1).lower()
242 ver = (m.group(2) or "").strip()
243 deps.append((pkg, ver))
244 return deps
247def _check_vuln_db(pkg: str, version_spec: str) -> list[AuditFinding]:
248 findings: list[AuditFinding] = []
249 for entry in _VULN_PATTERNS:
250 if entry["pkg"] != pkg:
251 continue
252 findings.append(
253 AuditFinding(
254 id=f"VULN-{entry['cve']}",
255 category="dependency",
256 severity=AuditSeverity(entry["severity"].lower()),
257 message=f"{pkg}{version_spec and ' ' + version_spec} is vulnerable — {entry['cve']}",
258 recommendation=f"Upgrade to {entry['range'].lstrip('<')}+",
259 cve=entry["cve"],
260 )
261 )
262 return findings
265def scan_dependencies(req_path: str | Path) -> AuditReport:
266 """Scan a requirements.txt or pyproject.toml for known-vulnerable dependencies."""
267 req_path = Path(req_path)
268 report = AuditReport()
270 if not req_path.exists():
271 report.findings.append(
272 AuditFinding(
273 id="DEP-001",
274 category="dependency",
275 severity=AuditSeverity.INFO,
276 message=f"Dependency file not found: {req_path}",
277 )
278 )
279 return report
281 content = req_path.read_text()
282 deps = _parse_requirements(content)
283 report.scanned_deps = len(deps)
285 for pkg, ver in deps:
286 report.findings.extend(_check_vuln_db(pkg, ver))
288 return report
291# ── Source scanner ────────────────────────────────────────────────────────────
294class _DangerousVisitor(ast.NodeVisitor):
295 """AST visitor that flags dangerous code patterns (exec, eval, subprocess, etc.)."""
297 def __init__(self) -> None:
298 self.findings: list[AuditFinding] = []
300 def _match(self, node: ast.AST, pattern: dict, lineno: int) -> AuditFinding | None:
301 name = pattern["name"]
302 severity = AuditSeverity(pattern["severity"].lower())
304 if "match_re" in pattern:
305 attr_path = pattern["attr"]
306 try:
307 val = eval(f"node.{attr_path}", {"node": node})
308 except Exception:
309 return None
310 if isinstance(val, str) and re.search(pattern["match_re"], val):
311 return AuditFinding(
312 id=f"SRC-{name.upper()}",
313 category="source",
314 severity=severity,
315 message=pattern["msg"],
316 location=f"line {lineno}",
317 recommendation="Remove or replace with a safe alternative",
318 )
319 return None
321 if "match_expr" in pattern:
322 # Special-case shell=True
323 for kw in getattr(node, "keywords", []):
324 if kw.arg == "shell" and getattr(kw.value, "value", None) is True:
325 return AuditFinding(
326 id=f"SRC-{name.upper()}",
327 category="source",
328 severity=severity,
329 message=pattern["msg"],
330 location=f"line {lineno}",
331 recommendation="Avoid shell=True; use list args",
332 )
333 return None
335 # Standard attr match
336 attr_path = pattern["attr"]
337 match_val = pattern["match"]
338 parent_attr = pattern.get("parent_attr")
339 parent_match = pattern.get("parent_match")
341 try:
342 val = eval(f"node.{attr_path}", {"node": node})
343 except Exception:
344 return None
346 if parent_attr is not None:
347 try:
348 pval = eval(f"node.{parent_attr}", {"node": node})
349 except Exception:
350 return None
351 if pval == parent_match and val == match_val:
352 return AuditFinding(
353 id=f"SRC-{name.upper()}",
354 category="source",
355 severity=severity,
356 message=pattern["msg"],
357 location=f"line {lineno}",
358 recommendation="Remove or replace with a safe alternative",
359 )
360 elif isinstance(val, str) and val == match_val:
361 return AuditFinding(
362 id=f"SRC-{name.upper()}",
363 category="source",
364 severity=severity,
365 message=pattern["msg"],
366 location=f"line {lineno}",
367 recommendation="Remove or replace with a safe alternative",
368 )
369 return None
371 def visit_Call(self, node: ast.Call) -> None: # noqa: N802
372 for pat in _DANGEROUS_PATTERNS:
373 if pat["node"] == "Call":
374 finding = self._match(node, pat, node.lineno)
375 if finding:
376 self.findings.append(finding)
377 self.generic_visit(node)
379 def visit_Assign(self, node: ast.Assign) -> None: # noqa: N802
380 for pat in _DANGEROUS_PATTERNS:
381 if pat["node"] == "Assign":
382 finding = self._match(node, pat, node.lineno)
383 if finding:
384 self.findings.append(finding)
385 self.generic_visit(node)
388def scan_source(source_dir: str | Path) -> AuditReport:
389 """AST-based source code security scan."""
390 source_dir = Path(source_dir)
391 report = AuditReport()
392 py_files = list(source_dir.rglob("*.py"))
394 for fpath in py_files:
395 try:
396 tree = ast.parse(fpath.read_text())
397 except SyntaxError:
398 continue
399 visitor = _DangerousVisitor()
400 visitor.visit(tree)
401 report.findings.extend(visitor.findings)
402 report.scanned_files += 1
404 return report
407# ── Security Auditor class ────────────────────────────────────────────────────
410class SecurityAuditor:
411 """High-level security auditor that orchestrates dependency and source scanning."""
413 def __init__(self, req_path: str | Path | None = None, source_dir: str | Path | None = None):
414 self.req_path: Path | None = Path(req_path) if req_path else None
415 self.source_dir: Path | None = Path(source_dir) if source_dir else None
417 def scan_dependencies(self, req_path: str | Path | None = None) -> AuditReport:
418 """Scan dependencies for known vulnerabilities."""
419 path = Path(req_path) if req_path else self.req_path
420 if not path or not path.exists():
421 return AuditReport()
422 return scan_dependencies(path)
424 def scan_source(self, paths: list[str | Path] | None = None) -> AuditReport:
425 """AST-based source code security scan."""
426 if paths:
427 report = AuditReport()
428 for p in paths:
429 r = scan_source(p)
430 report.findings.extend(r.findings)
431 report.scanned_files += r.scanned_files
432 return report
433 if not self.source_dir:
434 return AuditReport()
435 return scan_source(self.source_dir)
437 def full_audit(
438 self, source_dir: str | Path | None = None, req_path: str | Path | None = None
439 ) -> AuditReport:
440 """Run dependency + source audit and merge results."""
441 sd = source_dir or self.source_dir
442 rp = req_path or self.req_path
443 if not sd or not rp:
444 return AuditReport()
445 return full_audit(Path(sd), Path(rp))
448# ── Full audit (module-level) ─────────────────────────────────────────────────
451def full_audit(
452 source_dir: str | Path,
453 req_path: str | Path,
454) -> AuditReport:
455 """Run dependency + source audit and merge results."""
456 dep_report = scan_dependencies(req_path)
457 src_report = scan_source(source_dir)
459 merged = AuditReport(
460 findings=dep_report.findings + src_report.findings,
461 scanned_files=src_report.scanned_files,
462 scanned_deps=dep_report.scanned_deps,
463 )
464 return merged
467# ── Report export ─────────────────────────────────────────────────────────────
470def export_report(report: AuditReport, fmt: str = "json") -> str:
471 """Export audit report to JSON or Markdown."""
472 if fmt == "json":
473 return json.dumps(report.to_dict(), indent=2)
474 # Markdown
475 lines = [
476 "# Security Audit Report",
477 "",
478 f"**Summary**: {report.summary()}",
479 "",
480 "| Severity | Count |",
481 "|----------|-------|",
482 f"| Critical | {report.critical} |",
483 f"| High | {report.high} |",
484 f"| Medium | {report.medium} |",
485 f"| Low | {report.low} |",
486 "",
487 "## Findings",
488 "",
489 ]
490 for f in sorted(report.findings, key=lambda x: (4 - list(AuditSeverity).index(x.severity))):
491 lines.append(f"- **[{f.severity.value.upper()}]** {f.message} ")
492 if f.location:
493 lines.append(f" *Location*: {f.location}")
494 if f.recommendation:
495 lines.append(f" *Fix*: {f.recommendation}")
496 if f.cve:
497 lines.append(f" *CVE*: {f.cve}")
498 lines.append("")
500 return "\n".join(lines)