#!/usr/bin/env python3
"""OWASP Project Audit — revisión educativa no intrusiva (stdlib only).

Busca señales frecuentes alineadas con riesgos del OWASP Top 10 (credenciales
embebidas, SQL sospechoso, flags de debug, TLS verify desactivado, funciones
peligrosas y cabeceras HTTP ausentes). No es un pentest ni sustituye a un
análisis profesional: es un recordatorio para el día a día.

Requisitos: Python 3.10+. Solo biblioteca estándar.
Uso:
  python3 owasp_project_audit.py --path ./mi-proyecto
  python3 owasp_project_audit.py --url https://ejemplo.com
  python3 owasp_project_audit.py --path ./mi-proyecto --url https://ejemplo.com --json
"""
from __future__ import annotations

import argparse
import json
import re
import ssl
import sys
import urllib.error
import urllib.request
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Iterable

VERSION = "1.0.0-edu"
SKIP_DIRS = {
    ".git", ".hg", ".svn", ".venv", "venv", "node_modules", "__pycache__",
    "dist", "build", ".next", ".tox", ".mypy_cache", "vendor", "target",
}
TEXT_SUFFIXES = {
    ".py", ".js", ".ts", ".tsx", ".jsx", ".php", ".go", ".rb", ".java",
    ".cs", ".env", ".yml", ".yaml", ".json", ".toml", ".ini", ".cfg",
    ".conf", ".sh", ".bash", ".sql", ".md", ".txt", ".html", ".vue",
    ".rs", ".kt", ".swift", ".c", ".cpp", ".h", ".xml", ".properties",
}
MAX_FILE_BYTES = 1_500_000

SECURITY_HEADERS = (
    "Strict-Transport-Security",
    "Content-Security-Policy",
    "X-Content-Type-Options",
    "X-Frame-Options",
    "Referrer-Policy",
    "Permissions-Policy",
)

PATTERNS: list[tuple[str, str, re.Pattern[str], str]] = [
    (
        "embedded_credentials",
        "medium",
        re.compile(
            r"(?i)(api[_-]?key|secret|password|passwd|token|access[_-]?key)"
            r"\s*[=:]\s*['\"][^'\"]{8,}['\"]"
        ),
        "Posible credencial o secreto embebido en el código (A07 / A04).",
    ),
    (
        "suspicious_sql",
        "medium",
        re.compile(
            r"(?i)(execute|query|raw)\s*\(\s*(f['\"]|['\"].*?\+|['\"].*?\%|`.*?\{)"
        ),
        "Construcción dinámica de SQL: riesgo de inyección si no hay parámetros (A05).",
    ),
    (
        "sql_string_concat",
        "low",
        re.compile(
            r"(?i)(SELECT|INSERT|UPDATE|DELETE)\b[^;\n]{0,120}(\+|\%s|\$\{|f['\"])"
        ),
        "SQL mezclado con concatenación o interpolación; revisar prepared statements (A05).",
    ),
    (
        "debug_flags",
        "medium",
        re.compile(
            r"(?i)(DEBUG\s*=\s*True|APP_DEBUG\s*=\s*true|display_errors\s*=\s*On|"
            r"flask\.env\s*=\s*development)"
        ),
        "Flag de depuración activo: no debería llegar a producción (A02).",
    ),
    (
        "tls_verify_off",
        "high",
        re.compile(
            r"(?i)(verify\s*=\s*False|CURLOPT_SSL_VERIFYPEER\s*,\s*false|"
            r"InsecureSkipVerify\s*:\s*true|rejectUnauthorized\s*:\s*false|"
            r"ssl\._create_unverified_context)"
        ),
        "Verificación TLS desactivada: abre la puerta a MITM (A02 / A04).",
    ),
    (
        "dangerous_functions",
        "high",
        re.compile(
            r"(?i)\b(eval\s*\(|exec\s*\(|pickle\.loads\s*\(|unserialize\s*\(|"
            r"Runtime\.getRuntime\(\)\.exec|child_process\.exec\s*\(|"
            r"shell_exec\s*\(|passthru\s*\(|system\s*\(|assert\s*\()"
        ),
        "Función peligrosa / ejecución dinámica: revisar entrada y necesidad (A03 / A05 / A08).",
    ),
]


@dataclass
class Finding:
    check: str
    severity: str
    path: str
    line: int | None
    detail: str
    evidence: str | None = None


def iter_source_files(root: Path) -> Iterable[Path]:
    for path in root.rglob("*"):
        if not path.is_file():
            continue
        if any(part in SKIP_DIRS for part in path.parts):
            continue
        if path.suffix.lower() not in TEXT_SUFFIXES and path.name not in {
            "Dockerfile", "Makefile", ".env", ".env.local", ".env.production",
        }:
            continue
        try:
            if path.stat().st_size > MAX_FILE_BYTES:
                continue
        except OSError:
            continue
        yield path


def scan_path(root: Path) -> list[Finding]:
    findings: list[Finding] = []
    if not root.exists():
        findings.append(Finding(
            "path", "high", str(root), None,
            f"La ruta no existe: {root}",
        ))
        return findings
    if not root.is_dir():
        findings.append(Finding(
            "path", "high", str(root), None,
            "Se esperaba un directorio de proyecto.",
        ))
        return findings

    for file_path in iter_source_files(root):
        try:
            text = file_path.read_text(encoding="utf-8", errors="replace")
        except OSError as exc:
            findings.append(Finding(
                "io", "low", str(file_path), None, f"No se pudo leer: {exc}",
            ))
            continue
        rel = str(file_path.relative_to(root))
        lines = text.splitlines()
        for check, severity, pattern, detail in PATTERNS:
            for idx, line in enumerate(lines, start=1):
                if pattern.search(line):
                    evidence = line.strip()
                    if len(evidence) > 160:
                        evidence = evidence[:157] + "..."
                    findings.append(Finding(
                        check, severity, rel, idx, detail, evidence,
                    ))
    return findings


def scan_url(url: str, timeout: float = 12.0) -> list[Finding]:
    findings: list[Finding] = []
    if not url.lower().startswith(("http://", "https://")):
        findings.append(Finding(
            "url", "high", url, None,
            "La URL debe empezar por http:// o https://",
        ))
        return findings

    ctx = ssl.create_default_context()
    req = urllib.request.Request(
        url,
        method="GET",
        headers={"User-Agent": f"owasp-project-audit/{VERSION} (educational)"},
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
            headers = {k.title(): v for k, v in resp.headers.items()}
            final_url = resp.geturl()
            status = getattr(resp, "status", None) or resp.getcode()
    except urllib.error.HTTPError as exc:
        headers = {k.title(): v for k, v in (exc.headers or {}).items()}
        final_url = url
        status = exc.code
        findings.append(Finding(
            "http_status", "low", url, None,
            f"La URL respondió HTTP {exc.code}; igual revisamos cabeceras.",
        ))
    except Exception as exc:  # noqa: BLE001 — herramienta educativa
        findings.append(Finding(
            "url", "high", url, None,
            f"No se pudo consultar la URL: {exc}",
        ))
        return findings

    if final_url.lower().startswith("http://"):
        findings.append(Finding(
            "tls", "high", final_url, None,
            "El sitio se sirve por HTTP sin TLS (A02 / A04).",
        ))

    missing = [h for h in SECURITY_HEADERS if h not in headers]
    if missing:
        findings.append(Finding(
            "missing_http_headers",
            "medium" if len(missing) >= 3 else "low",
            final_url,
            None,
            "Cabeceras de seguridad ausentes o no visibles en la respuesta "
            f"(HTTP {status}): {', '.join(missing)}. Un WAF o CDN puede "
            "añadirlas después; verifica en el navegador también (A02 / A09).",
        ))
    else:
        findings.append(Finding(
            "http_headers_ok",
            "info",
            final_url,
            None,
            "Se observaron las cabeceras de seguridad básicas revisadas.",
        ))
    return findings


def summarize(findings: list[Finding]) -> dict[str, int]:
    counts = {"high": 0, "medium": 0, "low": 0, "info": 0}
    for f in findings:
        counts[f.severity] = counts.get(f.severity, 0) + 1
    return counts


def print_human(findings: list[Finding]) -> None:
    print(f"OWASP Project Audit v{VERSION} (educativo, no intrusivo)")
    print("=" * 60)
    if not findings:
        print("Sin hallazgos con las reglas actuales. No significa «seguro».")
        return
    for f in findings:
        loc = f.path
        if f.line:
            loc = f"{f.path}:{f.line}"
        print(f"[{f.severity.upper():6}] {f.check} — {loc}")
        print(f"         {f.detail}")
        if f.evidence:
            print(f"         evidencia: {f.evidence}")
        print()
    counts = summarize(findings)
    print("-" * 60)
    print(
        "Resumen:",
        ", ".join(f"{k}={v}" for k, v in counts.items() if v),
    )
    print(
        "Aviso: esto no es un pentest. Úsalo como lista de recordatorios "
        "antes de un despliegue o code review."
    )


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Auditoría educativa no intrusiva alineada a ideas OWASP.",
    )
    parser.add_argument("--path", type=Path, help="Directorio del proyecto a revisar")
    parser.add_argument("--url", help="URL pública para revisar cabeceras HTTP")
    parser.add_argument("--json", action="store_true", help="Salida JSON")
    parser.add_argument("--version", action="version", version=VERSION)
    args = parser.parse_args(argv)

    if not args.path and not args.url:
        parser.error("Indica al menos --path o --url")

    findings: list[Finding] = []
    if args.path:
        findings.extend(scan_path(args.path))
    if args.url:
        findings.extend(scan_url(args.url))

    if args.json:
        payload = {
            "tool": "owasp_project_audit",
            "version": VERSION,
            "summary": summarize(findings),
            "findings": [asdict(f) for f in findings],
            "disclaimer": (
                "Herramienta educativa. No sustituye pentest, SCA ni revisión humana."
            ),
        }
        json.dump(payload, sys.stdout, ensure_ascii=False, indent=2)
        print()
    else:
        print_human(findings)

    # Código de salida: 1 si hay high, 0 en el resto (útil en CI suave).
    return 1 if any(f.severity == "high" for f in findings) else 0


if __name__ == "__main__":
    raise SystemExit(main())