src/hermes/tools/security_scanner.py

"""Security scanner tool implementation."""

from __future__ import annotations

import logging
import re
from pathlib import Path
from typing import Any

from hermes.tools.base.tool import BaseTool, ToolSchema

logger = logging.getLogger(__name__)

SECRET_PATTERNS = [
    (r"(?:api[_-]?key|apikey)\s*[=:]\s*['\"]([^'\"]+)['\"]", "API Key"),
    (r"(?:secret[_-]?key|secretkey)\s*[=:]\s*['\"]([^'\"]+)['\"]", "Secret Key"),
    (r"(?:password|passwd|pwd)\s*[=:]\s*['\"]([^'\"]+)['\"]", "Password"),
    (r"(?:token|access[_-]?token|auth[_-]?token)\s*[=:]\s*['\"]([^'\"]+)['\"]", "Token"),
    (r"(?:aws[_-]?access[_-]?key[_-]?id)\s*[=:]\s*['\"]([^'\"]+)['\"]", "AWS Access Key"),
    (r"(?:aws[_-]?secret[_-]?access[_-]?key)\s*[=:]\s*['\"]([^'\"]+)['\"]", "AWS Secret Key"),
    (r"(?:private[_-]?key)\s*[=:]\s*['\"]([^'\"]+)['\"]", "Private Key"),
    (r"-----BEGIN\s+(RSA|EC|DSA)?\s*PRIVATE\s+KEY-----", "Private Key Block"),
    (r"(?:ghp|gho|ghu|ghs|ghr)[A-Za-z0-9]{36,}", "GitHub Token"),
    (r"sk-[A-Za-z0-9]{20,}", "OpenAI API Key"),
    (r"xox[bpsa]-[A-Za-z0-9-]+", "Slack Token"),
    (r"(?:AKIA|ASIA)[A-Z0-9]{16}", "AWS Access Key ID"),
]

VULNERABILITY_PATTERNS = [
    (r"eval\s*\(", "Code Injection", "high", "Use of eval() can lead to code injection"),
    (r"exec\s*\(", "Code Injection", "high", "Use of exec() can lead to code injection"),
    (r"subprocess\.call.*shell\s*=\s*True", "Shell Injection", "high", "Shell injection via subprocess"),
    (r"os\.system\s*\(", "Shell Injection", "high", "Shell injection via os.system"),
    (r"SELECT\s+.*FROM\s+.*WHERE.*%s", "SQL Injection", "high", "Potential SQL injection"),
    (r"SELECT\s+.*FROM\s+.*WHERE.*\{", "SQL Injection", "high", "Potential SQL injection with f-string"),
    (r"pickle\.loads?\s*\(", "Deserialization", "high", "Unsafe deserialization with pickle"),
    (r"yaml\.load\s*\([^)]*\)", "Deserialization", "medium", "Unsafe YAML loading"),
    (r"assert\s+", "Assertion Usage", "low", "Assertions can be disabled with -O flag"),
    (r"DEBUG\s*=\s*True", "Debug Mode", "medium", "Debug mode should not be enabled in production"),
    (r"ALLOWED_HOSTS\s*=\s*\[.*\*.*\]", "Host Header", "medium", "Wildcard allowed hosts"),
]

OWASP_CHECKS = [
    {
        "id": "A01:2021",
        "name": "Broken Access Control",
        "patterns": [r"permission\s*=\s*None", r"auth\s*=\s*None", r"skip_auth"],
        "severity": "high",
    },
    {
        "id": "A02:2021",
        "name": "Cryptographic Failures",
        "patterns": [r"md5\s*\(", r"sha1\s*\(", r"DES\s*\("],
        "severity": "medium",
    },
    {
        "id": "A03:2021",
        "name": "Injection",
        "patterns": [r"eval\s*\(", r"exec\s*\(", r"execute\s*\(.*%"],
        "severity": "high",
    },
    {
        "id": "A05:2021",
        "name": "Security Misconfiguration",
        "patterns": [r"DEBUG\s*=\s*True", r"SECRET_KEY\s*=\s*['\"]dev"],
        "severity": "medium",
    },
    {
        "id": "A06:2021",
        "name": "Vulnerable Components",
        "patterns": [r"requests\.get\s*\(", r"http://"],
        "severity": "low",
    },
]


class SecurityScannerTool(BaseTool):
    """Tool for scanning code for security issues."""

    SCAN_BASE_DIR: str = "."

    def __init__(self, scan_base_dir: str | None = None) -> None:
        super().__init__()
        if scan_base_dir:
            self.SCAN_BASE_DIR = scan_base_dir

    def _resolve_scan_path(self, path: str) -> str:
        """Resolve and validate scan path within base directory."""
        base = Path(self.SCAN_BASE_DIR).resolve()
        target = (base / path).resolve()
        try:
            target.relative_to(base)
        except ValueError:
            raise ValueError("Path escapes allowed scan directory") from None
        if not target.exists():
            raise FileNotFoundError(f"Path not found: {path}")
        return str(target)

    def _define_schema(self) -> ToolSchema:
        return ToolSchema(
            name="security_scanner",
            description="Scan code for security vulnerabilities, exposed secrets, and OWASP issues",
            parameters={
                "action": {
                    "type": "string",
                    "description": "Action: scan_file, scan_directory, scan_code",
                },
                "path": {
                    "type": "string",
                    "description": "File or directory path to scan",
                },
                "code": {
                    "type": "string",
                    "description": "Code string to scan (for scan_code action)",
                },
                "include_patterns": {
                    "type": "boolean",
                    "description": "Include pattern matches in results",
                    "default": True,
                },
            },
            required=["action"],
            category="security",
            tags=["security", "scan", "vulnerability"],
        )

    async def execute(self, **kwargs: Any) -> dict[str, Any]:
        """Execute security scan."""
        action = kwargs["action"]

        try:
            if action == "scan_file":
                return await self._scan_file(kwargs["path"], kwargs.get("include_patterns", True))
            elif action == "scan_directory":
                return await self._scan_directory(kwargs["path"], kwargs.get("include_patterns", True))
            elif action == "scan_code":
                return await self._scan_code(kwargs["code"], kwargs.get("include_patterns", True))
            else:
                return {"error": f"Unknown action: {action}"}
        except Exception as e:
            logger.error(f"Security scan error: {e}")
            return {"error": str(e)}

    async def _scan_code(self, code: str, include_patterns: bool) -> dict[str, Any]:
        """Scan code string."""
        secrets = self._find_secrets(code)
        vulnerabilities = self._find_vulnerabilities(code)
        owasp = self._check_owasp(code)

        return {
            "secrets": secrets,
            "vulnerabilities": vulnerabilities,
            "owasp_issues": owasp,
            "summary": {
                "total_secrets": len(secrets),
                "total_vulnerabilities": len(vulnerabilities),
                "total_owasp_issues": len(owasp),
                "risk_level": self._calculate_risk_level(secrets, vulnerabilities, owasp),
            },
        }

    async def _scan_file(self, path: str, include_patterns: bool) -> dict[str, Any]:
        """Scan a file with path traversal protection."""
        resolved = self._resolve_scan_path(path)
        code = Path(resolved).read_text(encoding="utf-8", errors="replace")
        result = await self._scan_code(code, include_patterns)
        result["file"] = resolved
        return result

    async def _scan_directory(self, path: str, include_patterns: bool) -> dict[str, Any]:
        """Scan a directory with path traversal protection."""
        resolved = self._resolve_scan_path(path)
        dir_path = Path(resolved)

        all_results = []
        scanned = 0

        for file_path in dir_path.rglob("*"):
            if file_path.is_file() and file_path.suffix in (".py", ".js", ".ts", ".yaml", ".yml", ".json", ".env"):
                try:
                    code = file_path.read_text(encoding="utf-8", errors="replace")
                    result = await self._scan_code(code, include_patterns)
                    if any(
                        [result["secrets"], result["vulnerabilities"], result["owasp_issues"]]
                    ):
                        result["file"] = str(file_path)
                        all_results.append(result)
                    scanned += 1
                except Exception:
                    continue

        total_secrets = sum(len(r["secrets"]) for r in all_results)
        total_vulns = sum(len(r["vulnerabilities"]) for r in all_results)
        total_owasp = sum(len(r["owasp_issues"]) for r in all_results)

        return {
            "files_scanned": scanned,
            "files_with_issues": len(all_results),
            "results": all_results,
            "summary": {
                "total_secrets": total_secrets,
                "total_vulnerabilities": total_vulns,
                "total_owasp_issues": total_owasp,
                "risk_level": self._calculate_risk_level_from_counts(total_secrets, total_vulns, total_owasp),
            },
        }

    def _find_secrets(self, code: str) -> list[dict[str, Any]]:
        """Find exposed secrets in code."""
        secrets = []
        for pattern, secret_type in SECRET_PATTERNS:
            matches = re.finditer(pattern, code, re.IGNORECASE)
            for match in matches:
                secrets.append(
                    {
                        "type": secret_type,
                        "match": match.group()[:50] + "..." if len(match.group()) > 50 else match.group(),
                        "line": code[:match.start()].count("\n") + 1,
                        "severity": "critical",
                    }
                )
        return secrets

    def _find_vulnerabilities(self, code: str) -> list[dict[str, Any]]:
        """Find vulnerabilities in code."""
        vulnerabilities = []
        for pattern, vuln_type, severity, description in VULNERABILITY_PATTERNS:
            matches = re.finditer(pattern, code, re.IGNORECASE)
            for match in matches:
                vulnerabilities.append(
                    {
                        "type": vuln_type,
                        "severity": severity,
                        "description": description,
                        "line": code[:match.start()].count("\n") + 1,
                        "match": match.group(),
                    }
                )
        return vulnerabilities

    def _check_owasp(self, code: str) -> list[dict[str, Any]]:
        """Check OWASP Top 10."""
        issues = []
        for check in OWASP_CHECKS:
            for pattern in check["patterns"]:
                if re.search(pattern, code, re.IGNORECASE):
                    issues.append(
                        {
                            "owasp_id": check["id"],
                            "name": check["name"],
                            "severity": check["severity"],
                            "pattern_matched": pattern,
                        }
                    )
                    break
        return issues

    def _calculate_risk_level(
        self, secrets: list, vulnerabilities: list, owasp: list
    ) -> str:
        """Calculate overall risk level."""
        return self._calculate_risk_level_from_counts(len(secrets), len(vulnerabilities), len(owasp))

    def _calculate_risk_level_from_counts(
        self, secret_count: int, vuln_count: int, owasp_count: int
    ) -> str:
        """Calculate risk level from counts."""
        if secret_count > 0 or vuln_count > 3:
            return "critical"
        elif vuln_count > 0 or owasp_count > 2:
            return "high"
        elif owasp_count > 0:
            return "medium"
        return "low"