diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index f5bb04c..e526c6d 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -28,3 +28,12 @@ repos:
- id: detect-secrets
args: [--baseline, .secrets.baseline]
exclude: .env.example
+
+ - repo: local
+ hooks:
+ - id: zeroclaw-scan
+ name: ZeroClaw Security Scan
+ entry: python -m zeroclaw.cli scan --target . --stream local-commit
+ language: system
+ pass_filenames: false
+ always_run: true
diff --git a/src/zeroclaw/cli.py b/src/zeroclaw/cli.py
index c29ff3c..55492c3 100644
--- a/src/zeroclaw/cli.py
+++ b/src/zeroclaw/cli.py
@@ -3,12 +3,13 @@
import argparse
import json
+import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
-from .models import ScanResult
+from zeroclaw.models import ScanResult
def main() -> None:
@@ -32,7 +33,7 @@ def main() -> None:
report_cmd = subparsers.add_parser("report", help="Generate a report from a saved scan")
report_cmd.add_argument("--format", choices=["terminal", "json"], default="terminal")
- report_cmd.add_argument("--input", default="zeroclaw_scan.json", help="Input JSON scan file")
+ report_cmd.add_argument("--input", default="reports/latest_scan.json", help="Input JSON scan file")
args = parser.parse_args()
@@ -49,17 +50,17 @@ def main() -> None:
def _run_scan(args: argparse.Namespace) -> None:
"""Execute the 3-phase scan pipeline: Gather → Enrich → Report."""
- from .scanners.pattern_scanner import scan_patterns
- from .scanners.dependency_scanner import scan_dependencies
- from .scanners.secret_scanner import scan_secrets
- from .scanners.auth_scanner import scan_fastapi_auth, scan_supabase_rls
+ from zeroclaw.scanners.pattern_scanner import scan_patterns
+ from zeroclaw.scanners.dependency_scanner import scan_dependencies
+ from zeroclaw.scanners.secret_scanner import scan_secrets
+ from zeroclaw.scanners.auth_scanner import scan_fastapi_auth, scan_supabase_rls
target = Path(args.target).resolve()
if not target.exists():
print(f"Error: target directory {target} does not exist")
sys.exit(1)
- stream = args.stream
+ stream = args.stream if args.stream else target.name
# ── Phase 1: Gathering ──────────────────────────────────────────────
print(f"[ZeroClaw] Phase 1/3 — Static analysis on {target}")
@@ -91,9 +92,8 @@ def _run_scan(args: argparse.Namespace) -> None:
raw_findings.extend(rls)
# Tag findings with stream label
- if stream:
- for f in raw_findings:
- f.stream = stream
+ for f in raw_findings:
+ f.stream = stream
print(f"\n[ZeroClaw] Total raw findings: {len(raw_findings)}")
@@ -113,7 +113,7 @@ def _run_scan(args: argparse.Namespace) -> None:
)
if enrichable:
- from .agent_client import ZeroClawClient
+ from zeroclaw.agent_client import ZeroClawClient
client = ZeroClawClient()
@@ -151,26 +151,43 @@ def _run_scan(args: argparse.Namespace) -> None:
# Build the ScanResult
stats = {}
for f in enriched_findings:
- sev = f.severity.value
- stats[sev] = stats.get(sev, 0) + 1
+ if f.false_positive:
+ continue
+ stats[f.severity.value] = stats.get(f.severity.value, 0) + 1
+ stats[f.category.value] = stats.get(f.category.value, 0) + 1
result = ScanResult(
stream=stream,
- repo_url=str(target),
+ repo_url=f"https://github.com/lifeatlas/{stream}",
scanned_at=datetime.now(timezone.utc),
findings=enriched_findings,
stats=stats,
)
- from .reporter import generate_terminal_report, generate_json_report
+ from zeroclaw.reporter import generate_terminal_report, generate_json_report
+
+ # Ensure reports directory exists
+ reports_dir = Path("reports")
+ reports_dir.mkdir(parents=True, exist_ok=True)
+
+ # Save to reports/latest_scan.json and reports/{stream}_scan.json
+ json_data = generate_json_report(result)
+ latest_file = reports_dir / "latest_scan.json"
+ stream_file = reports_dir / f"{stream}_scan.json"
+
+ # Datetime serializer for JSON
+ def dt_serializer(obj):
+ if isinstance(obj, datetime):
+ return obj.isoformat()
+ raise TypeError("Type not serializable")
+
+ with open(latest_file, "w", encoding="utf-8") as fh:
+ json.dump(json_data, fh, indent=2, default=dt_serializer)
+ with open(stream_file, "w", encoding="utf-8") as fh:
+ json.dump(json_data, fh, indent=2, default=dt_serializer)
if args.format == "json":
- report = generate_json_report(result)
- # Save to file
- out_path = Path("zeroclaw_scan.json")
- with open(out_path, "w", encoding="utf-8") as fh:
- json.dump(report, fh, indent=2, default=str)
- print(f"\n[ZeroClaw] JSON report saved to {out_path}")
+ print(json.dumps(json_data, indent=2, default=dt_serializer))
else:
output = generate_terminal_report(result)
print(output)
@@ -185,12 +202,17 @@ def _run_report(args: argparse.Namespace) -> None:
print(f"Error: input file {input_path} does not exist")
sys.exit(1)
- with open(input_path, encoding="utf-8") as fh:
- data = json.load(fh)
-
- result = ScanResult(**data)
+ try:
+ with open(input_path, encoding="utf-8") as fh:
+ data = json.load(fh)
+
+ # Parse back to ScanResult model (Pydantic resolves datetime strings)
+ result = ScanResult(**data)
+ except Exception as e:
+ print(f"Error reading or parsing scan results: {e}")
+ sys.exit(1)
- from .reporter import generate_terminal_report, generate_json_report
+ from zeroclaw.reporter import generate_terminal_report, generate_json_report
if args.format == "json":
report = generate_json_report(result)
diff --git a/src/zeroclaw/reporter.py b/src/zeroclaw/reporter.py
index 31369b2..772b111 100644
--- a/src/zeroclaw/reporter.py
+++ b/src/zeroclaw/reporter.py
@@ -1,229 +1,216 @@
"""Report generation: terminal, JSON, PDF."""
from __future__ import annotations
+from datetime import datetime
+import json
+from collections import defaultdict
-from zeroclaw.models import ScanResult, StreamScore
+from zeroclaw.models import ScanResult, StreamScore, Finding, Severity, Category
-def generate_terminal_report(result: ScanResult) -> str:
- """Rich terminal output of scan results with ZeroClaw enrichment data."""
- lines: list[str] = []
+def calculate_stream_score(result: ScanResult) -> StreamScore:
+ """Calculate 0-10 security score for a stream.
+
+ Formula:
+ Base score is 10.0.
+ For each finding:
+ - Critical: -2.5
+ - High: -1.5
+ - Medium: -0.75
+ - Low: -0.25
+ - Info: -0.0
+ The score is bounded to [0.0, 10.0] and rounded to 1 decimal place.
+ """
+ severity_deductions = {
+ Severity.CRITICAL: 2.5,
+ Severity.HIGH: 1.5,
+ Severity.MEDIUM: 0.75,
+ Severity.LOW: 0.25,
+ Severity.INFO: 0.0,
+ }
- # ── Header ──────────────────────────────────────────────────────────
- lines.append("")
- lines.append("=" * 72)
- lines.append(" ZEROCLAW SECURITY SCAN REPORT")
- lines.append("=" * 72)
- lines.append("")
- lines.append(f" Repository: {result.repo_url}")
- if result.stream:
- lines.append(f" Stream: {result.stream}")
- lines.append(f" Scanned at: {result.scanned_at.isoformat()}")
- lines.append("")
+ counts = {
+ "critical": 0,
+ "high": 0,
+ "medium": 0,
+ "low": 0,
+ "info": 0,
+ }
- # ── Executive Summary ───────────────────────────────────────────────
- total = len(result.findings)
- lines.append("─" * 72)
- lines.append(" EXECUTIVE SUMMARY")
- lines.append("─" * 72)
- lines.append("")
+ # Count findings by severity, ignoring false positives
+ for finding in result.findings:
+ if finding.false_positive:
+ continue
+ counts[finding.severity.value] = counts.get(finding.severity.value, 0) + 1
+
+ # Calculate score
+ deductions = sum(counts[sev] * severity_deductions[Severity(sev)] for sev in counts)
+ score_val = max(0.0, 10.0 - deductions)
+ score_val = round(score_val, 1)
+
+ # Prioritize and identify top issues
+ severity_order = {
+ Severity.CRITICAL: 0,
+ Severity.HIGH: 1,
+ Severity.MEDIUM: 2,
+ Severity.LOW: 3,
+ Severity.INFO: 4,
+ }
- if total == 0:
- lines.append(" ✅ No vulnerabilities detected.")
- lines.append("")
- lines.append("=" * 72)
- return "\n".join(lines)
+ real_findings = [f for f in result.findings if not f.false_positive]
+ real_findings.sort(key=lambda f: (severity_order.get(f.severity, 99), f.title))
- lines.append(f" Total Findings: {total}")
- lines.append("")
+ top_issues = []
+ seen_titles = set()
+ for f in real_findings:
+ if f.title not in seen_titles:
+ seen_titles.add(f.title)
+ top_issues.append(f"{f.severity.value.upper()}: {f.title}")
+ if len(top_issues) == 5:
+ break
- # Severity breakdown
- severity_order = ["critical", "high", "medium", "low", "info"]
- severity_icons = {
- "critical": "🔴",
- "high": "🟠",
- "medium": "🟡",
- "low": "🔵",
- "info": "⚪",
- }
+ return StreamScore(
+ stream=result.stream,
+ score=score_val,
+ findings_by_severity=counts,
+ top_issues=top_issues,
+ )
- for sev in severity_order:
- count = result.stats.get(sev, 0)
- if count > 0:
- icon = severity_icons.get(sev, " ")
- lines.append(f" {icon} {sev.upper():10s} {count}")
- lines.append("")
+def generate_json_report(result: ScanResult) -> dict:
+ """JSON report for dashboard consumption."""
+ scorecard = calculate_stream_score(result)
+
+ # Calculate stats
+ stats = {}
+ for f in result.findings:
+ if f.false_positive:
+ continue
+ stats[f.category.value] = stats.get(f.category.value, 0) + 1
+ stats[f.severity.value] = stats.get(f.severity.value, 0) + 1
+
+ result.stats = stats
- # Security posture
- critical = result.stats.get("critical", 0)
- high = result.stats.get("high", 0)
- if critical > 0:
- lines.append(" ⛔ Security Posture: CRITICAL — Immediate remediation required.")
- elif high > 0:
- lines.append(" ⚠️ Security Posture: AT RISK — High-severity issues must be addressed.")
+ if hasattr(result, "model_dump_json"):
+ serialized = json.loads(result.model_dump_json())
else:
- lines.append(" 📋 Security Posture: MODERATE — Review findings before production.")
+ serialized = json.loads(result.json())
+
+ scorecard_dict = scorecard.model_dump() if hasattr(scorecard, "model_dump") else scorecard.dict()
+ serialized["scorecard"] = scorecard_dict
+ return serialized
+
+
+REMEDIATIONS_GUIDES = {
+ Category.SECRET: {
+ "guideline": "API keys, passwords, and sensitive tokens should never be hardcoded in source files or committed to Git. Instead, load them from environment variables or a secret manager.",
+ "bad": 'API_KEY = "sk-ant-api03-exampleKeyValHere1234567890"',
+ "good": 'import os\nAPI_KEY = os.environ.get("API_KEY")'
+ },
+ Category.DEPENDENCY: {
+ "guideline": "Vulnerable dependencies expose applications to known exploits. Floating/unpinned dependencies risk supply chain compromise. Pin versions and use cryptographic hashes.",
+ "bad": 'requests>=2.25.0',
+ "good": 'requests==2.31.0 --hash=sha256:7486c32d... # or use lockfiles'
+ },
+ Category.CODE_PATTERN: {
+ "guideline": "Avoid dynamic SQL query building and direct innerHTML assignments. Use parameterized queries or secure DOM APIs.",
+ "bad": 'cursor.execute(f"SELECT * FROM users WHERE id = {user_id}")\n# Or in frontend:\nelement.innerHTML = user_input',
+ "good": 'cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))\n# Or in frontend:\nelement.textContent = user_input'
+ },
+ Category.AUTH: {
+ "guideline": "Endpoints that handle user data must enforce access control via authentication dependencies or Supabase row-level security (RLS) policies.",
+ "bad": '@app.get("/data")\ndef get_data(): ...',
+ "good": '@app.get("/data")\ndef get_data(user: User = Depends(get_current_user)): ...\n# SQL RLS:\nALTER TABLE profiles ENABLE ROW LEVEL SECURITY;'
+ }
+}
+
+def generate_terminal_report(result: ScanResult) -> str:
+ """Rich terminal output of scan results."""
+ scorecard = calculate_stream_score(result)
+
+ # Calculate stats
+ stats = {
+ "critical": 0,
+ "high": 0,
+ "medium": 0,
+ "low": 0,
+ "info": 0,
+ }
+ for f in result.findings:
+ if f.false_positive:
+ continue
+ stats[f.severity.value] = stats.get(f.severity.value, 0) + 1
+
+ lines = []
+ lines.append("=" * 80)
+ lines.append(f" ZEROCLAW SECURITY REPORT — {result.stream.upper()}")
+ lines.append(f" Scanned At: {result.scanned_at.strftime('%Y-%m-%d %H:%M:%S')}")
+ lines.append("=" * 80)
+ lines.append("")
+
+ # Score details
+ lines.append(f" GLASS Security Score: {scorecard.score}/10.0")
+ lines.append(f" Findings by Severity: CRITICAL: {stats['critical']} | HIGH: {stats['high']} | MEDIUM: {stats['medium']} | LOW: {stats['low']} | INFO: {stats['info']}")
lines.append("")
- # ── Detailed Findings ───────────────────────────────────────────────
- lines.append("─" * 72)
- lines.append(" DETAILED FINDINGS")
- lines.append("─" * 72)
+ if not result.findings:
+ lines.append(" ✅ No security findings detected! The repository is clean.")
+ lines.append("-" * 80)
+ return "\n".join(lines)
- for i, finding in enumerate(result.findings, 1):
- sev = finding.severity.value.upper()
- icon = severity_icons.get(finding.severity.value, " ")
+ # Group findings by category
+ grouped: dict[Category, list[Finding]] = defaultdict(list)
+ for f in result.findings:
+ grouped[f.category].append(f)
- lines.append("")
- lines.append(f" {icon} [{i}/{total}] {finding.id}")
- lines.append(f" {'─' * 60}")
- lines.append(f" Severity: {sev}")
- lines.append(f" Category: {finding.category.value}")
- lines.append(f" Title: {finding.title}")
- lines.append(f" File: {finding.file_path}")
- if finding.line_number is not None:
- lines.append(f" Line: {finding.line_number}")
- lines.append("")
- lines.append(f" Description:")
- for desc_line in finding.description.split("\n"):
- lines.append(f" {desc_line}")
- lines.append("")
- lines.append(f" Remediation:")
- for rem_line in finding.remediation.split("\n"):
- lines.append(f" {rem_line}")
+ # Print Findings grouped by Category
+ lines.append("DETAILED FINDINGS")
+ lines.append("-" * 80)
- # ── ZeroClaw Enrichment (if available) ──────────────────────────
- if finding.reasoning_chain:
+ for cat in Category:
+ cat_findings = grouped.get(cat, [])
+ if not cat_findings:
+ continue
+
+ lines.append(f"## {cat.value.upper()} SCANNER — {len(cat_findings)} findings")
+ lines.append("")
+
+ for idx, f in enumerate(cat_findings, start=1):
+ fp_flag = " [FALSE POSITIVE]" if f.false_positive else ""
+ lines.append(f" {idx}. [{f.severity.value.upper()}]{fp_flag} {f.title}")
+ lines.append(f" File: {f.file_path}:{f.line_number or 'N/A'}")
+ lines.append(f" ID: {f.id}")
+ lines.append(f" Description: {f.description}")
+ lines.append(f" Remediation: {f.remediation}")
+
+ # Print ZeroClaw Reasoning and Fixed Code if present
+ if getattr(f, "reasoning_chain", None):
+ lines.append("")
+ lines.append(" 🧠 ZeroClaw Reasoning:")
+ for chain_line in f.reasoning_chain.splitlines():
+ lines.append(f" {chain_line}")
+ if getattr(f, "fixed_code", None):
+ lines.append("")
+ lines.append(" 🔧 ZeroClaw Fixed Code:")
+ lines.append(" ┌" + "─" * 56 + "┐")
+ for code_line in f.fixed_code.splitlines():
+ lines.append(f" │ {code_line}")
+ lines.append(" └" + "─" * 56 + "┘")
lines.append("")
- lines.append(f" 🧠 ZeroClaw Reasoning:")
- for chain_line in finding.reasoning_chain.split("\n"):
- lines.append(f" {chain_line}")
- if finding.fixed_code:
+ # Add remediation guidance + code examples
+ guide = REMEDIATIONS_GUIDES.get(cat)
+ if guide:
+ lines.append(" Remediation Guidance:")
+ lines.append(f" {guide['guideline']}")
+ lines.append(" [BAD EXAMPLES]")
+ for bad_line in guide["bad"].splitlines():
+ lines.append(f" - {bad_line}")
+ lines.append(" [GOOD EXAMPLES]")
+ for good_line in guide["good"].splitlines():
+ lines.append(f" + {good_line}")
lines.append("")
- lines.append(f" 🔧 ZeroClaw Fixed Code:")
- lines.append(f" ┌{'─' * 56}┐")
- for code_line in finding.fixed_code.split("\n"):
- lines.append(f" │ {code_line}")
- lines.append(f" └{'─' * 56}┘")
-
- lines.append("")
-
- # ── Footer ──────────────────────────────────────────────────────────
- enriched_count = sum(1 for f in result.findings if f.reasoning_chain is not None)
- lines.append("=" * 72)
- lines.append(f" {total} findings | {enriched_count} AI-enriched | powered by ZeroClaw")
- lines.append("=" * 72)
- lines.append("")
+ lines.append("-" * 80)
return "\n".join(lines)
-
-def generate_json_report(result: ScanResult) -> dict:
- """JSON report formatted precisely to the LifeAtlasEcosystemSecurityFindingSchema."""
-
- findings_array = []
- for f in result.findings:
- # Default stride based on category
- stride = "Tampering"
- if f.category.value == "auth":
- stride = "Elevation of Privilege"
- elif f.category.value == "secret":
- stride = "Information Disclosure"
- elif f.category.value == "dependency":
- stride = "Tampering"
- elif f.category.value == "injection":
- stride = "Tampering"
-
- steps = f.remediation
- if f.fixed_code:
- steps += f"\n\nFixed Code:\n```\n{f.fixed_code}\n```"
-
- findings_array.append({
- "id": f.id,
- "reasoning_chain": f.reasoning_chain or "Static analysis identified the vulnerability; AI enrichment skipped or unavailable.",
- "severity": f.severity.value.upper(),
- "stride_classification": stride,
- "owasp_alignment": "LA-01",
- "affected_component": f.file_path,
- "description": f.description,
- "remediation": {
- "steps": steps
- }
- })
-
- stream_id = 1
- try:
- import re
- if result.stream:
- match = re.search(r'\d+', result.stream)
- if match:
- stream_id = int(match.group(0))
- except Exception:
- pass
-
- # Extract base name from repo_url or target path
- import os
- repo_name = os.path.basename(os.path.normpath(result.repo_url)) or "unknown-repo"
-
- return {
- "scan_metadata": {
- "timestamp": result.scanned_at.isoformat(),
- "scanner_tool": "custom-regex",
- "execution_environment": "local-dev-env"
- },
- "target_scope": {
- "stream_id": stream_id,
- "repository_name": repo_name,
- "commit_sha": "0000000000000000000000000000000000000000"
- },
- "summary": {
- "total_findings": len(result.findings),
- "critical_count": result.stats.get("critical", 0),
- "high_count": result.stats.get("high", 0),
- "medium_count": result.stats.get("medium", 0),
- "low_count": result.stats.get("low", 0)
- },
- "findings": findings_array
- }
-
-
-def calculate_stream_score(result: ScanResult) -> StreamScore:
- """Calculate 0-10 security score for a stream."""
- # Severity weights for scoring
- weights = {
- "critical": 10.0,
- "high": 5.0,
- "medium": 2.0,
- "low": 0.5,
- "info": 0.0,
- }
-
- total_penalty = 0.0
- findings_by_severity: dict[str, int] = {}
-
- for finding in result.findings:
- sev = finding.severity.value
- findings_by_severity[sev] = findings_by_severity.get(sev, 0) + 1
- total_penalty += weights.get(sev, 0)
-
- # Score: 10.0 (perfect) minus penalties, floored at 0.0
- raw_score = max(0.0, 10.0 - total_penalty)
- score = round(raw_score, 1)
-
- # Top issues: up to 5 highest-severity finding titles
- sorted_findings = sorted(
- result.findings,
- key=lambda f: list(weights.keys()).index(f.severity.value)
- if f.severity.value in weights
- else 999,
- )
- top_issues = [f.title for f in sorted_findings[:5]]
-
- return StreamScore(
- stream=result.stream,
- score=score,
- findings_by_severity=findings_by_severity,
- top_issues=top_issues,
- )
diff --git a/src/zeroclaw/scanners/api_security_tester.py b/src/zeroclaw/scanners/api_security_tester.py
new file mode 100644
index 0000000..aef78d0
--- /dev/null
+++ b/src/zeroclaw/scanners/api_security_tester.py
@@ -0,0 +1,447 @@
+"""API Security Tester for dynamic analysis of auth bypass, rate limiting, and session security."""
+from __future__ import annotations
+
+import asyncio
+import concurrent.futures
+import logging
+from typing import Any
+import httpx
+from fastapi import FastAPI
+
+logger = logging.getLogger(__name__)
+
+
+class SecurityTestError(AssertionError):
+ """Raised when a security test fails."""
+ pass
+
+
+def run_async(coro) -> Any:
+ """Run an async coroutine synchronously, handling any active event loops."""
+ try:
+ return asyncio.run(coro)
+ except RuntimeError:
+ with concurrent.futures.ThreadPoolExecutor() as executor:
+ future = executor.submit(asyncio.run, coro)
+ return future.result()
+
+
+class APISecurityTester:
+ """Utility class to dynamically test FastAPI applications for security vulnerabilities."""
+
+ @staticmethod
+ def test_auth_bypass(app: FastAPI, protected_routes: list[dict[str, Any]]) -> None:
+ """
+ Verify that protected endpoints reject unauthorized requests.
+
+ For each route in protected_routes:
+ - Hits the route without an auth token -> expects 401 or 403.
+ - Hits the route with an expired token -> expects 401 or 403.
+ - Hits the route with an incorrect user role -> expects 403.
+
+ Each route dict can specify:
+ - 'path' (str): the target endpoint path (required)
+ - 'method' (str): HTTP method, e.g., 'GET', 'POST' (default 'GET')
+ - 'params' (dict): query parameters (optional)
+ - 'json' (dict): JSON body (optional)
+ - 'headers' (dict): normal/base headers (optional)
+ - 'cookies' (dict): normal/base cookies (optional)
+ - 'expired_headers' (dict): headers for expired token test (optional)
+ - 'expired_cookies' (dict): cookies for expired token test (optional)
+ - 'wrong_role_headers' (dict): headers for wrong role test (optional)
+ - 'wrong_role_cookies' (dict): cookies for wrong role test (optional)
+ """
+ async def run():
+ failures = []
+ transport = httpx.ASGITransport(app=app)
+
+ async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
+ for idx, route in enumerate(protected_routes):
+ path = route.get("path")
+ if not path:
+ failures.append(f"Route at index {idx} is missing 'path'.")
+ continue
+
+ method = route.get("method", "GET").upper()
+ params = route.get("params")
+ json_body = route.get("json")
+ base_headers = route.get("headers") or {}
+ base_cookies = route.get("cookies") or {}
+
+ # 1. No Auth Check
+ # Remove any authorization headers or session cookies
+ no_auth_headers = {k: v for k, v in base_headers.items() if k.lower() != "authorization"}
+ no_auth_cookies = {k: v for k, v in base_cookies.items() if "session" not in k.lower()}
+
+ client.cookies.clear()
+ client.cookies.update(no_auth_cookies)
+
+ try:
+ response = await client.request(
+ method=method,
+ url=path,
+ params=params,
+ json=json_body,
+ headers=no_auth_headers,
+ )
+ if response.status_code not in (401, 403):
+ failures.append(
+ f"Auth Bypass Check Failed: Endpoint {method} {path} with no auth headers "
+ f"returned {response.status_code} instead of 401/403."
+ )
+ except Exception as e:
+ failures.append(f"Auth Bypass Check Error on {method} {path} (No Auth): {e}")
+
+ # 2. Expired Token Check
+ expired_headers = route.get("expired_headers")
+ expired_cookies = route.get("expired_cookies")
+
+ # Fallback to dummy values if not provided
+ if not expired_headers and not expired_cookies:
+ expired_headers = {"Authorization": "Bearer expired_token_xyz"}
+
+ req_headers = {**base_headers, **(expired_headers or {})}
+ req_cookies = {**base_cookies, **(expired_cookies or {})}
+
+ client.cookies.clear()
+ client.cookies.update(req_cookies)
+
+ try:
+ response = await client.request(
+ method=method,
+ url=path,
+ params=params,
+ json=json_body,
+ headers=req_headers,
+ )
+ if response.status_code not in (401, 403):
+ failures.append(
+ f"Auth Bypass Check Failed: Endpoint {method} {path} with expired credentials "
+ f"returned {response.status_code} instead of 401/403."
+ )
+ except Exception as e:
+ failures.append(f"Auth Bypass Check Error on {method} {path} (Expired Auth): {e}")
+
+ # 3. Wrong Role Check
+ wrong_role_headers = route.get("wrong_role_headers")
+ wrong_role_cookies = route.get("wrong_role_cookies")
+
+ # Fallback to dummy values if not provided
+ if not wrong_role_headers and not wrong_role_cookies:
+ wrong_role_headers = {"Authorization": "Bearer wrong_role_token_xyz"}
+
+ req_headers = {**base_headers, **(wrong_role_headers or {})}
+ req_cookies = {**base_cookies, **(wrong_role_cookies or {})}
+
+ client.cookies.clear()
+ client.cookies.update(req_cookies)
+
+ try:
+ response = await client.request(
+ method=method,
+ url=path,
+ params=params,
+ json=json_body,
+ headers=req_headers,
+ )
+ # Wrong role should result in a 403 Forbidden
+ if response.status_code != 403:
+ failures.append(
+ f"Auth Bypass Check Failed: Endpoint {method} {path} with incorrect role "
+ f"returned {response.status_code} instead of 403."
+ )
+ except Exception as e:
+ failures.append(f"Auth Bypass Check Error on {method} {path} (Wrong Role): {e}")
+
+ if failures:
+ raise SecurityTestError("\n".join(failures))
+
+ run_async(run())
+
+ @staticmethod
+ def test_rate_limiting(app: FastAPI, rate_limited_routes: list[dict[str, Any]]) -> None:
+ """
+ Verify that rate-limited routes enforce limits and return 429 Too Many Requests.
+
+ For each route in rate_limited_routes:
+ - Sends requests sequentially exceeding the expected limit.
+ - Asserts that we eventually receive a 429 status code.
+
+ Each route dict can specify:
+ - 'path' (str): the target endpoint path (required)
+ - 'method' (str): HTTP method, e.g., 'GET', 'POST' (default 'GET')
+ - 'limit' (int): the configured limit before hitting 429 (default 5)
+ - 'params' (dict): query parameters (optional)
+ - 'json' (dict): JSON body (optional)
+ - 'headers' (dict): request headers (optional)
+ - 'cookies' (dict): request cookies (optional)
+ """
+ async def run():
+ failures = []
+ transport = httpx.ASGITransport(app=app)
+
+ async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
+ for idx, route in enumerate(rate_limited_routes):
+ path = route.get("path")
+ if not path:
+ failures.append(f"Route at index {idx} is missing 'path'.")
+ continue
+
+ method = route.get("method", "GET").upper()
+ limit = route.get("limit", 5)
+ params = route.get("params")
+ json_body = route.get("json")
+ headers = route.get("headers")
+ cookies = route.get("cookies") or {}
+
+ client.cookies.clear()
+ client.cookies.update(cookies)
+
+ hit_429 = False
+ status_codes = []
+
+ # We attempt up to limit + 10 requests to trigger rate limiting
+ max_attempts = limit + 10
+ for i in range(max_attempts):
+ try:
+ response = await client.request(
+ method=method,
+ url=path,
+ params=params,
+ json=json_body,
+ headers=headers,
+ )
+ status_codes.append(response.status_code)
+ if response.status_code == 429:
+ hit_429 = True
+ break
+ except Exception as e:
+ failures.append(f"Rate Limiting Check Error on {method} {path} at request {i+1}: {e}")
+ break
+
+ if not hit_429:
+ failures.append(
+ f"Rate Limiting Check Failed: Endpoint {method} {path} did not return 429 after {max_attempts} "
+ f"requests. Response status codes: {status_codes}"
+ )
+
+ if failures:
+ raise SecurityTestError("\n".join(failures))
+
+ run_async(run())
+
+ @staticmethod
+ def test_session_security(app: FastAPI, session_routes: list[dict[str, Any]]) -> None:
+ """
+ Verify session fixation, session hijacking (signature tampering), and session timeout.
+
+ Each item in session_routes can specify:
+ - 'login_path' (str): login endpoint to test session fixation
+ - 'login_method' (str): HTTP method for login (default 'POST')
+ - 'login_data' (dict): credentials payload for login (optional)
+ - 'session_cookie_name' (str): name of cookie used for sessions (optional)
+ - 'session_header_name' (str): name of header or JSON response key for session tokens (optional)
+ - 'protected_path' (str): a protected path to verify hijacking and timeouts (required for hijacking/timeout)
+ - 'protected_method' (str): method for protected path (default 'GET')
+ - 'valid_headers' (dict): valid session headers if login bypass is preferred (optional)
+ - 'valid_cookies' (dict): valid session cookies if login bypass is preferred (optional)
+ - 'expired_headers' (dict): expired session headers (optional)
+ - 'expired_cookies' (dict): expired session cookies (optional)
+ """
+ async def run():
+ failures = []
+ transport = httpx.ASGITransport(app=app)
+
+ for idx, route in enumerate(session_routes):
+ cookie_name = route.get("session_cookie_name")
+ header_name = route.get("session_header_name")
+ if not cookie_name and not header_name:
+ cookie_name = "session"
+
+ # 1. Session Fixation Check
+ login_path = route.get("login_path")
+ if login_path:
+ login_method = route.get("login_method", "POST").upper()
+ login_data = route.get("login_data")
+
+ async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
+ # Set pre-existing session token/cookie
+ dummy_token = "pre_existing_session_id_123456"
+ if isinstance(cookie_name, str):
+ client.cookies[cookie_name] = dummy_token
+
+ req_headers = {}
+ if header_name:
+ req_headers[header_name] = f"Bearer {dummy_token}"
+
+ try:
+ response = await client.request(
+ method=login_method,
+ url=login_path,
+ json=login_data,
+ headers=req_headers,
+ )
+
+ # Verify fixation protection without ValueError on duplicates
+ if isinstance(cookie_name, str):
+ new_cookie_val = response.cookies.get(cookie_name)
+ if not new_cookie_val:
+ # Inspect client jar safely
+ matching_vals = [c.value for c in client.cookies.jar if c.name == cookie_name]
+ if len(matching_vals) == 1 and matching_vals[0] == dummy_token:
+ failures.append(
+ f"Session Fixation Vulnerability: Login at {login_path} did not set a new "
+ f"session cookie '{cookie_name}'. Value remained the pre-existing '{dummy_token}'."
+ )
+ else:
+ if new_cookie_val == dummy_token:
+ failures.append(
+ f"Session Fixation Vulnerability: Login at {login_path} set session cookie "
+ f"'{cookie_name}' but kept the pre-existing value '{dummy_token}'."
+ )
+
+ if header_name:
+ new_header_val = response.headers.get(header_name)
+ try:
+ res_json = response.json()
+ new_json_val = res_json.get(header_name) or res_json.get("access_token") or res_json.get("token")
+ except Exception:
+ new_json_val = None
+
+ new_val = new_header_val or new_json_val
+ if new_val == dummy_token:
+ failures.append(
+ f"Session Fixation Vulnerability: Login at {login_path} did not rotate the "
+ f"session header/token '{header_name}'. Value remained '{dummy_token}'."
+ )
+
+ except Exception as e:
+ failures.append(f"Session Fixation Check Error at {login_path}: {e}")
+
+ # 2. Session Hijacking & Tampering / Removal
+ protected_path = route.get("protected_path")
+ if protected_path:
+ protected_method = route.get("protected_method", "GET").upper()
+ valid_headers = route.get("valid_headers") or {}
+ valid_cookies = route.get("valid_cookies") or {}
+
+ # If login_path is specified and we don't have valid headers/cookies, log in dynamically
+ if not valid_headers and not valid_cookies and login_path:
+ async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
+ try:
+ login_method = route.get("login_method", "POST").upper()
+ login_data = route.get("login_data")
+ response = await client.request(method=login_method, url=login_path, json=login_data)
+ if response.status_code in (200, 201):
+ valid_cookies = dict(client.cookies)
+ try:
+ res_json = response.json()
+ token = res_json.get("access_token") or res_json.get("token")
+ if token:
+ valid_headers = {"Authorization": f"Bearer {token}"}
+ else:
+ cookie_val = response.cookies.get(cookie_name or "session")
+ if cookie_val:
+ valid_cookies = {cookie_name or "session": cookie_val}
+ except Exception:
+ pass
+ except Exception as e:
+ failures.append(f"Login setup for hijacking check failed: {e}")
+
+ # Establish baseline request works with valid credentials
+ async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
+ client.cookies.clear()
+ client.cookies.update(valid_cookies)
+ try:
+ res_ok = await client.request(
+ method=protected_method,
+ url=protected_path,
+ headers=valid_headers,
+ )
+ if res_ok.status_code not in (200, 201, 204):
+ logger.warning(
+ "Baseline request for hijacking test on %s %s returned %s instead of success. "
+ "Will proceed with tampering checks.",
+ protected_method, protected_path, res_ok.status_code
+ )
+ except Exception as e:
+ logger.warning("Baseline request for hijacking test error: %s", e)
+
+ # Tamper test 1: Removal of credentials
+ client.cookies.clear()
+ try:
+ res_removed = await client.request(
+ method=protected_method,
+ url=protected_path,
+ )
+ if res_removed.status_code not in (401, 403):
+ failures.append(
+ f"Session Hijacking / Credentials Removal Vulnerability: Protected route "
+ f"{protected_method} {protected_path} accessed without credentials returned "
+ f"{res_removed.status_code} instead of 401/403."
+ )
+ except Exception as e:
+ failures.append(f"Session Hijacking Removal Check Error: {e}")
+
+ # Tamper test 2: Modified/Corrupted credentials
+ tampered_headers = {}
+ for k, v in valid_headers.items():
+ if k.lower() == "authorization":
+ if "bearer " in v.lower():
+ tampered_headers[k] = v + "invalidsignature"
+ else:
+ tampered_headers[k] = v + "_tampered"
+ else:
+ tampered_headers[k] = v + "_altered"
+
+ tampered_cookies = {k: v + "_tampered" for k, v in valid_cookies.items()}
+
+ client.cookies.clear()
+ client.cookies.update(tampered_cookies)
+ if tampered_headers or tampered_cookies:
+ try:
+ res_tampered = await client.request(
+ method=protected_method,
+ url=protected_path,
+ headers=tampered_headers,
+ )
+ if res_tampered.status_code not in (401, 403):
+ failures.append(
+ f"Session Hijacking / Signature Tampering Vulnerability: Protected route "
+ f"{protected_method} {protected_path} accessed with tampered credentials "
+ f"returned {res_tampered.status_code} instead of 401/403."
+ )
+ except Exception as e:
+ failures.append(f"Session Hijacking Tampering Check Error: {e}")
+
+ # 3. Session Timeout Check
+ if protected_path:
+ expired_headers = route.get("expired_headers")
+ expired_cookies = route.get("expired_cookies")
+
+ if not expired_headers and not expired_cookies:
+ expired_headers = {"Authorization": "Bearer expired_session_token_123"}
+ expired_cookies = {"session": "expired_session_cookie_123"}
+
+ async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
+ client.cookies.clear()
+ client.cookies.update(expired_cookies or {})
+ try:
+ res_timeout = await client.request(
+ method=protected_method,
+ url=protected_path,
+ headers=expired_headers,
+ )
+ if res_timeout.status_code not in (401, 403):
+ failures.append(
+ f"Session Timeout / Expiration Vulnerability: Protected route "
+ f"{protected_method} {protected_path} accessed with expired credentials "
+ f"returned {res_timeout.status_code} instead of 401/403."
+ )
+ except Exception as e:
+ failures.append(f"Session Timeout Check Error: {e}")
+
+ if failures:
+ raise SecurityTestError("\n".join(failures))
+
+ run_async(run())
diff --git a/src/zeroclaw/scanners/dependency_scanner.py b/src/zeroclaw/scanners/dependency_scanner.py
index cf64792..362b573 100644
--- a/src/zeroclaw/scanners/dependency_scanner.py
+++ b/src/zeroclaw/scanners/dependency_scanner.py
@@ -1,4 +1,4 @@
-"""Dependency vulnerability scanner: npm audit + pip-audit + CVE lookup."""
+import os
import json
import re
import subprocess
diff --git a/tests/test_api_security_tester.py b/tests/test_api_security_tester.py
new file mode 100644
index 0000000..262a87a
--- /dev/null
+++ b/tests/test_api_security_tester.py
@@ -0,0 +1,194 @@
+import pytest
+from fastapi import FastAPI, Depends, HTTPException, status, Response, Request
+from fastapi.responses import JSONResponse
+from zeroclaw.scanners.api_security_tester import APISecurityTester, SecurityTestError
+
+# ----------------- Helper Mock Apps -----------------
+
+def create_secure_app() -> FastAPI:
+ app = FastAPI()
+
+ # Simple auth dependency
+ def get_user(request: Request):
+ auth = request.headers.get("Authorization")
+ if not auth:
+ raise HTTPException(status_code=401, detail="Missing token")
+ if "expired" in auth:
+ raise HTTPException(status_code=401, detail="Token expired")
+ if "wrong_role" in auth:
+ raise HTTPException(status_code=403, detail="Forbidden: wrong role")
+ return {"user": "alice"}
+
+ @app.get("/protected")
+ def protected_route(user: dict = Depends(get_user)):
+ return {"status": "success", "user": user}
+
+ # Rate limiting mock (state stored in app state for simplicity in tests)
+ app.state.request_count = 0
+
+ @app.get("/rate-limited")
+ def rate_limited_route():
+ app.state.request_count += 1
+ if app.state.request_count > 5:
+ return JSONResponse(status_code=429, content={"detail": "Too many requests"})
+ return {"status": "ok"}
+
+ # Login and session management mocks
+ @app.post("/login")
+ def login(request: Request, response: Response):
+ # Session fixation defense: always rotate session ID
+ response.set_cookie(key="session", value="new_secure_session_id_789")
+ return {"status": "logged_in"}
+
+ @app.get("/session-protected")
+ def session_protected(request: Request):
+ cookie = request.cookies.get("session")
+ if not cookie:
+ raise HTTPException(status_code=401, detail="No session")
+ if "tampered" in cookie or "expired" in cookie:
+ raise HTTPException(status_code=401, detail="Invalid session signature or expired")
+ return {"status": "ok"}
+
+ return app
+
+
+def create_vulnerable_app() -> FastAPI:
+ app = FastAPI()
+
+ # Unprotected route (vulnerable to auth bypass)
+ @app.get("/protected")
+ def protected_route():
+ return {"status": "success", "bypass": True}
+
+ # Endpoint with no rate limiting (vulnerable)
+ @app.get("/rate-limited")
+ def rate_limited_route():
+ return {"status": "ok"}
+
+ # Login route vulnerable to session fixation
+ @app.post("/login")
+ def login(request: Request, response: Response):
+ # Does NOT rotate cookie if pre-existing
+ old_session = request.cookies.get("session")
+ if old_session:
+ response.set_cookie(key="session", value=old_session)
+ return {"status": "logged_in", "rotated": False}
+ response.set_cookie(key="session", value="new_session")
+ return {"status": "logged_in", "rotated": True}
+
+ # Protected route vulnerable to hijacking (accepts tampered sessions)
+ @app.get("/session-protected")
+ def session_protected(request: Request):
+ cookie = request.cookies.get("session")
+ # Vuln 1: doesn't require session at all, or accepts tampered/expired ones
+ if cookie and "tampered" in cookie:
+ # Accepts tampered cookie!
+ return {"status": "ok", "tampered_accepted": True}
+ if not cookie:
+ # Accepts missing cookie as well!
+ return {"status": "ok", "no_session_accepted": True}
+ return {"status": "ok"}
+
+ return app
+
+
+# ----------------- Tests for APISecurityTester -----------------
+
+class TestAPISecurityTesterAuthBypass:
+ def test_auth_bypass_passes_on_secure_app(self):
+ app = create_secure_app()
+ protected_routes = [
+ {
+ "path": "/protected",
+ "method": "GET",
+ "headers": {"Authorization": "Bearer valid_token"},
+ "expired_headers": {"Authorization": "Bearer expired_token"},
+ "wrong_role_headers": {"Authorization": "Bearer wrong_role_token"},
+ }
+ ]
+ # Should execute without raising SecurityTestError
+ APISecurityTester.test_auth_bypass(app, protected_routes)
+
+ def test_auth_bypass_fails_on_vulnerable_app(self):
+ app = create_vulnerable_app()
+ protected_routes = [
+ {
+ "path": "/protected",
+ "method": "GET",
+ "headers": {"Authorization": "Bearer valid_token"},
+ "expired_headers": {"Authorization": "Bearer expired_token"},
+ "wrong_role_headers": {"Authorization": "Bearer wrong_role_token"},
+ }
+ ]
+ with pytest.raises(SecurityTestError) as exc_info:
+ APISecurityTester.test_auth_bypass(app, protected_routes)
+
+ assert "Auth Bypass Check Failed" in str(exc_info.value)
+ assert "/protected" in str(exc_info.value)
+
+
+class TestAPISecurityTesterRateLimiting:
+ def test_rate_limiting_passes_on_secure_app(self):
+ app = create_secure_app()
+ rate_limited_routes = [
+ {
+ "path": "/rate-limited",
+ "method": "GET",
+ "limit": 5,
+ }
+ ]
+ # Should pass because secure app returns 429 on 6th request
+ APISecurityTester.test_rate_limiting(app, rate_limited_routes)
+
+ def test_rate_limiting_fails_on_vulnerable_app(self):
+ app = create_vulnerable_app()
+ rate_limited_routes = [
+ {
+ "path": "/rate-limited",
+ "method": "GET",
+ "limit": 5,
+ }
+ ]
+ with pytest.raises(SecurityTestError) as exc_info:
+ APISecurityTester.test_rate_limiting(app, rate_limited_routes)
+
+ assert "Rate Limiting Check Failed" in str(exc_info.value)
+ assert "/rate-limited" in str(exc_info.value)
+
+
+class TestAPISecurityTesterSessionSecurity:
+ def test_session_security_passes_on_secure_app(self):
+ app = create_secure_app()
+ session_routes = [
+ {
+ "login_path": "/login",
+ "login_method": "POST",
+ "session_cookie_name": "session",
+ "protected_path": "/session-protected",
+ "protected_method": "GET",
+ "valid_cookies": {"session": "valid_session_123"},
+ "expired_cookies": {"session": "expired_session_123"},
+ }
+ ]
+ # Should pass cleanly
+ APISecurityTester.test_session_security(app, session_routes)
+
+ def test_session_security_fails_on_vulnerable_app(self):
+ app = create_vulnerable_app()
+ session_routes = [
+ {
+ "login_path": "/login",
+ "login_method": "POST",
+ "session_cookie_name": "session",
+ "protected_path": "/session-protected",
+ "protected_method": "GET",
+ "valid_cookies": {"session": "valid_session_123"},
+ "expired_cookies": {"session": "expired_session_123"},
+ }
+ ]
+ with pytest.raises(SecurityTestError) as exc_info:
+ APISecurityTester.test_session_security(app, session_routes)
+
+ err_msg = str(exc_info.value)
+ # Verify fixation vulnerability, hijacking vulnerability, or timeout vulnerability is flagged
+ assert "Session Fixation" in err_msg or "Session Hijacking" in err_msg or "Session Timeout" in err_msg
diff --git a/tests/test_auth_scanner.py b/tests/test_auth_scanner.py
new file mode 100644
index 0000000..7dacee1
--- /dev/null
+++ b/tests/test_auth_scanner.py
@@ -0,0 +1,179 @@
+from pathlib import Path
+import pytest
+from zeroclaw.scanners.auth_scanner import scan_fastapi_auth, scan_supabase_rls
+from zeroclaw.models import Category, Severity
+
+
+class TestAuthScanner:
+ def test_detects_unprotected_endpoints(self, tmp_path):
+ """Should detect endpoints that do not have Depends or Security parameters."""
+ file = tmp_path / "routes.py"
+ file.write_text('''
+@router.post("/auth/login")
+async def auth_login():
+ return {"status": "ok"}
+
+@app.get("/users/{user_id}")
+def get_user(user_id: str):
+ return {"user_id": user_id}
+''')
+ findings = scan_fastapi_auth(tmp_path)
+ assert len(findings) == 2
+
+ # Verify first finding
+ assert findings[0].category == Category.AUTH
+ assert findings[0].severity == Severity.HIGH
+ assert findings[0].line_number == 2
+ assert "auth_login" in findings[0].description
+
+ # Verify second finding
+ assert findings[1].line_number == 6
+ assert "get_user" in findings[1].description
+
+ def test_ignores_protected_endpoints(self, tmp_path):
+ """Should ignore routes that contain Depends or Security in their signature or decorator."""
+ file = tmp_path / "routes.py"
+ file.write_text('''
+@router.post("/auth/logout")
+async def auth_logout(admin: AdminContext = Depends(_admin_dep)):
+ return {"status": "ok"}
+
+@router.get("/users", dependencies=[Depends(_admin_dep)])
+async def users_list():
+ return []
+
+@app.get("/secure")
+def secure_route(api_key: str = Security(api_key_header)):
+ return {"status": "secure"}
+''')
+ findings = scan_fastapi_auth(tmp_path)
+ assert len(findings) == 0
+
+ def test_unreadable_file_logged(self, tmp_path, caplog):
+ """Should log warning when file cannot be read due to OS error."""
+ import logging
+ from unittest.mock import patch
+
+ bad_file = tmp_path / "unreadable.py"
+ bad_file.write_text("@app.get('/unprotected')\ndef bad_route(): pass")
+
+ with patch("os.open", side_effect=PermissionError("Mocked permission error")):
+ with caplog.at_level(logging.WARNING):
+ findings = scan_fastapi_auth(tmp_path)
+ assert len(findings) == 0
+
+ warnings = [rec.message for rec in caplog.records if rec.levelno == logging.WARNING]
+ assert any("Could not read file" in w and "Mocked permission error" in w for w in warnings)
+
+ def test_large_file_skipped(self, tmp_path):
+ """Should skip scanning files that exceed the 5MB size limit."""
+ large_file = tmp_path / "large.py"
+ large_file.write_text("@app.get('/unprotected')\ndef bad(): pass" + (" " * 5 * 1024 * 1024))
+ findings = scan_fastapi_auth(tmp_path)
+ assert len(findings) == 0
+
+ def test_path_traversal_prevention(self, tmp_path):
+ """Should prevent scanning files resolving outside the target boundary."""
+ from unittest.mock import MagicMock, patch
+ from pathlib import Path
+
+ file_mock = MagicMock(spec=Path)
+ file_mock.is_symlink.return_value = False
+ file_mock.is_file.return_value = True
+ file_mock.suffix = ".py"
+ file_mock.resolve.return_value = Path("C:/Windows") if Path("C:/").exists() else Path("/etc")
+ file_mock.stat.return_value.st_size = 100
+
+ with patch.object(Path, "rglob", return_value=[file_mock]):
+ findings = scan_fastapi_auth(tmp_path)
+ assert len(findings) == 0
+
+ def test_comment_bypass_prevented(self, tmp_path):
+ """Should detect unprotected endpoint even if comments or string literals contain 'Depends'."""
+ file = tmp_path / "routes.py"
+ file.write_text('''
+@router.post("/auth/login")
+async def auth_login():
+ # This is a comment containing Depends() or Security()
+ dummy = "Depends(auth_provider)"
+ return {"status": "ok"}
+''')
+ findings = scan_fastapi_auth(tmp_path)
+ assert len(findings) == 1
+ assert findings[0].line_number == 2
+ assert "auth_login" in findings[0].description
+
+ def test_relative_target_dir_no_value_error(self, tmp_path):
+ """Should support relative Path objects without raising a ValueError on relative_to."""
+ file = tmp_path / "routes.py"
+ file.write_text('''
+@router.get("/unprotected")
+def unprotected():
+ pass
+''')
+ import os
+ # Change current working directory to parent of tmp_path to make a relative path
+ old_cwd = os.getcwd()
+ os.chdir(tmp_path.parent)
+ try:
+ relative_path = Path(tmp_path.name)
+ # This should not raise ValueError
+ findings = scan_fastapi_auth(relative_path)
+ assert len(findings) == 1
+ assert findings[0].file_path == "routes.py"
+ finally:
+ os.chdir(old_cwd)
+
+
+class TestSupabaseRLSScanner:
+ def test_detects_missing_rls(self, tmp_path):
+ """Should detect created tables that do not have RLS enabled."""
+ file = tmp_path / "migration.sql"
+ file.write_text('''
+CREATE TABLE users (
+ id UUID PRIMARY KEY,
+ email TEXT
+);
+
+CREATE TABLE IF EXISTS posts (
+ id SERIAL PRIMARY KEY,
+ title TEXT
+);
+''')
+ findings = scan_supabase_rls(tmp_path)
+ assert len(findings) == 2
+
+ assert findings[0].category == Category.AUTH
+ assert findings[0].severity == Severity.HIGH
+ assert "users" in findings[0].description
+ assert findings[0].line_number == 2
+
+ assert "posts" in findings[1].description
+ assert findings[1].line_number == 7
+
+ def test_ignores_enabled_rls(self, tmp_path):
+ """Should ignore tables that have RLS enabled in the SQL file."""
+ file = tmp_path / "migration.sql"
+ file.write_text('''
+CREATE TABLE users (
+ id UUID PRIMARY KEY
+);
+
+ALTER TABLE users ENABLE ROW LEVEL SECURITY;
+''')
+ findings = scan_supabase_rls(tmp_path)
+ assert len(findings) == 0
+
+ def test_supabase_rls_mixed_case_and_formatting(self, tmp_path):
+ """Should correctly handle formatting changes, newlines, and case insensitivity."""
+ file = tmp_path / "migration.sql"
+ file.write_text('''
+Create Table "Users" (
+ id UUID PRIMARY KEY
+);
+
+Alter Table "users"
+ Enable Row Level Security;
+''')
+ findings = scan_supabase_rls(tmp_path)
+ assert len(findings) == 0
diff --git a/tests/test_pattern_scanner.py b/tests/test_pattern_scanner.py
index 165193d..00b0909 100644
--- a/tests/test_pattern_scanner.py
+++ b/tests/test_pattern_scanner.py
@@ -34,3 +34,89 @@ def test_finding_has_line_number(self, vulnerable_repo):
findings = scan_patterns(vulnerable_repo)
for f in findings:
assert f.line_number is not None, f"Finding {f.id} missing line number"
+
+ def test_xss_in_frontend_templates(self, tmp_path):
+ """Should detect XSS patterns in HTML, Vue, and Svelte templates."""
+ html_file = tmp_path / "index.html"
+ html_file.write_text("element.innerHTML = 'bad'
")
+
+ vue_file = tmp_path / "App.vue"
+ vue_file.write_text("")
+
+ svelte_file = tmp_path / "App.svelte"
+ svelte_file.write_text("")
+
+ findings = scan_patterns(tmp_path)
+ assert len(findings) >= 3
+
+ def test_unreadable_file_logged(self, tmp_path, caplog):
+ """Should log a warning message when a file cannot be read due to OS error."""
+ import logging
+ from unittest.mock import patch
+
+ bad_file = tmp_path / "unreadable.py"
+ bad_file.write_text("execute(f'SELECT *')")
+
+ with patch("os.open", side_effect=PermissionError("Mocked permission error")):
+ with caplog.at_level(logging.WARNING):
+ findings = scan_patterns(tmp_path)
+ assert len(findings) == 0
+
+ warnings = [rec.message for rec in caplog.records if rec.levelno == logging.WARNING]
+ assert any("Could not read file" in w and "Mocked permission error" in w for w in warnings)
+
+ def test_large_file_skipped(self, tmp_path):
+ """Should skip scanning files that exceed the 5MB size limit."""
+ large_file = tmp_path / "large.py"
+ large_file.write_text("execute(f'SELECT *')" + (" " * 5 * 1024 * 1024))
+ findings = scan_patterns(tmp_path)
+ assert len(findings) == 0
+
+ def test_path_traversal_prevention(self, tmp_path):
+ """Should prevent scanning files resolving outside the target boundary."""
+ from unittest.mock import MagicMock, patch
+ from pathlib import Path
+
+ file_mock = MagicMock(spec=Path)
+ file_mock.is_symlink.return_value = False
+ file_mock.is_file.return_value = True
+ file_mock.suffix = ".py"
+ file_mock.resolve.return_value = Path("C:/Windows") if Path("C:/").exists() else Path("/etc")
+ file_mock.stat.return_value.st_size = 100
+
+ with patch.object(Path, "rglob", return_value=[file_mock]):
+ findings = scan_patterns(tmp_path)
+ assert len(findings) == 0
+
+ def test_multiline_command_injection(self, tmp_path):
+ """Should detect multi-line subprocess execution with shell=True."""
+ cmd_file = tmp_path / "command.py"
+ cmd_file.write_text(
+ "subprocess.run(\n"
+ " ['ls', '-l'],\n"
+ " shell=True\n"
+ ")\n"
+ )
+ findings = scan_patterns(tmp_path)
+ assert len(findings) == 1
+ assert "Command injection" in findings[0].title
+
+ def test_is_safe_bypass_prevented(self, tmp_path):
+ """Should detect dangerous patterns even if safe-looking comments are present on the same line."""
+ bypass_file = tmp_path / "bypass.html"
+ bypass_file.write_text("el.innerHTML = bad; // .textContent = safe\n")
+ findings = scan_patterns(tmp_path)
+ assert len(findings) == 1
+ assert "XSS risk" in findings[0].title
+
+ def test_long_line_bypass_prevented(self, tmp_path):
+ """Should detect dangerous patterns on lines exceeding MAX_LINE_LENGTH by truncating them."""
+ long_line_file = tmp_path / "long_line.js"
+ payload = "element.innerHTML = bad;" + (" " * 3000) + "\n"
+ long_line_file.write_text(payload)
+ findings = scan_patterns(tmp_path)
+ assert len(findings) == 1
+ assert "XSS risk" in findings[0].title
+
+
+
diff --git a/tests/test_secret_scanner.py b/tests/test_secret_scanner.py
index 0d34c09..97326c5 100644
--- a/tests/test_secret_scanner.py
+++ b/tests/test_secret_scanner.py
@@ -81,3 +81,18 @@ def test_default_fallback_token_detected(self, tmp_path):
assert len(findings) == 1
assert "Anthropic API key" in findings[0].title
+ def test_unreadable_file_logged(self, tmp_path, capsys):
+ """Should log a warning message when a file cannot be read."""
+ from unittest.mock import patch
+ unreadable_file = tmp_path / "unreadable.py"
+ unreadable_file.write_text("API_KEY = 'sk-ant-12345678901234567890'")
+
+ with patch("builtins.open", side_effect=PermissionError("Mocked permission error")):
+ findings = scan_secrets(tmp_path)
+ assert len(findings) == 0
+
+ captured = capsys.readouterr()
+ assert "WARNING: Could not read file" in captured.out
+ assert "Mocked permission error" in captured.out
+
+