Hugging Face's logo

Spaces:
YOSOYYONOSOYOTRO
/
crowdata
Paused

App Files Community
crowdata
File size: 4,953 Bytes
83bdb4a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
import os
import secrets
import hashlib
from typing import Callable, Optional
from fastapi import Request, Response, HTTPException, status
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse


class CSRFMiddleware(BaseHTTPMiddleware):
    """
    CSRF Protection usando patr贸n Double-Submit Cookie.
    
    - Genera cookie `csrf_token` (HttpOnly=False, SameSite=Lax) al iniciar sesi贸n
    - Valida header `X-CSRF-Token` en m茅todos mutantes (POST, PUT, PATCH, DELETE)
    - Excluye: GET, HEAD, OPTIONS, endpoints de auth (/api/auth/*)
    - Skip si no hay cookie (p.ej. APIs program谩ticas con Bearer token)
    """
    
    def __init__(
        self, 
        app, 
        cookie_name: str = "csrf_token",
        header_name: str = "X-CSRF-Token",
        cookie_secure: bool = True,
        cookie_samesite: str = "lax",
        excluded_paths: Optional[list] = None,
        excluded_methods: Optional[list] = None,
    ):
        super().__init__(app)
        self.cookie_name = cookie_name
        self.header_name = header_name
        self.cookie_secure = cookie_secure
        self.cookie_samesite = cookie_samesite
        self.excluded_paths = excluded_paths or [
            "/api/auth/jwt/login",
            "/api/auth/jwt/logout", 
            "/api/auth/register",
            "/api/auth/forgot-password",
            "/api/auth/reset-password",
            "/api/auth/users/me",
        ]
        self.excluded_methods = excluded_methods or ["GET", "HEAD", "OPTIONS"]
    
    def _get_csrf_token(self, request: Request) -> Optional[str]:
        """Obtener token CSRF de la cookie."""
        cookie_header = request.headers.get("cookie", "")
        for cookie in cookie_header.split(";"):
            cookie = cookie.strip()
            if cookie.startswith(f"{self.cookie_name}="):
                return cookie.split("=", 1)[1]
        return None
    
    def _generate_csrf_token(self) -> str:
        """Generar token CSRF criptogr谩ficamente seguro."""
        return secrets.token_urlsafe(32)
    
    def _is_excluded_path(self, path: str) -> bool:
        """Verificar si el path est谩 excluido de validaci贸n CSRF."""
        for excluded in self.excluded_paths:
            if path.startswith(excluded):
                return True
        return False
    
    async def dispatch(self, request: Request, call_next: Callable):
        # Skip CSRF para m茅todos seguros
        if request.method in self.excluded_methods:
            return await call_next(request)
        
        # Skip CSRF para paths excluidos
        if self._is_excluded_path(request.url.path):
            return await call_next(request)
        
        # Solo validar si hay cookie de sesi贸n (usuario logueado via cookie)
        # APIs program谩ticas con Bearer token no necesitan CSRF
        session_cookie = request.cookies.get("cd_token")
        if not session_cookie:
            # No hay cookie de sesi贸n, asumimos API program谩tica
            return await call_next(request)
        
        # Obtener token CSRF de cookie
        csrf_cookie = self._get_csrf_token(request)
        csrf_header = request.headers.get(self.header_name)
        
        if not csrf_cookie:
            # No hay token CSRF en cookie - generar uno nuevo y setear
            response = await call_next(request)
            new_token = self._generate_csrf_token()
            response.set_cookie(
                key=self.cookie_name,
                value=new_token,
                max_age=7 * 24 * 60 * 60,  # 7 d铆as
                httponly=False,  # JS necesita leerlo para enviar en header
                secure=self.cookie_secure,
                samesite=self.cookie_samesite,
                path="/",
            )
            return response
        
        if not csrf_header:
            return JSONResponse(
                status_code=status.HTTP_403_FORBIDDEN,
                content={"detail": "CSRF token requerido en header X-CSRF-Token"},
            )
        
        # Validar token (comparaci贸n timing-safe)
        if not secrets.compare_digest(csrf_cookie, csrf_header):
            return JSONResponse(
                status_code=status.HTTP_403_FORBIDDEN,
                content={"detail": "CSRF token inv谩lido"},
            )
        
        # Token v谩lido - proceder
        return await call_next(request)


def get_csrf_middleware(
    cookie_secure: bool = True,
    cookie_samesite: str = "lax",
    excluded_paths: Optional[list] = None,
) -> type:
    """Factory para crear middleware CSRF con configuraci贸n personalizada."""
    class ConfiguredCSRFMiddleware(CSRFMiddleware):
        def __init__(self, app):
            super().__init__(
                app,
                cookie_secure=cookie_secure,
                cookie_samesite=cookie_samesite,
                excluded_paths=excluded_paths,
            )
    return ConfiguredCSRFMiddleware