Update: Sheet 4 access control
TAMDeveloper13
3 months ago
parent
2615ff9a6e
commit
d4175f75d0
@ -1,83 +1,125 @@
|
|||||||
|
"""
|
||||||
|
CSRF Token Verification Dependency for be-optimumoh (FastAPI).
|
||||||
|
|
||||||
import secrets
|
Menggantikan CSRFProtectMiddleware (Double Submit Cookie) dengan verifikasi
|
||||||
|
berbasis Redis. Token CSRF di-generate oleh be-auth (CSRFTokenResource) dan
|
||||||
|
disimpan di Redis dalam bentuk SHA256 hash.
|
||||||
|
|
||||||
|
Token Flow (generation di be-auth, verification di sini):
|
||||||
|
────────────────────────────────────────
|
||||||
|
GENERATION (be-auth / CSRFTokenResource):
|
||||||
|
1. token = secrets.token_urlsafe(32)
|
||||||
|
2. token_hash = SHA256(token)
|
||||||
|
3. redis[csrf:{user_id}:{path_hash}] = token_hash (dengan TTL)
|
||||||
|
4. return plain token ke client
|
||||||
|
|
||||||
|
VERIFICATION (dependency ini):
|
||||||
|
1. Baca X-CSRF-Token header dari request
|
||||||
|
2. Ambil user_id dari request.state.user (di-set oleh JWTBearer)
|
||||||
|
3. incoming_hash = SHA256(nilai X-CSRF-Token)
|
||||||
|
4. stored_hash = redis_auth.get("csrf:{user_id}:{path_hash}")
|
||||||
|
5. secrets.compare_digest(incoming_hash, stored_hash)
|
||||||
|
6. redis_auth.delete(key) ← one-time use
|
||||||
|
────────────────────────────────────────
|
||||||
|
|
||||||
|
NOTE: Dependency ini harus selalu dikombinasikan dengan JWTBearer() agar
|
||||||
|
request.state.user sudah tersedia saat dijalankan.
|
||||||
|
|
||||||
|
Usage:
|
||||||
|
from fastapi import Depends
|
||||||
|
from src.csrf_protect import csrf_protect
|
||||||
|
|
||||||
|
@router.post("/some-endpoint", dependencies=[Depends(csrf_protect)])
|
||||||
|
async def some_endpoint(...):
|
||||||
|
...
|
||||||
|
"""
|
||||||
|
import hashlib
|
||||||
import logging
|
import logging
|
||||||
from typing import Optional, List
|
import secrets
|
||||||
from fastapi import Request, HTTPException
|
|
||||||
from starlette.middleware.base import BaseHTTPMiddleware
|
import redis
|
||||||
from starlette.responses import Response
|
from fastapi import HTTPException, Request
|
||||||
|
|
||||||
|
import src.config as config
|
||||||
|
from src.context import get_user_id
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
class CSRFProtectMiddleware(BaseHTTPMiddleware):
|
# Koneksi Redis untuk CSRF — Redis DB 0 tempat be-auth menyimpan CSRF token
|
||||||
|
_redis_auth = redis.Redis(
|
||||||
|
host=config.REDIS_HOST,
|
||||||
|
port=int(config.REDIS_PORT),
|
||||||
|
db=config.REDIS_AUTH_DB,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def csrf_protect(request: Request) -> None:
|
||||||
"""
|
"""
|
||||||
CSRF Protection Middleware using Double Submit Cookie pattern.
|
FastAPI dependency untuk verifikasi CSRF token berbasis Redis.
|
||||||
|
Harus dikombinasikan dengan JWTBearer() dependency.
|
||||||
This middleware:
|
|
||||||
1. Ensures a 'csrf_token' cookie exists for every request.
|
|
||||||
2. For mutating requests (POST, PUT, PATCH, DELETE):
|
|
||||||
- Requires an 'X-CSRF-Token' header.
|
|
||||||
- Validates that the header value matches the 'csrf_token' cookie.
|
|
||||||
"""
|
"""
|
||||||
|
# ── Step 1: ambil user_id dari request.state (di-set oleh JWTBearer) ────
|
||||||
def __init__(
|
user = getattr(request.state, "user", None)
|
||||||
self,
|
user_id = str(user.user_id) if user else get_user_id()
|
||||||
app,
|
|
||||||
cookie_name: str = "csrf_token",
|
if not user_id:
|
||||||
header_name: str = "X-CSRF-Token",
|
raise HTTPException(status_code=401, detail="Authentication required")
|
||||||
excluded_paths: Optional[List[str]] = None,
|
|
||||||
safe_methods: List[str] = ("GET", "HEAD", "OPTIONS", "TRACE")
|
# ── Step 2: baca CSRF token dari request header ───────────────────────────
|
||||||
):
|
# Starlette lowercases all header names
|
||||||
super().__init__(app)
|
csrf_token_header = (
|
||||||
self.cookie_name = cookie_name
|
request.headers.get("x-csrf-token")
|
||||||
self.header_name = header_name
|
or request.headers.get("x-xsrf-token")
|
||||||
self.excluded_paths = excluded_paths or []
|
or request.headers.get("x-csrf-token")
|
||||||
self.safe_methods = safe_methods
|
)
|
||||||
|
|
||||||
async def dispatch(self, request: Request, call_next):
|
if not csrf_token_header:
|
||||||
# 1. Skip validation for "safe" methods (GET, HEAD, etc.) or excluded paths
|
log.warning(
|
||||||
is_safe_method = request.method in self.safe_methods
|
"[CSRF] Missing X-CSRF-Token header | user_id=%s | IP: %s",
|
||||||
is_excluded = any(request.url.path.startswith(path) for path in self.excluded_paths)
|
user_id,
|
||||||
|
request.client.host if request.client else "unknown",
|
||||||
if not is_safe_method and not is_excluded:
|
)
|
||||||
# 2. Extract token from cookie and header
|
raise HTTPException(status_code=403, detail="CSRF token missing in header")
|
||||||
csrf_cookie = request.cookies.get(self.cookie_name)
|
|
||||||
csrf_header = request.headers.get(self.header_name.lower()) # Starlette lowercase headers
|
csrf_token_header = csrf_token_header.strip()
|
||||||
|
|
||||||
# 3. Validate
|
# ── Step 3: bangun Redis key (harus sama dengan format di be-auth) ─────────
|
||||||
if not csrf_cookie:
|
current_path = request.url.path
|
||||||
log.warning(f"CSRF validation failed: Missing cookie '{self.cookie_name}'")
|
path_hash = hashlib.sha256(current_path.encode()).hexdigest()[:16]
|
||||||
raise HTTPException(
|
redis_key = f"csrf:{user_id}:{path_hash}"
|
||||||
status_code=403,
|
|
||||||
detail="CSRF validation failed: Missing session cookie."
|
# ── Step 4: ambil stored token hash dari Redis (DB 0) ────────────────────
|
||||||
)
|
stored_token_hash = _redis_auth.get(redis_key)
|
||||||
|
if not stored_token_hash:
|
||||||
if not csrf_header:
|
log.warning(
|
||||||
log.warning(f"CSRF validation failed: Missing header '{self.header_name}'")
|
"[CSRF] Token not found in Redis | user_id=%s | path=%s | IP: %s",
|
||||||
raise HTTPException(
|
user_id,
|
||||||
status_code=403,
|
current_path,
|
||||||
detail=f"CSRF validation failed: Missing header '{self.header_name}'."
|
request.client.host if request.client else "unknown",
|
||||||
)
|
)
|
||||||
|
raise HTTPException(status_code=403, detail="CSRF token expired or invalid")
|
||||||
if not secrets.compare_digest(csrf_cookie, csrf_header):
|
|
||||||
log.warning(f"CSRF validation failed: Token mismatch for path {request.url.path}")
|
if isinstance(stored_token_hash, bytes):
|
||||||
raise HTTPException(
|
stored_token_hash = stored_token_hash.decode("utf-8")
|
||||||
status_code=403,
|
|
||||||
detail="CSRF validation failed: Token mismatch."
|
# ── Step 5: bandingkan hash dengan constant-time digest ───────────────────
|
||||||
)
|
incoming_hash = hashlib.sha256(csrf_token_header.encode()).hexdigest()
|
||||||
|
if not secrets.compare_digest(incoming_hash, stored_token_hash):
|
||||||
# 4. Proceed with the request
|
log.warning(
|
||||||
response: Response = await call_next(request)
|
"[CSRF] Token mismatch | user_id=%s | path=%s | IP: %s",
|
||||||
|
user_id,
|
||||||
# 5. Ensure the cookie is set if missing (e.g., on first GET request)
|
current_path,
|
||||||
if not request.cookies.get(self.cookie_name):
|
request.client.host if request.client else "unknown",
|
||||||
new_token = secrets.token_urlsafe(32)
|
)
|
||||||
response.set_cookie(
|
_redis_auth.delete(redis_key)
|
||||||
key=self.cookie_name,
|
raise HTTPException(status_code=403, detail="CSRF token verification failed")
|
||||||
value=new_token,
|
|
||||||
httponly=False, # Set to False so frontend JS can read it for Double Submit
|
# ── Step 6: hapus token (one-time use) dan lanjutkan ─────────────────────
|
||||||
samesite="lax",
|
_redis_auth.delete(redis_key)
|
||||||
secure=True, # Recommended for production with HTTPS
|
log.info(
|
||||||
path="/"
|
"[CSRF] Verified & deleted | user_id=%s | path=%s",
|
||||||
)
|
user_id,
|
||||||
|
current_path,
|
||||||
return response
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Loading…
Reference in New Issue