mirror of
https://github.com/zvx-echo6/central.git
synced 2026-05-21 18:14:44 +02:00
fix(csrf): replace fastapi-csrf-protect with session-bound CSRF
Fixes CSRF race condition where every GET rotated the CSRF token, causing POST failures when users had multiple tabs or slow connections. Changes: - Remove fastapi-csrf-protect dependency - Add session-bound CSRF tokens stored in config.sessions table - Add pre-auth CSRF for unauthenticated routes (/login, /setup/operator) - Add csrf.py module for pre-auth token generation/validation - Update routes to use new CSRF token handling - Add migration 013 to add csrf_token column to sessions The session-bound approach ensures CSRF tokens remain stable for the duration of a session, eliminating the race condition. Note: Route tests (test_wizard.py, test_adapters.py, etc.) need refactoring to mock get_settings() instead of CsrfProtect dependency. Core auth/CSRF handler tests pass (74 tests). Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Matt Johnson
parent
84044a4d45
commit
c317c9ab01
11 changed files with 410 additions and 208 deletions
|
|
@ -16,7 +16,6 @@ dependencies = [
|
|||
"asyncpg>=0.31.0",
|
||||
"cloudevents>=2.0.0",
|
||||
"cryptography>=44.0.0",
|
||||
"fastapi-csrf-protect>=0.4.0",
|
||||
"fastapi>=0.115.0",
|
||||
"jinja2>=3.1.6",
|
||||
"nats-py>=2.14.0",
|
||||
|
|
|
|||
9
sql/migrations/013_add_session_csrf_token.sql
Normal file
9
sql/migrations/013_add_session_csrf_token.sql
Normal file
|
|
@ -0,0 +1,9 @@
|
|||
-- Add CSRF token column to sessions table
|
||||
-- Session-bound CSRF tokens prevent race conditions from cookie rotation
|
||||
|
||||
ALTER TABLE config.sessions
|
||||
ADD COLUMN csrf_token TEXT NOT NULL
|
||||
DEFAULT encode(gen_random_bytes(32), 'hex');
|
||||
|
||||
-- Comment
|
||||
COMMENT ON COLUMN config.sessions.csrf_token IS 'Session-bound CSRF token for synchronizer token pattern';
|
||||
|
|
@ -29,23 +29,6 @@ _cleanup_task: asyncio.Task | None = None
|
|||
_app: FastAPI | None = None
|
||||
|
||||
|
||||
def _configure_csrf() -> None:
|
||||
"""Configure CSRF protection. Must be called before app starts."""
|
||||
from fastapi_csrf_protect import CsrfProtect
|
||||
from pydantic import BaseModel
|
||||
from central.bootstrap_config import get_settings
|
||||
|
||||
class CsrfSettings(BaseModel):
|
||||
secret_key: str
|
||||
token_location: str = "body"
|
||||
token_key: str = "csrf_token"
|
||||
|
||||
@CsrfProtect.load_config
|
||||
def get_csrf_config():
|
||||
settings = get_settings()
|
||||
return CsrfSettings(secret_key=settings.csrf_secret)
|
||||
|
||||
|
||||
async def _session_cleanup_loop() -> None:
|
||||
"""Periodically clean up expired sessions."""
|
||||
global _shutdown_event
|
||||
|
|
@ -117,9 +100,6 @@ def _create_app() -> FastAPI:
|
|||
from central.gui.middleware import SessionMiddleware, SetupGateMiddleware
|
||||
from central.gui.routes import router
|
||||
|
||||
# Configure CSRF before creating app
|
||||
_configure_csrf()
|
||||
|
||||
app = FastAPI(
|
||||
title="Central GUI",
|
||||
lifespan=lifespan,
|
||||
|
|
@ -137,16 +117,19 @@ def _create_app() -> FastAPI:
|
|||
app.include_router(router)
|
||||
|
||||
# CSRF exception handler - return friendly error instead of 500
|
||||
from fastapi_csrf_protect.exceptions import CsrfProtectError
|
||||
from central.gui.auth import CsrfValidationError
|
||||
from central.gui.csrf import generate_pre_auth_csrf, set_pre_auth_csrf_cookie
|
||||
from central.bootstrap_config import get_settings
|
||||
from fastapi.responses import RedirectResponse
|
||||
|
||||
@app.exception_handler(CsrfProtectError)
|
||||
async def csrf_exception_handler(request, exc: CsrfProtectError):
|
||||
from fastapi_csrf_protect import CsrfProtect
|
||||
@app.exception_handler(CsrfValidationError)
|
||||
async def csrf_exception_handler(request, exc: CsrfValidationError):
|
||||
from central.gui.db import get_pool
|
||||
|
||||
csrf_protect = CsrfProtect()
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
settings = get_settings()
|
||||
# For pre-auth paths, generate a new pre-auth token
|
||||
# For session paths, we'll just show the error (session token stays valid)
|
||||
csrf_token, signed_token = generate_pre_auth_csrf(settings.csrf_secret)
|
||||
error_msg = "Your session expired. Please try again."
|
||||
|
||||
if request.url.path == "/login":
|
||||
|
|
@ -155,7 +138,7 @@ def _create_app() -> FastAPI:
|
|||
name="login.html",
|
||||
context={"csrf_token": csrf_token, "error": error_msg},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
set_pre_auth_csrf_cookie(response, signed_token)
|
||||
return response
|
||||
|
||||
elif request.url.path == "/setup":
|
||||
|
|
@ -168,7 +151,7 @@ def _create_app() -> FastAPI:
|
|||
name="setup_operator.html",
|
||||
context={"csrf_token": csrf_token, "error": error_msg, "form_data": None},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
set_pre_auth_csrf_cookie(response, signed_token)
|
||||
return response
|
||||
|
||||
elif request.url.path == "/setup/system":
|
||||
|
|
@ -201,7 +184,7 @@ def _create_app() -> FastAPI:
|
|||
"system": system,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
set_pre_auth_csrf_cookie(response, signed_token)
|
||||
return response
|
||||
|
||||
elif request.url.path == "/setup/keys":
|
||||
|
|
@ -228,7 +211,7 @@ def _create_app() -> FastAPI:
|
|||
"error": error_msg,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
set_pre_auth_csrf_cookie(response, signed_token)
|
||||
return response
|
||||
|
||||
elif request.url.path == "/setup/adapters":
|
||||
|
|
@ -283,7 +266,7 @@ def _create_app() -> FastAPI:
|
|||
"form_data": None,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
set_pre_auth_csrf_cookie(response, signed_token)
|
||||
return response
|
||||
|
||||
elif request.url.path == "/setup/finish":
|
||||
|
|
@ -323,7 +306,7 @@ def _create_app() -> FastAPI:
|
|||
"error": error_msg,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
set_pre_auth_csrf_cookie(response, signed_token)
|
||||
return response
|
||||
|
||||
elif request.url.path == "/logout":
|
||||
|
|
@ -335,7 +318,7 @@ def _create_app() -> FastAPI:
|
|||
name="change_password.html",
|
||||
context={"csrf_token": csrf_token, "error": error_msg},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
set_pre_auth_csrf_cookie(response, signed_token)
|
||||
return response
|
||||
|
||||
elif request.url.path.startswith("/adapters/"):
|
||||
|
|
|
|||
|
|
@ -12,6 +12,11 @@ from argon2.exceptions import VerifyMismatchError
|
|||
_hasher = PasswordHasher()
|
||||
|
||||
|
||||
class CsrfValidationError(Exception):
|
||||
"""Raised when CSRF token validation fails."""
|
||||
pass
|
||||
|
||||
|
||||
@dataclass
|
||||
class Operator:
|
||||
"""Operator account."""
|
||||
|
|
@ -46,39 +51,46 @@ def generate_token() -> str:
|
|||
return secrets.token_urlsafe(32)
|
||||
|
||||
|
||||
def generate_csrf_token() -> str:
|
||||
"""Generate a cryptographically secure CSRF token."""
|
||||
return secrets.token_hex(32)
|
||||
|
||||
|
||||
async def create_session(
|
||||
conn: Any, # asyncpg.Connection
|
||||
operator_id: int,
|
||||
lifetime_days: int,
|
||||
) -> tuple[str, datetime]:
|
||||
) -> tuple[str, datetime, str]:
|
||||
"""Create a new session for an operator.
|
||||
|
||||
Returns (token, expires_at).
|
||||
Returns (token, expires_at, csrf_token).
|
||||
"""
|
||||
token = generate_token()
|
||||
csrf_token = generate_csrf_token()
|
||||
expires_at = datetime.now(timezone.utc) + timedelta(days=lifetime_days)
|
||||
|
||||
await conn.execute(
|
||||
"""
|
||||
INSERT INTO config.sessions (token, operator_id, expires_at)
|
||||
VALUES ($1, $2, $3)
|
||||
INSERT INTO config.sessions (token, operator_id, expires_at, csrf_token)
|
||||
VALUES ($1, $2, $3, $4)
|
||||
""",
|
||||
token,
|
||||
operator_id,
|
||||
expires_at,
|
||||
csrf_token,
|
||||
)
|
||||
|
||||
return token, expires_at
|
||||
return token, expires_at, csrf_token
|
||||
|
||||
|
||||
async def get_session(conn: Any, token: str) -> Operator | None:
|
||||
"""Look up a session and return the associated operator.
|
||||
async def get_session(conn: Any, token: str) -> tuple[Operator, str] | None:
|
||||
"""Look up a session and return the associated operator and csrf_token.
|
||||
|
||||
Returns None if token is invalid or expired.
|
||||
Returns (Operator, csrf_token) or None if token is invalid or expired.
|
||||
"""
|
||||
row = await conn.fetchrow(
|
||||
"""
|
||||
SELECT o.id, o.username, o.created_at, o.password_changed_at
|
||||
SELECT o.id, o.username, o.created_at, o.password_changed_at, s.csrf_token
|
||||
FROM config.sessions s
|
||||
JOIN config.operators o ON s.operator_id = o.id
|
||||
WHERE s.token = $1 AND s.expires_at > now()
|
||||
|
|
@ -89,13 +101,15 @@ async def get_session(conn: Any, token: str) -> Operator | None:
|
|||
if row is None:
|
||||
return None
|
||||
|
||||
return Operator(
|
||||
operator = Operator(
|
||||
id=row["id"],
|
||||
username=row["username"],
|
||||
created_at=row["created_at"],
|
||||
password_changed_at=row.get("password_changed_at"),
|
||||
)
|
||||
|
||||
return operator, row["csrf_token"]
|
||||
|
||||
|
||||
async def delete_session(conn: Any, token: str) -> None:
|
||||
"""Delete a session."""
|
||||
|
|
|
|||
72
src/central/gui/csrf.py
Normal file
72
src/central/gui/csrf.py
Normal file
|
|
@ -0,0 +1,72 @@
|
|||
"""Pre-auth CSRF protection for login and setup/operator pages.
|
||||
|
||||
These routes cannot use session-bound CSRF because no session exists yet.
|
||||
Uses a simple cookie-based pattern with short-lived tokens.
|
||||
"""
|
||||
|
||||
import secrets
|
||||
from typing import Optional
|
||||
|
||||
from itsdangerous import URLSafeTimedSerializer, BadSignature, SignatureExpired
|
||||
from starlette.requests import Request
|
||||
from starlette.responses import Response
|
||||
|
||||
|
||||
# 10 minute max age for pre-auth CSRF tokens
|
||||
PRE_AUTH_CSRF_MAX_AGE = 600
|
||||
PRE_AUTH_CSRF_COOKIE = "central_preauth_csrf"
|
||||
|
||||
|
||||
def _get_serializer(secret_key: str) -> URLSafeTimedSerializer:
|
||||
"""Get a timed serializer for CSRF tokens."""
|
||||
return URLSafeTimedSerializer(secret_key, salt="preauth-csrf")
|
||||
|
||||
|
||||
def generate_pre_auth_csrf(secret_key: str) -> tuple[str, str]:
|
||||
"""Generate a pre-auth CSRF token pair.
|
||||
|
||||
Returns (plain_token, signed_token).
|
||||
The plain_token goes in the form, signed_token goes in the cookie.
|
||||
"""
|
||||
plain_token = secrets.token_hex(32)
|
||||
serializer = _get_serializer(secret_key)
|
||||
signed_token = serializer.dumps(plain_token)
|
||||
return plain_token, signed_token
|
||||
|
||||
|
||||
def set_pre_auth_csrf_cookie(response: Response, signed_token: str) -> None:
|
||||
"""Set the pre-auth CSRF cookie on a response."""
|
||||
response.set_cookie(
|
||||
PRE_AUTH_CSRF_COOKIE,
|
||||
signed_token,
|
||||
max_age=PRE_AUTH_CSRF_MAX_AGE,
|
||||
path="/",
|
||||
httponly=True,
|
||||
samesite="lax",
|
||||
)
|
||||
|
||||
|
||||
def validate_pre_auth_csrf(
|
||||
request: Request,
|
||||
form_token: str,
|
||||
secret_key: str,
|
||||
) -> bool:
|
||||
"""Validate a pre-auth CSRF token.
|
||||
|
||||
Returns True if valid, False otherwise.
|
||||
"""
|
||||
cookie_value = request.cookies.get(PRE_AUTH_CSRF_COOKIE)
|
||||
if not cookie_value or not form_token:
|
||||
return False
|
||||
|
||||
serializer = _get_serializer(secret_key)
|
||||
try:
|
||||
expected_token = serializer.loads(cookie_value, max_age=PRE_AUTH_CSRF_MAX_AGE)
|
||||
return secrets.compare_digest(form_token, expected_token)
|
||||
except (BadSignature, SignatureExpired):
|
||||
return False
|
||||
|
||||
|
||||
def unset_pre_auth_csrf_cookie(response: Response) -> None:
|
||||
"""Remove the pre-auth CSRF cookie."""
|
||||
response.delete_cookie(PRE_AUTH_CSRF_COOKIE, path="/")
|
||||
|
|
@ -113,13 +113,14 @@ class SetupGateMiddleware(BaseHTTPMiddleware):
|
|||
|
||||
|
||||
class SessionMiddleware(BaseHTTPMiddleware):
|
||||
"""Load session from cookie and attach operator to request.state."""
|
||||
"""Load session from cookie and attach operator + csrf_token to request.state."""
|
||||
|
||||
async def dispatch(self, request: Request, call_next) -> Response:
|
||||
path = request.url.path
|
||||
|
||||
# Initialize operator to None
|
||||
# Initialize state
|
||||
request.state.operator = None
|
||||
request.state.csrf_token = None
|
||||
|
||||
# Try to load session from cookie
|
||||
session_token = request.cookies.get("central_session")
|
||||
|
|
@ -128,11 +129,15 @@ class SessionMiddleware(BaseHTTPMiddleware):
|
|||
if pool is not None:
|
||||
try:
|
||||
async with pool.acquire() as conn:
|
||||
operator = await get_session(conn, session_token)
|
||||
request.state.operator = operator
|
||||
result = await get_session(conn, session_token)
|
||||
if result is not None:
|
||||
operator, csrf_token = result
|
||||
request.state.operator = operator
|
||||
request.state.csrf_token = csrf_token
|
||||
except Exception:
|
||||
logger.warning("Failed to load session", exc_info=True)
|
||||
request.state.operator = None
|
||||
request.state.csrf_token = None
|
||||
|
||||
# Check if auth is required
|
||||
if not _is_exempt(path, AUTH_EXEMPT_PATHS, AUTH_EXEMPT_PREFIXES):
|
||||
|
|
|
|||
|
|
@ -9,9 +9,16 @@ logger = logging.getLogger("central.gui.routes")
|
|||
|
||||
from fastapi import APIRouter, Depends, Form, Request
|
||||
from fastapi.responses import HTMLResponse, RedirectResponse, Response
|
||||
from fastapi_csrf_protect import CsrfProtect
|
||||
from central.bootstrap_config import get_settings
|
||||
from central.gui.csrf import (
|
||||
generate_pre_auth_csrf,
|
||||
set_pre_auth_csrf_cookie,
|
||||
validate_pre_auth_csrf,
|
||||
unset_pre_auth_csrf_cookie,
|
||||
)
|
||||
|
||||
from central.gui.auth import (
|
||||
CsrfValidationError,
|
||||
create_session,
|
||||
delete_session,
|
||||
hash_password,
|
||||
|
|
@ -103,17 +110,16 @@ async def health() -> dict:
|
|||
|
||||
|
||||
@router.get("/", response_class=HTMLResponse)
|
||||
async def index(request: Request, csrf_protect: CsrfProtect = Depends()) -> HTMLResponse:
|
||||
async def index(request: Request) -> HTMLResponse:
|
||||
"""Render the index page."""
|
||||
templates = _get_templates()
|
||||
operator = getattr(request.state, "operator", None)
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="index.html",
|
||||
context={"operator": operator, "csrf_token": csrf_token},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
|
||||
|
|
@ -262,12 +268,12 @@ async def dashboard_polls(request: Request) -> HTMLResponse:
|
|||
@router.get("/setup/operator", response_class=HTMLResponse)
|
||||
async def setup_operator_form(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
) -> HTMLResponse:
|
||||
"""Render the setup operator form (step 1)."""
|
||||
templates = _get_templates()
|
||||
pool = get_pool()
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
settings = get_settings()
|
||||
csrf_token, signed_token = generate_pre_auth_csrf(settings.csrf_secret)
|
||||
|
||||
# Check if operator already exists
|
||||
existing_operator = None
|
||||
|
|
@ -288,7 +294,7 @@ async def setup_operator_form(
|
|||
"existing_operator": existing_operator,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
set_pre_auth_csrf_cookie(response, signed_token)
|
||||
return response
|
||||
|
||||
|
||||
|
|
@ -298,14 +304,18 @@ async def setup_operator_submit(
|
|||
username: str = Form(...),
|
||||
password: str = Form(...),
|
||||
confirm_password: str = Form(...),
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> Response:
|
||||
"""Process the setup operator form (step 1)."""
|
||||
templates = _get_templates()
|
||||
pool = get_pool()
|
||||
|
||||
# Validate CSRF
|
||||
await csrf_protect.validate_csrf(request)
|
||||
settings = get_settings()
|
||||
form = await request.form()
|
||||
form_csrf = form.get("csrf_token", "")
|
||||
if not validate_pre_auth_csrf(request, form_csrf, settings.csrf_secret):
|
||||
raise CsrfValidationError("Invalid CSRF token")
|
||||
|
||||
# Check if operator already exists (single-operator-per-install design)
|
||||
async with pool.acquire() as conn:
|
||||
|
|
@ -315,7 +325,7 @@ async def setup_operator_submit(
|
|||
existing = await conn.fetchrow(
|
||||
"SELECT username FROM config.operators ORDER BY id LIMIT 1"
|
||||
)
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="setup_operator.html",
|
||||
|
|
@ -326,7 +336,6 @@ async def setup_operator_submit(
|
|||
"existing_operator": {"username": existing["username"]},
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
# Validate input
|
||||
|
|
@ -340,7 +349,7 @@ async def setup_operator_submit(
|
|||
error = str(e)
|
||||
|
||||
if error:
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="setup_operator.html",
|
||||
|
|
@ -352,7 +361,6 @@ async def setup_operator_submit(
|
|||
},
|
||||
status_code=200,
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
# Create operator
|
||||
|
|
@ -384,7 +392,7 @@ async def setup_operator_submit(
|
|||
lifetime_days = sysrow["session_lifetime_days"] if sysrow else 90
|
||||
|
||||
# Create session
|
||||
token, expires_at = await create_session(conn, operator_id, lifetime_days)
|
||||
token, expires_at, _ = await create_session(conn, operator_id, lifetime_days)
|
||||
|
||||
# Redirect to next step with session cookie
|
||||
response = RedirectResponse(url="/setup/system", status_code=302)
|
||||
|
|
@ -395,7 +403,7 @@ async def setup_operator_submit(
|
|||
@router.get("/setup/system", response_class=HTMLResponse)
|
||||
async def setup_system_form(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> HTMLResponse:
|
||||
"""Render the system settings form (step 2)."""
|
||||
# Require authentication for this step
|
||||
|
|
@ -415,7 +423,7 @@ async def setup_system_form(
|
|||
"map_attribution": row["map_attribution"] if row else "© OpenStreetMap contributors",
|
||||
}
|
||||
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="setup_system.html",
|
||||
|
|
@ -427,14 +435,13 @@ async def setup_system_form(
|
|||
"system": system,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
|
||||
@router.post("/setup/system")
|
||||
async def setup_system_submit(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> Response:
|
||||
"""Process the system settings form (step 2)."""
|
||||
# Require authentication for this step
|
||||
|
|
@ -445,7 +452,10 @@ async def setup_system_submit(
|
|||
templates = _get_templates()
|
||||
pool = get_pool()
|
||||
|
||||
await csrf_protect.validate_csrf(request)
|
||||
form = await request.form()
|
||||
form_csrf = form.get("csrf_token", "")
|
||||
if not form_csrf or form_csrf != request.state.csrf_token:
|
||||
raise CsrfValidationError("Invalid CSRF token")
|
||||
|
||||
form = await request.form()
|
||||
map_tile_url = form.get("map_tile_url", "").strip()
|
||||
|
|
@ -478,7 +488,7 @@ async def setup_system_submit(
|
|||
"map_attribution": row["map_attribution"] if row else "",
|
||||
}
|
||||
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="setup_system.html",
|
||||
|
|
@ -491,7 +501,6 @@ async def setup_system_submit(
|
|||
},
|
||||
status_code=200,
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
# Get current values for audit
|
||||
|
|
@ -530,7 +539,7 @@ async def setup_system_submit(
|
|||
@router.get("/setup/keys", response_class=HTMLResponse)
|
||||
async def setup_keys_form(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> HTMLResponse:
|
||||
"""Render the API keys form (step 3)."""
|
||||
# Require authentication for this step
|
||||
|
|
@ -549,7 +558,7 @@ async def setup_keys_form(
|
|||
)
|
||||
keys = [{"alias": row["alias"], "created_at": row["created_at"]} for row in rows]
|
||||
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="setup_keys.html",
|
||||
|
|
@ -561,14 +570,13 @@ async def setup_keys_form(
|
|||
"success": None,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
|
||||
@router.post("/setup/keys")
|
||||
async def setup_keys_submit(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> Response:
|
||||
"""Process the API keys form (step 3)."""
|
||||
# Require authentication for this step
|
||||
|
|
@ -576,7 +584,10 @@ async def setup_keys_submit(
|
|||
if operator is None:
|
||||
return RedirectResponse(url="/setup/operator", status_code=302)
|
||||
|
||||
await csrf_protect.validate_csrf(request)
|
||||
form = await request.form()
|
||||
form_csrf = form.get("csrf_token", "")
|
||||
if not form_csrf or form_csrf != request.state.csrf_token:
|
||||
raise CsrfValidationError("Invalid CSRF token")
|
||||
|
||||
form = await request.form()
|
||||
action = form.get("action", "add")
|
||||
|
|
@ -627,7 +638,7 @@ async def setup_keys_submit(
|
|||
keys = [{"alias": row["alias"], "created_at": row["created_at"]} for row in keys]
|
||||
|
||||
if errors:
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="setup_keys.html",
|
||||
|
|
@ -640,7 +651,6 @@ async def setup_keys_submit(
|
|||
},
|
||||
status_code=200,
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
# Encrypt the key
|
||||
|
|
@ -674,7 +684,7 @@ async def setup_keys_submit(
|
|||
keys = [{"alias": row["alias"], "created_at": row["created_at"]} for row in keys]
|
||||
|
||||
# Re-render with success message
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="setup_keys.html",
|
||||
|
|
@ -686,14 +696,13 @@ async def setup_keys_submit(
|
|||
"success": f"API key '{alias}' added successfully.",
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
|
||||
@router.get("/setup/adapters", response_class=HTMLResponse)
|
||||
async def setup_adapters_form(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> HTMLResponse:
|
||||
"""Render the adapters configuration form (step 4)."""
|
||||
# Require authentication for this step
|
||||
|
|
@ -734,7 +743,7 @@ async def setup_adapters_form(
|
|||
tile_url = sys_row["map_tile_url"] if sys_row else "https://tile.openstreetmap.org/{z}/{x}/{y}.png"
|
||||
tile_attribution = sys_row["map_attribution"] if sys_row else "© OpenStreetMap contributors"
|
||||
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="setup_adapters.html",
|
||||
|
|
@ -751,14 +760,13 @@ async def setup_adapters_form(
|
|||
"form_data": None,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
|
||||
@router.post("/setup/adapters")
|
||||
async def setup_adapters_submit(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> Response:
|
||||
"""Process the adapters configuration form (step 4)."""
|
||||
# Require authentication for this step
|
||||
|
|
@ -769,7 +777,10 @@ async def setup_adapters_submit(
|
|||
templates = _get_templates()
|
||||
pool = get_pool()
|
||||
|
||||
await csrf_protect.validate_csrf(request)
|
||||
form = await request.form()
|
||||
form_csrf = form.get("csrf_token", "")
|
||||
if not form_csrf or form_csrf != request.state.csrf_token:
|
||||
raise CsrfValidationError("Invalid CSRF token")
|
||||
|
||||
form = await request.form()
|
||||
errors: dict[str, str] = {}
|
||||
|
|
@ -917,7 +928,7 @@ async def setup_adapters_submit(
|
|||
tile_url = sys_row["map_tile_url"] if sys_row else "https://tile.openstreetmap.org/{z}/{x}/{y}.png"
|
||||
tile_attribution = sys_row["map_attribution"] if sys_row else "© OpenStreetMap contributors"
|
||||
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="setup_adapters.html",
|
||||
|
|
@ -935,7 +946,6 @@ async def setup_adapters_submit(
|
|||
},
|
||||
status_code=200,
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
return RedirectResponse(url="/setup/finish", status_code=302)
|
||||
|
|
@ -944,7 +954,7 @@ async def setup_adapters_submit(
|
|||
@router.get("/setup/finish", response_class=HTMLResponse)
|
||||
async def setup_finish_form(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> HTMLResponse:
|
||||
"""Render the finish setup page (step 5)."""
|
||||
# Require authentication for this step
|
||||
|
|
@ -985,7 +995,7 @@ async def setup_finish_form(
|
|||
for row in rows
|
||||
]
|
||||
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="setup_finish.html",
|
||||
|
|
@ -997,14 +1007,13 @@ async def setup_finish_form(
|
|||
"adapters": adapters,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
|
||||
@router.post("/setup/finish")
|
||||
async def setup_finish_submit(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> Response:
|
||||
"""Complete the setup wizard."""
|
||||
# Require authentication for this step
|
||||
|
|
@ -1014,7 +1023,10 @@ async def setup_finish_submit(
|
|||
|
||||
pool = get_pool()
|
||||
|
||||
await csrf_protect.validate_csrf(request)
|
||||
form = await request.form()
|
||||
form_csrf = form.get("csrf_token", "")
|
||||
if not form_csrf or form_csrf != request.state.csrf_token:
|
||||
raise CsrfValidationError("Invalid CSRF token")
|
||||
|
||||
async with pool.acquire() as conn:
|
||||
# Mark setup complete
|
||||
|
|
@ -1036,17 +1048,17 @@ async def setup_finish_submit(
|
|||
@router.get("/login", response_class=HTMLResponse)
|
||||
async def login_form(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
) -> HTMLResponse:
|
||||
"""Render the login form."""
|
||||
templates = _get_templates()
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
settings = get_settings()
|
||||
csrf_token, signed_token = generate_pre_auth_csrf(settings.csrf_secret)
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="login.html",
|
||||
context={"csrf_token": csrf_token, "error": None},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
set_pre_auth_csrf_cookie(response, signed_token)
|
||||
return response
|
||||
|
||||
|
||||
|
|
@ -1055,14 +1067,18 @@ async def login_submit(
|
|||
request: Request,
|
||||
username: str = Form(...),
|
||||
password: str = Form(...),
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> Response:
|
||||
"""Process the login form."""
|
||||
templates = _get_templates()
|
||||
pool = get_pool()
|
||||
|
||||
# Validate CSRF
|
||||
await csrf_protect.validate_csrf(request)
|
||||
settings = get_settings()
|
||||
form = await request.form()
|
||||
form_csrf = form.get("csrf_token", "")
|
||||
if not validate_pre_auth_csrf(request, form_csrf, settings.csrf_secret):
|
||||
raise CsrfValidationError("Invalid CSRF token")
|
||||
|
||||
# Look up operator
|
||||
async with pool.acquire() as conn:
|
||||
|
|
@ -1078,27 +1094,25 @@ async def login_submit(
|
|||
if row is None:
|
||||
# Unknown user - still audit the attempt
|
||||
await write_audit(conn, AUTH_LOGIN_FAILED, target=username)
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="login.html",
|
||||
context={"csrf_token": csrf_token, "error": "Invalid username or password"},
|
||||
status_code=200,
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
# Verify password
|
||||
if not verify_password(password, row["password_hash"]):
|
||||
await write_audit(conn, AUTH_LOGIN_FAILED, operator_id=row["id"], target=username)
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="login.html",
|
||||
context={"csrf_token": csrf_token, "error": "Invalid username or password"},
|
||||
status_code=200,
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
# Get session lifetime
|
||||
|
|
@ -1108,7 +1122,7 @@ async def login_submit(
|
|||
lifetime_days = sysrow["session_lifetime_days"] if sysrow else 90
|
||||
|
||||
# Create session
|
||||
token, expires_at = await create_session(conn, row["id"], lifetime_days)
|
||||
token, expires_at, _ = await create_session(conn, row["id"], lifetime_days)
|
||||
|
||||
# Audit login
|
||||
await write_audit(conn, AUTH_LOGIN, operator_id=row["id"], target=username)
|
||||
|
|
@ -1122,13 +1136,16 @@ async def login_submit(
|
|||
@router.post("/logout")
|
||||
async def logout(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> Response:
|
||||
"""Log out the current user."""
|
||||
pool = get_pool()
|
||||
|
||||
# Validate CSRF
|
||||
await csrf_protect.validate_csrf(request)
|
||||
form = await request.form()
|
||||
form_csrf = form.get("csrf_token", "")
|
||||
if not form_csrf or form_csrf != request.state.csrf_token:
|
||||
raise CsrfValidationError("Invalid CSRF token")
|
||||
|
||||
# Get current session
|
||||
session_token = request.cookies.get("central_session")
|
||||
|
|
@ -1149,17 +1166,16 @@ async def logout(
|
|||
@router.get("/change-password", response_class=HTMLResponse)
|
||||
async def change_password_form(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> HTMLResponse:
|
||||
"""Render the change password form."""
|
||||
templates = _get_templates()
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="change_password.html",
|
||||
context={"csrf_token": csrf_token, "error": None, "success": False},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
|
||||
|
|
@ -1169,7 +1185,7 @@ async def change_password_submit(
|
|||
current_password: str = Form(...),
|
||||
new_password: str = Form(...),
|
||||
confirm_password: str = Form(...),
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> Response:
|
||||
"""Process the change password form."""
|
||||
templates = _get_templates()
|
||||
|
|
@ -1177,7 +1193,10 @@ async def change_password_submit(
|
|||
operator = request.state.operator
|
||||
|
||||
# Validate CSRF
|
||||
await csrf_protect.validate_csrf(request)
|
||||
form = await request.form()
|
||||
form_csrf = form.get("csrf_token", "")
|
||||
if not form_csrf or form_csrf != request.state.csrf_token:
|
||||
raise CsrfValidationError("Invalid CSRF token")
|
||||
|
||||
# Get current password hash
|
||||
async with pool.acquire() as conn:
|
||||
|
|
@ -1200,14 +1219,13 @@ async def change_password_submit(
|
|||
error = str(e)
|
||||
|
||||
if error:
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="change_password.html",
|
||||
context={"csrf_token": csrf_token, "error": error, "success": False},
|
||||
status_code=200,
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
# Update password
|
||||
|
|
@ -1242,7 +1260,7 @@ async def change_password_submit(
|
|||
@router.get("/adapters", response_class=HTMLResponse)
|
||||
async def adapters_list(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> HTMLResponse:
|
||||
"""List all adapters."""
|
||||
templates = _get_templates()
|
||||
|
|
@ -1270,7 +1288,7 @@ async def adapters_list(
|
|||
"updated_at": row["updated_at"],
|
||||
})
|
||||
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="adapters_list.html",
|
||||
|
|
@ -1280,7 +1298,6 @@ async def adapters_list(
|
|||
"adapters": adapters,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
|
||||
|
|
@ -1288,7 +1305,7 @@ async def adapters_list(
|
|||
async def adapters_edit_form(
|
||||
request: Request,
|
||||
name: str,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> Response:
|
||||
"""Render the adapter edit form."""
|
||||
templates = _get_templates()
|
||||
|
|
@ -1330,7 +1347,7 @@ async def adapters_edit_form(
|
|||
"updated_at": row["updated_at"],
|
||||
}
|
||||
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="adapters_edit.html",
|
||||
|
|
@ -1347,7 +1364,6 @@ async def adapters_edit_form(
|
|||
"tile_attribution": tile_attribution,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
|
||||
|
|
@ -1355,7 +1371,7 @@ async def adapters_edit_form(
|
|||
async def adapters_edit_submit(
|
||||
request: Request,
|
||||
name: str,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> Response:
|
||||
"""Process the adapter edit form."""
|
||||
templates = _get_templates()
|
||||
|
|
@ -1363,7 +1379,10 @@ async def adapters_edit_submit(
|
|||
operator = request.state.operator
|
||||
|
||||
# Validate CSRF
|
||||
await csrf_protect.validate_csrf(request)
|
||||
form = await request.form()
|
||||
form_csrf = form.get("csrf_token", "")
|
||||
if not form_csrf or form_csrf != request.state.csrf_token:
|
||||
raise CsrfValidationError("Invalid CSRF token")
|
||||
|
||||
# Parse form data
|
||||
form = await request.form()
|
||||
|
|
@ -1506,7 +1525,7 @@ async def adapters_edit_submit(
|
|||
tile_url = sys_row["map_tile_url"] if sys_row else "https://{s}.tile.openstreetmap.org/{z}/{x}/{y}.png"
|
||||
tile_attribution = sys_row["map_attribution"] if sys_row else "© OpenStreetMap contributors"
|
||||
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="adapters_edit.html",
|
||||
|
|
@ -1524,7 +1543,6 @@ async def adapters_edit_submit(
|
|||
},
|
||||
status_code=200,
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
# Build before state for audit
|
||||
|
|
@ -1575,7 +1593,7 @@ async def adapters_edit_submit(
|
|||
@router.get("/streams", response_class=HTMLResponse)
|
||||
async def streams_list(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> HTMLResponse:
|
||||
"""List all streams with live data."""
|
||||
from central.gui.nats import get_js
|
||||
|
|
@ -1658,7 +1676,7 @@ async def streams_list(
|
|||
|
||||
streams.append(stream_data)
|
||||
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="streams_list.html",
|
||||
|
|
@ -1668,7 +1686,6 @@ async def streams_list(
|
|||
"streams": streams,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
|
||||
|
|
@ -1676,7 +1693,7 @@ async def streams_list(
|
|||
async def streams_update(
|
||||
request: Request,
|
||||
name: str,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> Response:
|
||||
"""Update stream max_age_s."""
|
||||
from central.gui.nats import get_js
|
||||
|
|
@ -1686,7 +1703,10 @@ async def streams_update(
|
|||
operator = request.state.operator
|
||||
|
||||
# Validate CSRF
|
||||
await csrf_protect.validate_csrf(request)
|
||||
form = await request.form()
|
||||
form_csrf = form.get("csrf_token", "")
|
||||
if not form_csrf or form_csrf != request.state.csrf_token:
|
||||
raise CsrfValidationError("Invalid CSRF token")
|
||||
|
||||
form = await request.form()
|
||||
max_age_s_str = form.get("max_age_s", "").strip()
|
||||
|
|
@ -1755,7 +1775,7 @@ async def streams_update(
|
|||
|
||||
streams.append(stream_data)
|
||||
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="streams_list.html",
|
||||
|
|
@ -1766,7 +1786,6 @@ async def streams_update(
|
|||
"errors": errors,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
old_max_age_s = row["max_age_s"]
|
||||
|
|
@ -1802,7 +1821,7 @@ ALIAS_REGEX = re.compile(r'^[a-zA-Z0-9_]+$')
|
|||
@router.get("/api-keys", response_class=HTMLResponse)
|
||||
async def api_keys_list(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> HTMLResponse:
|
||||
"""List all API keys."""
|
||||
templates = _get_templates()
|
||||
|
|
@ -1838,7 +1857,7 @@ async def api_keys_list(
|
|||
"used_by": [a["name"] for a in adapters],
|
||||
})
|
||||
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="api_keys_list.html",
|
||||
|
|
@ -1848,20 +1867,19 @@ async def api_keys_list(
|
|||
"keys": keys,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
|
||||
@router.get("/api-keys/new", response_class=HTMLResponse)
|
||||
async def api_keys_new(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> HTMLResponse:
|
||||
"""Show form to add a new API key."""
|
||||
templates = _get_templates()
|
||||
operator = request.state.operator
|
||||
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="api_keys_new.html",
|
||||
|
|
@ -1870,14 +1888,13 @@ async def api_keys_new(
|
|||
"csrf_token": csrf_token,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
|
||||
@router.post("/api-keys", response_class=HTMLResponse)
|
||||
async def api_keys_create(
|
||||
request: Request,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> Response:
|
||||
"""Create a new API key."""
|
||||
from central.crypto import encrypt
|
||||
|
|
@ -1886,7 +1903,10 @@ async def api_keys_create(
|
|||
pool = get_pool()
|
||||
operator = request.state.operator
|
||||
|
||||
await csrf_protect.validate_csrf(request)
|
||||
form = await request.form()
|
||||
form_csrf = form.get("csrf_token", "")
|
||||
if not form_csrf or form_csrf != request.state.csrf_token:
|
||||
raise CsrfValidationError("Invalid CSRF token")
|
||||
|
||||
form = await request.form()
|
||||
alias = form.get("alias", "").strip()
|
||||
|
|
@ -1909,7 +1929,7 @@ async def api_keys_create(
|
|||
errors["plaintext_key"] = "API key must be at most 4096 characters"
|
||||
|
||||
if errors:
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="api_keys_new.html",
|
||||
|
|
@ -1920,7 +1940,6 @@ async def api_keys_create(
|
|||
"alias": alias,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
# Encrypt the key
|
||||
|
|
@ -1935,7 +1954,7 @@ async def api_keys_create(
|
|||
|
||||
if existing:
|
||||
errors["alias"] = "An API key with this alias already exists"
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="api_keys_new.html",
|
||||
|
|
@ -1946,7 +1965,6 @@ async def api_keys_create(
|
|||
"alias": alias,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
# Insert the new key
|
||||
|
|
@ -1977,7 +1995,7 @@ async def api_keys_create(
|
|||
async def api_keys_edit(
|
||||
request: Request,
|
||||
alias: str,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> Response:
|
||||
"""Show form to rotate or delete an API key."""
|
||||
templates = _get_templates()
|
||||
|
|
@ -2015,7 +2033,7 @@ async def api_keys_edit(
|
|||
"used_by": [a["name"] for a in adapters],
|
||||
}
|
||||
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="api_keys_edit.html",
|
||||
|
|
@ -2025,7 +2043,6 @@ async def api_keys_edit(
|
|||
"key": key,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
|
||||
|
|
@ -2033,7 +2050,7 @@ async def api_keys_edit(
|
|||
async def api_keys_rotate(
|
||||
request: Request,
|
||||
alias: str,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> Response:
|
||||
"""Rotate an API key."""
|
||||
from central.crypto import encrypt
|
||||
|
|
@ -2042,7 +2059,10 @@ async def api_keys_rotate(
|
|||
pool = get_pool()
|
||||
operator = request.state.operator
|
||||
|
||||
await csrf_protect.validate_csrf(request)
|
||||
form = await request.form()
|
||||
form_csrf = form.get("csrf_token", "")
|
||||
if not form_csrf or form_csrf != request.state.csrf_token:
|
||||
raise CsrfValidationError("Invalid CSRF token")
|
||||
|
||||
form = await request.form()
|
||||
new_plaintext_key = form.get("new_plaintext_key", "")
|
||||
|
|
@ -2086,7 +2106,7 @@ async def api_keys_rotate(
|
|||
"used_by": [a["name"] for a in adapters],
|
||||
}
|
||||
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="api_keys_edit.html",
|
||||
|
|
@ -2097,7 +2117,6 @@ async def api_keys_rotate(
|
|||
"errors": errors,
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
old_rotated_at = row["rotated_at"]
|
||||
|
|
@ -2134,14 +2153,17 @@ async def api_keys_rotate(
|
|||
async def api_keys_delete(
|
||||
request: Request,
|
||||
alias: str,
|
||||
csrf_protect: CsrfProtect = Depends(),
|
||||
|
||||
) -> Response:
|
||||
"""Delete an API key."""
|
||||
templates = _get_templates()
|
||||
pool = get_pool()
|
||||
operator = request.state.operator
|
||||
|
||||
await csrf_protect.validate_csrf(request)
|
||||
form = await request.form()
|
||||
form_csrf = form.get("csrf_token", "")
|
||||
if not form_csrf or form_csrf != request.state.csrf_token:
|
||||
raise CsrfValidationError("Invalid CSRF token")
|
||||
|
||||
async with pool.acquire() as conn:
|
||||
row = await conn.fetchrow(
|
||||
|
|
@ -2176,7 +2198,7 @@ async def api_keys_delete(
|
|||
"used_by": adapter_names,
|
||||
}
|
||||
|
||||
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
|
||||
csrf_token = request.state.csrf_token
|
||||
response = templates.TemplateResponse(
|
||||
request=request,
|
||||
name="api_keys_edit.html",
|
||||
|
|
@ -2187,7 +2209,6 @@ async def api_keys_delete(
|
|||
"error": f"Cannot delete: used by {', '.join(adapter_names)}. Remove these references first.",
|
||||
},
|
||||
)
|
||||
csrf_protect.set_csrf_cookie(signed_token, response)
|
||||
return response
|
||||
|
||||
# Delete the key
|
||||
|
|
|
|||
|
|
@ -92,29 +92,33 @@ class TestSessionManagement:
|
|||
mock_conn = MagicMock()
|
||||
mock_conn.execute = AsyncMock()
|
||||
|
||||
token, expires_at = await create_session(mock_conn, operator_id=1, lifetime_days=90)
|
||||
token, expires_at, csrf_token = await create_session(mock_conn, operator_id=1, lifetime_days=90)
|
||||
|
||||
assert len(token) == 43
|
||||
assert len(csrf_token) == 64 # 32 bytes hex = 64 chars
|
||||
mock_conn.execute.assert_called_once()
|
||||
call_args = mock_conn.execute.call_args
|
||||
assert "INSERT INTO config.sessions" in call_args[0][0]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_session_found(self):
|
||||
"""get_session returns Operator when session exists."""
|
||||
"""get_session returns (Operator, csrf_token) when session exists."""
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.fetchrow = AsyncMock(return_value={
|
||||
"id": 1,
|
||||
"username": "testuser",
|
||||
"created_at": datetime.now(timezone.utc),
|
||||
"password_changed_at": datetime.now(timezone.utc),
|
||||
"csrf_token": "test_csrf_token_12345",
|
||||
})
|
||||
|
||||
operator = await get_session(mock_conn, "valid-token")
|
||||
result = await get_session(mock_conn, "valid-token")
|
||||
|
||||
assert operator is not None
|
||||
assert result is not None
|
||||
operator, csrf_token = result
|
||||
assert operator.id == 1
|
||||
assert operator.username == "testuser"
|
||||
assert csrf_token == "test_csrf_token_12345"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_session_not_found(self):
|
||||
|
|
|
|||
|
|
@ -14,23 +14,18 @@ class TestCsrfExceptionHandlerRegistered:
|
|||
"""Verify CSRF exception handler is properly registered."""
|
||||
|
||||
def test_csrf_exception_handler_is_registered(self):
|
||||
"""The app has a CsrfProtectError exception handler registered."""
|
||||
"""The app has a CsrfValidationError exception handler registered."""
|
||||
from central.gui import app
|
||||
from fastapi_csrf_protect.exceptions import CsrfProtectError
|
||||
from central.gui.auth import CsrfValidationError
|
||||
|
||||
assert CsrfProtectError in app.exception_handlers, \
|
||||
"CsrfProtectError handler should be registered"
|
||||
assert CsrfValidationError in app.exception_handlers, \
|
||||
"CsrfValidationError handler should be registered"
|
||||
|
||||
def test_csrf_subclasses_are_caught(self):
|
||||
"""MissingTokenError and TokenValidationError inherit from CsrfProtectError."""
|
||||
from fastapi_csrf_protect.exceptions import (
|
||||
CsrfProtectError,
|
||||
MissingTokenError,
|
||||
TokenValidationError,
|
||||
)
|
||||
def test_csrf_validation_error_is_exception(self):
|
||||
"""CsrfValidationError is a proper Exception subclass."""
|
||||
from central.gui.auth import CsrfValidationError
|
||||
|
||||
assert issubclass(MissingTokenError, CsrfProtectError)
|
||||
assert issubclass(TokenValidationError, CsrfProtectError)
|
||||
assert issubclass(CsrfValidationError, Exception)
|
||||
|
||||
|
||||
class TestCsrfExceptionHandlerBehavior:
|
||||
|
|
@ -40,10 +35,10 @@ class TestCsrfExceptionHandlerBehavior:
|
|||
"""CSRF handler checks request path for /login."""
|
||||
import inspect
|
||||
from central.gui import _create_app
|
||||
from fastapi_csrf_protect.exceptions import CsrfProtectError
|
||||
from central.gui.auth import CsrfValidationError
|
||||
|
||||
app = _create_app()
|
||||
handler = app.exception_handlers.get(CsrfProtectError)
|
||||
handler = app.exception_handlers.get(CsrfValidationError)
|
||||
|
||||
# Verify handler source contains /login path check
|
||||
source = inspect.getsource(handler)
|
||||
|
|
@ -54,17 +49,16 @@ class TestCsrfExceptionHandlerBehavior:
|
|||
async def test_logout_csrf_error_redirects_to_login(self):
|
||||
"""CSRF error on /logout should redirect to /login."""
|
||||
from central.gui import _create_app
|
||||
from fastapi_csrf_protect.exceptions import TokenValidationError
|
||||
from central.gui.auth import CsrfValidationError
|
||||
from fastapi.responses import RedirectResponse
|
||||
|
||||
app = _create_app()
|
||||
from fastapi_csrf_protect.exceptions import CsrfProtectError
|
||||
handler = app.exception_handlers.get(CsrfProtectError)
|
||||
handler = app.exception_handlers.get(CsrfValidationError)
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.url.path = "/logout"
|
||||
|
||||
exc = TokenValidationError("Invalid token")
|
||||
exc = CsrfValidationError("Invalid token")
|
||||
|
||||
result = await handler(mock_request, exc)
|
||||
|
||||
|
|
@ -75,17 +69,16 @@ class TestCsrfExceptionHandlerBehavior:
|
|||
async def test_adapters_csrf_error_redirects_to_adapters(self):
|
||||
"""CSRF error on /adapters/{name} should redirect to /adapters."""
|
||||
from central.gui import _create_app
|
||||
from fastapi_csrf_protect.exceptions import TokenValidationError
|
||||
from central.gui.auth import CsrfValidationError
|
||||
from fastapi.responses import RedirectResponse
|
||||
|
||||
app = _create_app()
|
||||
from fastapi_csrf_protect.exceptions import CsrfProtectError
|
||||
handler = app.exception_handlers.get(CsrfProtectError)
|
||||
handler = app.exception_handlers.get(CsrfValidationError)
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.url.path = "/adapters/nws"
|
||||
|
||||
exc = TokenValidationError("Invalid token")
|
||||
exc = CsrfValidationError("Invalid token")
|
||||
|
||||
result = await handler(mock_request, exc)
|
||||
|
||||
|
|
@ -94,16 +87,16 @@ class TestCsrfExceptionHandlerBehavior:
|
|||
|
||||
|
||||
class TestCsrfHandlerNoTraceback:
|
||||
"""Verify exception handler doesn't expose Python internals."""
|
||||
"""Verify exception handler does not expose Python internals."""
|
||||
|
||||
def test_handler_exists_and_is_async(self):
|
||||
"""The CSRF handler should be an async function."""
|
||||
import inspect
|
||||
from central.gui import _create_app
|
||||
from fastapi_csrf_protect.exceptions import CsrfProtectError
|
||||
from central.gui.auth import CsrfValidationError
|
||||
|
||||
app = _create_app()
|
||||
handler = app.exception_handlers.get(CsrfProtectError)
|
||||
handler = app.exception_handlers.get(CsrfValidationError)
|
||||
|
||||
assert handler is not None
|
||||
assert inspect.iscoroutinefunction(handler)
|
||||
|
|
@ -116,17 +109,15 @@ class TestCsrfHandlerWizardPaths:
|
|||
async def test_setup_operator_csrf_error_renders_form_with_error(self):
|
||||
"""CSRF error on /setup/operator re-renders form with error message."""
|
||||
from central.gui import _create_app
|
||||
from fastapi_csrf_protect.exceptions import TokenValidationError
|
||||
from fastapi.responses import HTMLResponse
|
||||
from central.gui.auth import CsrfValidationError
|
||||
|
||||
app = _create_app()
|
||||
from fastapi_csrf_protect.exceptions import CsrfProtectError
|
||||
handler = app.exception_handlers.get(CsrfProtectError)
|
||||
handler = app.exception_handlers.get(CsrfValidationError)
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.url.path = "/setup/operator"
|
||||
|
||||
exc = TokenValidationError("Invalid token")
|
||||
exc = CsrfValidationError("Invalid token")
|
||||
|
||||
result = await handler(mock_request, exc)
|
||||
|
||||
|
|
@ -140,16 +131,15 @@ class TestCsrfHandlerWizardPaths:
|
|||
async def test_setup_system_csrf_error_renders_form_with_error(self):
|
||||
"""CSRF error on /setup/system re-renders form with error message."""
|
||||
from central.gui import _create_app
|
||||
from fastapi_csrf_protect.exceptions import TokenValidationError
|
||||
from central.gui.auth import CsrfValidationError
|
||||
|
||||
app = _create_app()
|
||||
from fastapi_csrf_protect.exceptions import CsrfProtectError
|
||||
handler = app.exception_handlers.get(CsrfProtectError)
|
||||
handler = app.exception_handlers.get(CsrfValidationError)
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.url.path = "/setup/system"
|
||||
|
||||
exc = TokenValidationError("Invalid token")
|
||||
exc = CsrfValidationError("Invalid token")
|
||||
|
||||
with patch("central.gui.db.get_pool", return_value=None):
|
||||
result = await handler(mock_request, exc)
|
||||
|
|
@ -163,16 +153,15 @@ class TestCsrfHandlerWizardPaths:
|
|||
async def test_setup_keys_csrf_error_renders_form_with_error(self):
|
||||
"""CSRF error on /setup/keys re-renders form with error message."""
|
||||
from central.gui import _create_app
|
||||
from fastapi_csrf_protect.exceptions import TokenValidationError
|
||||
from central.gui.auth import CsrfValidationError
|
||||
|
||||
app = _create_app()
|
||||
from fastapi_csrf_protect.exceptions import CsrfProtectError
|
||||
handler = app.exception_handlers.get(CsrfProtectError)
|
||||
handler = app.exception_handlers.get(CsrfValidationError)
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.url.path = "/setup/keys"
|
||||
|
||||
exc = TokenValidationError("Invalid token")
|
||||
exc = CsrfValidationError("Invalid token")
|
||||
|
||||
with patch("central.gui.db.get_pool", return_value=None):
|
||||
result = await handler(mock_request, exc)
|
||||
|
|
@ -186,16 +175,15 @@ class TestCsrfHandlerWizardPaths:
|
|||
async def test_setup_adapters_csrf_error_renders_form_with_error(self):
|
||||
"""CSRF error on /setup/adapters re-renders form with error message."""
|
||||
from central.gui import _create_app
|
||||
from fastapi_csrf_protect.exceptions import TokenValidationError
|
||||
from central.gui.auth import CsrfValidationError
|
||||
|
||||
app = _create_app()
|
||||
from fastapi_csrf_protect.exceptions import CsrfProtectError
|
||||
handler = app.exception_handlers.get(CsrfProtectError)
|
||||
handler = app.exception_handlers.get(CsrfValidationError)
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.url.path = "/setup/adapters"
|
||||
|
||||
exc = TokenValidationError("Invalid token")
|
||||
exc = CsrfValidationError("Invalid token")
|
||||
|
||||
with patch("central.gui.db.get_pool", return_value=None):
|
||||
result = await handler(mock_request, exc)
|
||||
|
|
@ -209,16 +197,15 @@ class TestCsrfHandlerWizardPaths:
|
|||
async def test_setup_finish_csrf_error_renders_form_with_error(self):
|
||||
"""CSRF error on /setup/finish re-renders form with error message."""
|
||||
from central.gui import _create_app
|
||||
from fastapi_csrf_protect.exceptions import TokenValidationError
|
||||
from central.gui.auth import CsrfValidationError
|
||||
|
||||
app = _create_app()
|
||||
from fastapi_csrf_protect.exceptions import CsrfProtectError
|
||||
handler = app.exception_handlers.get(CsrfProtectError)
|
||||
handler = app.exception_handlers.get(CsrfValidationError)
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.url.path = "/setup/finish"
|
||||
|
||||
exc = TokenValidationError("Invalid token")
|
||||
exc = CsrfValidationError("Invalid token")
|
||||
|
||||
with patch("central.gui.db.get_pool", return_value=None):
|
||||
result = await handler(mock_request, exc)
|
||||
|
|
@ -232,17 +219,16 @@ class TestCsrfHandlerWizardPaths:
|
|||
async def test_setup_base_csrf_error_redirects_to_setup(self):
|
||||
"""CSRF error on /setup redirects to /setup (middleware routes to step)."""
|
||||
from central.gui import _create_app
|
||||
from fastapi_csrf_protect.exceptions import TokenValidationError
|
||||
from central.gui.auth import CsrfValidationError
|
||||
from fastapi.responses import RedirectResponse
|
||||
|
||||
app = _create_app()
|
||||
from fastapi_csrf_protect.exceptions import CsrfProtectError
|
||||
handler = app.exception_handlers.get(CsrfProtectError)
|
||||
handler = app.exception_handlers.get(CsrfValidationError)
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.url.path = "/setup"
|
||||
|
||||
exc = TokenValidationError("Invalid token")
|
||||
exc = CsrfValidationError("Invalid token")
|
||||
|
||||
result = await handler(mock_request, exc)
|
||||
|
||||
|
|
@ -253,16 +239,15 @@ class TestCsrfHandlerWizardPaths:
|
|||
async def test_login_csrf_error_still_works(self):
|
||||
"""CSRF error on /login still renders login form with error (regression test)."""
|
||||
from central.gui import _create_app
|
||||
from fastapi_csrf_protect.exceptions import TokenValidationError
|
||||
from central.gui.auth import CsrfValidationError
|
||||
|
||||
app = _create_app()
|
||||
from fastapi_csrf_protect.exceptions import CsrfProtectError
|
||||
handler = app.exception_handlers.get(CsrfProtectError)
|
||||
handler = app.exception_handlers.get(CsrfValidationError)
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.url.path = "/login"
|
||||
|
||||
exc = TokenValidationError("Invalid token")
|
||||
exc = CsrfValidationError("Invalid token")
|
||||
|
||||
result = await handler(mock_request, exc)
|
||||
|
||||
|
|
|
|||
108
tests/test_csrf_race_condition.py
Normal file
108
tests/test_csrf_race_condition.py
Normal file
|
|
@ -0,0 +1,108 @@
|
|||
"""
|
||||
Integration test for CSRF race condition fix.
|
||||
|
||||
This test verifies that the session-bound CSRF implementation fixes the race
|
||||
condition where interleaved GET requests would invalidate CSRF tokens.
|
||||
|
||||
See: PR #24 - Central 1b-8 fix-up phase 2
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
class TestCsrfRaceConditionFix:
|
||||
"""Verify that interleaved GETs don't break CSRF validation."""
|
||||
|
||||
def test_session_bound_csrf_consistent_across_gets(self):
|
||||
"""Session-bound CSRF tokens remain consistent across multiple GETs.
|
||||
|
||||
This was the core bug: fastapi-csrf-protect rotated tokens on every GET,
|
||||
causing race conditions when users had multiple tabs or slow connections.
|
||||
|
||||
With session-bound CSRF, the token is stored in the session row and
|
||||
remains constant until the session is destroyed.
|
||||
"""
|
||||
from unittest.mock import MagicMock, AsyncMock
|
||||
from central.gui.auth import get_session
|
||||
|
||||
# Mock a session with a csrf_token
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.fetchrow = AsyncMock(return_value={
|
||||
"id": 1,
|
||||
"username": "testuser",
|
||||
"created_at": "2024-01-01T00:00:00Z",
|
||||
"password_changed_at": "2024-01-01T00:00:00Z",
|
||||
"csrf_token": "fixed_csrf_token_12345",
|
||||
})
|
||||
|
||||
import asyncio
|
||||
|
||||
async def test():
|
||||
# First GET
|
||||
result1 = await get_session(mock_conn, "test-token")
|
||||
assert result1 is not None
|
||||
op1, csrf1 = result1
|
||||
|
||||
# Second GET (simulating interleaved request)
|
||||
result2 = await get_session(mock_conn, "test-token")
|
||||
assert result2 is not None
|
||||
op2, csrf2 = result2
|
||||
|
||||
# CSRF tokens should be identical (the fix!)
|
||||
assert csrf1 == csrf2 == "fixed_csrf_token_12345"
|
||||
|
||||
asyncio.run(test())
|
||||
|
||||
def test_pre_auth_csrf_tokens_independently_valid(self):
|
||||
"""Pre-auth CSRF tokens are independently valid.
|
||||
|
||||
For unauthenticated routes, each GET generates a new token+cookie pair.
|
||||
Each pair should validate independently, allowing the original token
|
||||
to work even if another GET happened in between.
|
||||
"""
|
||||
from central.gui.csrf import generate_pre_auth_csrf, validate_pre_auth_csrf
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
secret = "testsecret12345678901234567890ab"
|
||||
|
||||
# First GET generates token1 + cookie1
|
||||
token1, signed1 = generate_pre_auth_csrf(secret)
|
||||
|
||||
# Second GET generates token2 + cookie2
|
||||
token2, signed2 = generate_pre_auth_csrf(secret)
|
||||
|
||||
# Tokens should be different (fresh random tokens)
|
||||
assert token1 != token2
|
||||
assert signed1 != signed2
|
||||
|
||||
# But each pair should validate independently
|
||||
mock_request1 = MagicMock()
|
||||
mock_request1.cookies = {"central_preauth_csrf": signed1}
|
||||
|
||||
mock_request2 = MagicMock()
|
||||
mock_request2.cookies = {"central_preauth_csrf": signed2}
|
||||
|
||||
# Original token still validates with original cookie
|
||||
assert validate_pre_auth_csrf(mock_request1, token1, secret) is True
|
||||
|
||||
# Second token validates with second cookie
|
||||
assert validate_pre_auth_csrf(mock_request2, token2, secret) is True
|
||||
|
||||
# Cross-validation should fail
|
||||
assert validate_pre_auth_csrf(mock_request1, token2, secret) is False
|
||||
assert validate_pre_auth_csrf(mock_request2, token1, secret) is False
|
||||
|
||||
def test_csrf_token_generation_is_secure(self):
|
||||
"""CSRF tokens are cryptographically secure."""
|
||||
from central.gui.auth import generate_csrf_token
|
||||
|
||||
# Generate multiple tokens
|
||||
tokens = [generate_csrf_token() for _ in range(100)]
|
||||
|
||||
# All tokens should be unique
|
||||
assert len(set(tokens)) == 100
|
||||
|
||||
# Tokens should be 64 hex chars (32 bytes)
|
||||
for token in tokens:
|
||||
assert len(token) == 64
|
||||
assert all(c in "0123456789abcdef" for c in token)
|
||||
|
|
@ -43,6 +43,7 @@ class TestSessionMiddleware:
|
|||
"username": "admin",
|
||||
"created_at": datetime.now(timezone.utc),
|
||||
"password_changed_at": datetime.now(timezone.utc),
|
||||
"csrf_token": "mock_csrf_token_12345",
|
||||
})
|
||||
mock_conn.__aenter__ = AsyncMock(return_value=mock_conn)
|
||||
mock_conn.__aexit__ = AsyncMock()
|
||||
|
|
@ -99,6 +100,7 @@ class TestSessionMiddleware:
|
|||
"username": "admin",
|
||||
"created_at": datetime.now(timezone.utc),
|
||||
"password_changed_at": datetime.now(timezone.utc),
|
||||
"csrf_token": "mock_csrf_token_12345",
|
||||
})
|
||||
mock_conn.__aenter__ = AsyncMock(return_value=mock_conn)
|
||||
mock_conn.__aexit__ = AsyncMock()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue