From 5a17f89283dba8ab114f60d80dba2f18d00db543 Mon Sep 17 00:00:00 2001 From: A R R R Associates Date: Fri, 26 Jun 2026 15:41:38 +0530 Subject: [PATCH] Final fix CSRF unsafe POST guard for remaining UAT VAPT failures --- app/core/middleware/csrf_guard.py | 119 ++++++++++++++++++++++++++++++ app/main.py | 4 + 2 files changed, 123 insertions(+) create mode 100644 app/core/middleware/csrf_guard.py diff --git a/app/core/middleware/csrf_guard.py b/app/core/middleware/csrf_guard.py new file mode 100644 index 0000000..080b369 --- /dev/null +++ b/app/core/middleware/csrf_guard.py @@ -0,0 +1,119 @@ +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[^\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) diff --git a/app/main.py b/app/main.py index c6085c5..08fa55f 100644 --- a/app/main.py +++ b/app/main.py @@ -6,6 +6,7 @@ from app.core.settings import get_settings from app.core.middleware.context import ContextResolveMiddleware from app.core.middleware.domain_resolver import DomainResolverMiddleware from app.core.middleware.security_headers import SecurityHeadersMiddleware +from app.core.middleware.csrf_guard import CsrfPostGuardMiddleware from app.core.startup import on_startup from app.core.api import api_router from app.core.http_responses import auth_required_response, forbidden_response, not_found_response, wants_json @@ -45,6 +46,9 @@ def create_app() -> FastAPI: # Phase 7T.2: added after context so it resolves the request host before # context-aware middleware/routes need tenant/branch/domain state. app.add_middleware(DomainResolverMiddleware) + # Runs after SessionMiddleware has populated request.session and before routes. + # It only guards selected UI unsafe POST endpoints and does not change valid forms. + app.add_middleware(CsrfPostGuardMiddleware) # SessionMiddleware is added last so it is available to downstream # middleware/routes in Starlette's middleware execution order. app.add_middleware(