Add missing csrf.py module required by router imports

The opi.utils.csrf module is imported by router.py and
router_self_service.py but was never committed, causing
pyright reportMissingImports errors in CI.

Author: anneschuth

operations-manager/python/opi/utils/csrf.py

@@ -0,0 +1,168 @@
+"""
+CSRF (Cross-Site Request Forgery) protection utilities.
+
+This module provides CSRF token generation and validation to protect against
+cross-site request forgery attacks. It works alongside Origin/Referer header
+validation to provide defense in depth.
+
+Usage:
+    # In route handler that renders forms:
+    csrf_token = ensure_csrf_token(request)
+    return templates.TemplateResponse("form.html", {"csrf_token": csrf_token})
+
+    # In route handler that processes form submission:
+    validate_csrf_token(request)  # Raises HTTPException if invalid
+"""
+
+import logging
+import secrets
+from typing import Any
+
+from fastapi import HTTPException, Request
+
+logger = logging.getLogger(__name__)
+
+# CSRF token configuration
+CSRF_TOKEN_LENGTH = 32  # 256 bits of entropy
+CSRF_TOKEN_SESSION_KEY = "_csrf_token"
+CSRF_TOKEN_HEADER_NAME = "X-CSRF-Token"
+CSRF_TOKEN_FORM_FIELD = "csrf_token"
+
+
+def generate_csrf_token() -> str:
+    """Generate a cryptographically secure CSRF token.
+
+    Returns:
+        A hex-encoded random token with CSRF_TOKEN_LENGTH bytes of entropy.
+    """
+    return secrets.token_hex(CSRF_TOKEN_LENGTH)
+
+
+def get_csrf_token_from_session(request: Request) -> str | None:
+    """Get the CSRF token from the session if it exists.
+
+    Args:
+        request: The FastAPI/Starlette request object
+
+    Returns:
+        The CSRF token string or None if not set
+    """
+    return request.session.get(CSRF_TOKEN_SESSION_KEY)
+
+
+def ensure_csrf_token(request: Request) -> str:
+    """Ensure a CSRF token exists in the session, creating one if needed.
+
+    This should be called when rendering pages that contain forms.
+
+    Args:
+        request: The FastAPI/Starlette request object
+
+    Returns:
+        The CSRF token string (existing or newly generated)
+    """
+    token = get_csrf_token_from_session(request)
+    if not token:
+        token = generate_csrf_token()
+        request.session[CSRF_TOKEN_SESSION_KEY] = token
+        logger.debug("Generated new CSRF token for session")
+    return token
+
+
+def get_csrf_token_from_request(request: Request, form_data: dict[str, Any] | None = None) -> str | None:
+    """Extract the CSRF token from the request (header or form data).
+
+    Checks in order:
+    1. X-CSRF-Token header (for AJAX requests)
+    2. csrf_token form field (for regular form submissions)
+
+    Args:
+        request: The FastAPI/Starlette request object
+        form_data: Optional pre-parsed form data dictionary
+
+    Returns:
+        The CSRF token string or None if not found
+    """
+    # Check header first (for AJAX requests)
+    header_token = request.headers.get(CSRF_TOKEN_HEADER_NAME)
+    if header_token:
+        return header_token.strip()
+
+    # Check form data if provided
+    if form_data and CSRF_TOKEN_FORM_FIELD in form_data:
+        form_token = form_data.get(CSRF_TOKEN_FORM_FIELD)
+        if form_token:
+            return str(form_token).strip()
+
+    return None
+
+
+def validate_csrf_token(
+    request: Request,
+    form_data: dict[str, Any] | None = None,
+    raise_on_missing: bool = True,
+) -> bool:
+    """Validate the CSRF token from the request against the session token.
+
+    Args:
+        request: The FastAPI/Starlette request object
+        form_data: Optional pre-parsed form data dictionary
+        raise_on_missing: If True, raise HTTPException when token is missing
+
+    Returns:
+        True if token is valid
+
+    Raises:
+        HTTPException: If token is missing (when raise_on_missing=True) or invalid
+    """
+    session_token = get_csrf_token_from_session(request)
+    request_token = get_csrf_token_from_request(request, form_data)
+
+    # Check if session has a token
+    if not session_token:
+        logger.warning("CSRF validation failed: No token in session")
+        if raise_on_missing:
+            raise HTTPException(
+                status_code=403,
+                detail="CSRF validation failed: session expired or invalid"
+            )
+        return False
+
+    # Check if request includes a token
+    if not request_token:
+        logger.warning("CSRF validation failed: No token in request")
+        if raise_on_missing:
+            raise HTTPException(
+                status_code=403,
+                detail="CSRF validation failed: missing token"
+            )
+        return False
+
+    # Constant-time comparison to prevent timing attacks
+    if not secrets.compare_digest(session_token, request_token):
+        logger.warning("CSRF validation failed: Token mismatch")
+        raise HTTPException(
+            status_code=403,
+            detail="CSRF validation failed: invalid token"
+        )
+
+    logger.debug("CSRF token validated successfully")
+    return True
+
+
+def rotate_csrf_token(request: Request) -> str:
+    """Rotate the CSRF token after a successful state-changing operation.
+
+    This can be called after sensitive operations to prevent token reuse.
+    It's optional but recommended for high-security operations.
+
+    Args:
+        request: The FastAPI/Starlette request object
+
+    Returns:
+        The new CSRF token
+    """
+    new_token = generate_csrf_token()
+    request.session[CSRF_TOKEN_SESSION_KEY] = new_token
+    logger.debug("Rotated CSRF token")
+    return new_token