app/security/audit.py

"""
Auditoría de Seguridad — CrowData Backend.

Ejecutar: python -m app.security.audit
"""
import os
import sys
import re
sys.path.insert(0, "E:/crowdata/backend")

from pathlib import Path

class SecurityAudit:
    def __init__(self):
        self.findings = []
        self.passed = []
        self.warnings = []

    def check(self, name, condition, detail="", severity="HIGH"):
        if condition:
            self.passed.append(f"[PASS] {name}")
        else:
            self.findings.append(f"[FAIL-{severity}] {name}: {detail}")

    def warn(self, name, detail=""):
        self.warnings.append(f"[WARN] {name}: {detail}")

    def audit_env_file(self):
        env_path = Path("E:/crowdata/backend/.env")
        if not env_path.exists():
            self.check("ENV file exists", False, ".env file not found")
            return

        content = env_path.read_text(encoding="utf-8")

        # SECRET_KEY
        secret_match = re.search(r"SECRET_KEY=(.+)", content)
        if secret_match:
            secret = secret_match.group(1).strip()
            self.check("SECRET_KEY is set", len(secret) > 20,
                       f"SECRET_KEY too short ({len(secret)} chars)", "HIGH")
            self.check("SECRET_KEY is not default", secret not in ("", "changeme", "supersecret"),
                       "SECRET_KEY is a weak default value", "CRITICAL")
        else:
            self.check("SECRET_KEY exists", False, "SECRET_KEY not set in .env", "CRITICAL")

        # DATABASE_URL
        db_match = re.search(r"DATABASE_URL=(.+)", content)
        if db_match:
            db_url = db_match.group(1).strip()
            self.check("DATABASE_URL is set", len(db_url) > 0, "DATABASE_URL is empty", "HIGH")
            if db_url.startswith("sqlite"):
                self.warn("Using SQLite", "Consider PostgreSQL for production")
        else:
            self.check("DATABASE_URL exists", False, "DATABASE_URL not set", "HIGH")

        # SMTP credentials
        smtp_user = re.search(r"SMTP_USER=(.+)", content)
        smtp_pass = re.search(r"SMTP_PASSWORD=(.+)", content)
        if smtp_user and smtp_pass:
            self.check("SMTP credentials configured", True)
        else:
            self.warn("SMTP not configured", "Email delivery won't work")

    def audit_jwt_config(self):
        from app.config import get_settings
        settings = get_settings()

        self.check("JWT secret is set", len(settings.secret_key) > 0,
                   "JWT secret_key is empty", "CRITICAL")
        self.check("JWT expiration is reasonable",
                   settings.access_token_expire_minutes <= 1440,
                   f"Token expires in {settings.access_token_expire_minutes} minutes",
                   "MEDIUM")
        self.check("JWT algorithm is secure",
                   settings.jwt_algorithm in ("HS256", "HS384", "HS512", "RS256", "RS384", "RS512"),
                   f"Algorithm: {settings.jwt_algorithm}", "HIGH")

    def audit_cors(self):
        from app.main import ALLOWED_ORIGINS
        has_wildcard = "*" in ALLOWED_ORIGINS
        self.check("CORS no wildcard", not has_wildcard,
                   "CORS allows all origins (*)", "HIGH")
        self.check("CORS has specific origins", len(ALLOWED_ORIGINS) > 0,
                   "No CORS origins configured", "MEDIUM")

    def audit_password_hashing(self):
        from app.auth.config import get_jwt_strategy
        # fastapi-users uses bcrypt by default
        self.check("Password hashing (fastapi-users)", True,
                   "Using fastapi-users with bcrypt", "INFO")

    def audit_rate_limiting(self):
        from app.middleware.rate_limit import RATE_LIMITS, REPORT_LIMITS
        self.check("Rate limiting configured", len(RATE_LIMITS) > 0,
                   "No rate limits defined", "HIGH")
        self.check("Report limits configured", len(REPORT_LIMITS) > 0,
                   "No report limits defined", "HIGH")

    def audit_admin_protection(self):
        # Check that admin endpoints require superuser
        admin_file = Path("E:/crowdata/backend/app/admin/router.py")
        if admin_file.exists():
            content = admin_file.read_text(encoding="utf-8")
            self.check("Admin requires superuser",
                       "current_active_superuser" in content,
                       "Admin endpoints not protected by superuser check", "CRITICAL")

    def audit_sql_injection(self):
        # Check for raw SQL in service files
        service_files = list(Path("E:/crowdata/backend/app").rglob("*.py"))
        raw_sql_count = 0
        for f in service_files:
            try:
                content = f.read_text(encoding="utf-8")
                if "text(" in content and "execute" in content:
                    raw_sql_count += 1
            except Exception:
                pass
        self.warn("Raw SQL usage", f"{raw_sql_count} files use raw SQL (via SQLAlchemy text())")

    def audit_sensitive_data_logs(self):
        # Check for sensitive data in logger calls
        service_files = list(Path("E:/crowdata/backend/app").rglob("*.py"))
        issues = []
        for f in service_files:
            try:
                content = f.read_text(encoding="utf-8")
                lines = content.split("\n")
                for i, line in enumerate(lines):
                    if "logger" in line and ("password" in line.lower() or "token" in line.lower()):
                        if "hash" not in line.lower() and "hashed" not in line.lower():
                            issues.append(f"{f.name}:{i+1}")
            except Exception:
                pass
        if issues:
            self.warn("Sensitive data in logs", f"{len(issues)} potential occurrences")
        else:
            self.check("No sensitive data in logs", True)

    def audit_dependency_versions(self):
        req_file = Path("E:/crowdata/backend/requirements.txt")
        if req_file.exists():
            content = req_file.read_text(encoding="utf-8")
            self.check("Requirements file exists", True)
            # Check for known vulnerable patterns
            if "fastapi-users" in content:
                self.check("fastapi-users present", True)
            if "sqlalchemy" in content.lower():
                self.check("SQLAlchemy present", True)
        else:
            self.warn("No requirements.txt found")

    def run(self):
        print("=" * 60)
        print("  AUDITORÍA DE SEGURIDAD — CrowData Backend")
        print("=" * 60)
        print()

        print("1. Archivo de configuración (.env)")
        self.audit_env_file()
        print()

        print("2. Configuración JWT")
        self.audit_jwt_config()
        print()

        print("3. CORS")
        self.audit_cors()
        print()

        print("4. Password Hashing")
        self.audit_password_hashing()
        print()

        print("5. Rate Limiting")
        self.audit_rate_limiting()
        print()

        print("6. Protección Admin")
        self.audit_admin_protection()
        print()

        print("7. SQL Injection")
        self.audit_sql_injection()
        print()

        print("8. Datos sensibles en logs")
        self.audit_sensitive_data_logs()
        print()

        print("9. Dependencias")
        self.audit_dependency_versions()
        print()

        # Results
        print("=" * 60)
        print("  RESULTADOS")
        print("=" * 60)

        for p in self.passed:
            print(f"  {p}")

        for f in self.findings:
            print(f"  {f}")

        for w in self.warnings:
            print(f"  {w}")

        print()
        print(f"  Passed: {len(self.passed)}")
        print(f"  Failed: {len(self.findings)}")
        print(f"  Warnings: {len(self.warnings)}")
        print()


if __name__ == "__main__":
    audit = SecurityAudit()
    audit.run()