auth: Sprint 02 — auth hardening (Pydantic models, challenge binding, admin gates)

- Add max_length validation to LoginRequest (username 128, password 1024)
- Replace raw request.json() with Pydantic models in RBAC override/update endpoints
- Replace raw request.json() with Pydantic models in passkey register/login/delete
- Bind passkey challenges to session-bound challenge_id (prevents cross-session replay)
- Gate audit log and security log endpoints behind admin role
- Fix fragile opc_db.json.dumps() → import json directly
- Add COOKIE_SECURE=False startup warning (suppress with ALLOW_INSECURE_COOKIES)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Gan, Jimmy
2026-05-03 13:45:42 +08:00
parent 323ae474a8
commit 1a37f34401
7 changed files with 101 additions and 62 deletions
+19 -21
View File
@@ -7,7 +7,7 @@ from datetime import timedelta
import httpx
import jwt
from fastapi import APIRouter, Body, Depends, HTTPException, Request, Response, status
from pydantic import BaseModel
from pydantic import BaseModel, Field
from slowapi import Limiter
from slowapi.util import get_remote_address
@@ -20,8 +20,8 @@ limiter = Limiter(key_func=get_remote_address)
class LoginRequest(BaseModel):
username: str
password: str
username: str = Field(..., max_length=128)
password: str = Field(..., max_length=1024)
totp_code: str = ""
@@ -239,20 +239,23 @@ async def rbac_config(current_user=Depends(auth.get_current_user)):
return load_rbac()
class RbacOverrideRequest(BaseModel):
pages: list[str] | str
class RbacUpdateRoleRequest(BaseModel):
pages: list[str] | str
sidebar_links: list[str] | str | None = None
@router.put("/rbac/overrides/{username}")
async def rbac_set_override(username: str, request: Request, current_user=Depends(auth.get_current_user)):
async def rbac_set_override(username: str, body: RbacOverrideRequest, current_user=Depends(auth.get_current_user)):
if current_user.role != "admin":
raise HTTPException(status_code=403, detail="Admin only")
from rbac import update_rbac
body = await request.json()
pages = body.get("pages")
if pages is None:
raise HTTPException(status_code=400, detail="Missing pages field")
def mutator(rbac):
existing = rbac.setdefault("user_overrides", {}).get(username, {})
existing["pages"] = pages
existing["pages"] = body.pages
rbac["user_overrides"][username] = existing
update_rbac(mutator)
@@ -301,25 +304,20 @@ async def rbac_delete_override(username: str, current_user=Depends(auth.get_curr
@router.put("/rbac/roles/{role}")
async def rbac_update_role(role: str, request: Request, current_user=Depends(auth.get_current_user)):
async def rbac_update_role(role: str, body: RbacUpdateRoleRequest, current_user=Depends(auth.get_current_user)):
if current_user.role != "admin":
raise HTTPException(status_code=403, detail="Admin only")
from rbac import update_rbac
body = await request.json()
pages = body.get("pages")
sidebar_links = body.get("sidebar_links")
if pages is None:
raise HTTPException(status_code=400, detail="Missing pages field")
if pages != "*" and not isinstance(pages, list):
if body.pages != "*" and not isinstance(body.pages, list):
raise HTTPException(status_code=400, detail="pages must be '*' or a list")
if sidebar_links is not None and sidebar_links != "*" and not isinstance(sidebar_links, list):
if body.sidebar_links is not None and body.sidebar_links != "*" and not isinstance(body.sidebar_links, list):
raise HTTPException(status_code=400, detail="sidebar_links must be '*' or a list")
def mutator(rbac):
role_config = {"pages": pages}
if sidebar_links is not None:
role_config["sidebar_links"] = sidebar_links
role_config = {"pages": body.pages}
if body.sidebar_links is not None:
role_config["sidebar_links"] = body.sidebar_links
rbac.setdefault("role_defaults", {})[role] = role_config
update_rbac(mutator)