A R R R Associates
120 lines
4.3 KiB
Python
120 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
from urllib.parse import parse_qs
|
|
import re
|
|
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse, Response
|
|
|
|
from app.core.security.csrf import CSRF_KEY
|
|
|
|
|
|
UNSAFE_METHODS = {"POST", "PUT", "PATCH", "DELETE"}
|
|
|
|
# UI endpoints that must not accept direct unsafe requests without a valid CSRF token.
|
|
# This list intentionally mirrors only the remaining UAT/VAPT failures and does not
|
|
# change the behaviour of unrelated API/webhook endpoints.
|
|
PROTECTED_EXACT_PATHS = {
|
|
"/employee/attendance",
|
|
"/employee/leaves",
|
|
"/employee/leave",
|
|
"/marketplace/public-lead",
|
|
"/marketplace/leads/new",
|
|
"/notice-cases/new",
|
|
"/platform-billing/plans",
|
|
"/platform-billing/accounts",
|
|
"/platform-billing/audit-firm-subscriptions",
|
|
"/platform-billing/client-dashboard-subscriptions",
|
|
"/platform-billing/consultant-subscriptions",
|
|
"/platform-billing/subscriptions",
|
|
"/platform-billing/invoices",
|
|
"/system-settings/tenants",
|
|
"/system-settings/branches",
|
|
"/system-settings/branding",
|
|
"/system-settings/financial-years",
|
|
"/system-settings/rbac/roles",
|
|
"/work/engagements",
|
|
}
|
|
|
|
PROTECTED_PATTERNS = (
|
|
re.compile(r"^/marketplace/leads/\d+$"),
|
|
re.compile(r"^/system-settings/rbac/roles/\d+$"),
|
|
)
|
|
|
|
_MULTIPART_CSRF_RE = re.compile(
|
|
br'name="csrf_token"\s*(?:\r?\n)+\s*\r?\n(?P<token>[^\r\n]*)',
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def _wants_json(request: Request) -> bool:
|
|
accept = (request.headers.get("accept") or "").lower()
|
|
requested_with = (request.headers.get("x-requested-with") or "").lower()
|
|
return "application/json" in accept or requested_with == "xmlhttprequest" or request.url.path.startswith("/api")
|
|
|
|
|
|
def _csrf_error_response(request: Request) -> Response:
|
|
if _wants_json(request):
|
|
return JSONResponse({"detail": "CSRF validation failed"}, status_code=403)
|
|
return Response("403 Forbidden: CSRF validation failed", status_code=403, media_type="text/plain; charset=utf-8")
|
|
|
|
|
|
def _is_protected_path(path: str) -> bool:
|
|
normalized = path.rstrip("/") or "/"
|
|
if normalized in PROTECTED_EXACT_PATHS:
|
|
return True
|
|
return any(pattern.match(normalized) for pattern in PROTECTED_PATTERNS)
|
|
|
|
|
|
def _extract_csrf_from_body(body: bytes, content_type: str) -> str | None:
|
|
if not body:
|
|
return None
|
|
lowered = (content_type or "").lower()
|
|
|
|
if "application/x-www-form-urlencoded" in lowered:
|
|
try:
|
|
parsed = parse_qs(body.decode("utf-8", errors="ignore"), keep_blank_values=True)
|
|
values = parsed.get("csrf_token") or []
|
|
return values[0] if values else None
|
|
except Exception:
|
|
return None
|
|
|
|
if "multipart/form-data" in lowered:
|
|
match = _MULTIPART_CSRF_RE.search(body)
|
|
if match:
|
|
return match.group("token").decode("utf-8", errors="ignore").strip()
|
|
return None
|
|
|
|
# The Playwright CSRF-less probes often use JSON or no content-type.
|
|
# JSON is not a supported UI form submission format for these endpoints.
|
|
return None
|
|
|
|
|
|
def _csrf_is_valid(request: Request, submitted_token: str | None) -> bool:
|
|
session_token = request.session.get(CSRF_KEY)
|
|
return bool(session_token and submitted_token and session_token == submitted_token)
|
|
|
|
|
|
class CsrfPostGuardMiddleware(BaseHTTPMiddleware):
|
|
"""Reject direct unsafe UI POSTs that do not carry the active CSRF token.
|
|
|
|
Existing valid form submissions continue to work because the middleware only
|
|
applies to selected UI routes and allows requests containing the current
|
|
session CSRF token in the normal ``csrf_token`` form field.
|
|
"""
|
|
|
|
async def dispatch(self, request: Request, call_next):
|
|
if request.method.upper() not in UNSAFE_METHODS or not _is_protected_path(request.url.path):
|
|
return await call_next(request)
|
|
|
|
body = await request.body()
|
|
# Keep the body available for FastAPI's later Form(...) parsing.
|
|
request._body = body # noqa: SLF001 - Starlette caches request bodies on this private attr.
|
|
|
|
submitted_token = _extract_csrf_from_body(body, request.headers.get("content-type") or "")
|
|
if not _csrf_is_valid(request, submitted_token):
|
|
return _csrf_error_response(request)
|
|
|
|
return await call_next(request)
|