refactor: split auth router into focused modules
Deploy Dashboard (Dev) / deploy-dev (push) Has been cancelled
Deploy Dashboard (Dev) / deploy-dev (push) Has been cancelled
Split the 429-line auth.py into three modules for better maintainability: - auth.py (264 lines): core auth, login, token refresh, RBAC, preferences - passkey.py (187 lines): WebAuthn passkey registration and authentication - totp.py (41 lines): TOTP 2FA setup and verification All 21 endpoints maintain backward compatibility under /api/auth prefix. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,187 @@
|
||||
"""WebAuthn Passkey authentication endpoints."""
|
||||
import json
|
||||
import time
|
||||
from datetime import timedelta
|
||||
from fastapi import APIRouter, Depends, HTTPException, Request
|
||||
from fastapi.responses import JSONResponse
|
||||
from slowapi import Limiter
|
||||
from slowapi.util import get_remote_address
|
||||
from webauthn import (
|
||||
generate_registration_options,
|
||||
verify_registration_response,
|
||||
generate_authentication_options,
|
||||
verify_authentication_response
|
||||
)
|
||||
from webauthn.helpers.structs import PublicKeyCredentialDescriptor, UserVerificationRequirement
|
||||
from webauthn.helpers import bytes_to_base64url, base64url_to_bytes, options_to_json
|
||||
import auth
|
||||
import config
|
||||
|
||||
router = APIRouter()
|
||||
limiter = Limiter(key_func=get_remote_address)
|
||||
|
||||
# In-memory challenge store with TTL
|
||||
_challenges = {}
|
||||
|
||||
def _store_challenge(challenge: bytes):
|
||||
"""Store a WebAuthn challenge with timestamp for TTL tracking."""
|
||||
_challenges[challenge] = time.time()
|
||||
# Clean up expired challenges (>5 minutes old)
|
||||
now = time.time()
|
||||
expired = [k for k, v in _challenges.items() if now - v > 300]
|
||||
for k in expired:
|
||||
_challenges.pop(k, None)
|
||||
|
||||
def _get_challenge(client_data_b64: str) -> bytes:
|
||||
"""Extract and validate challenge from clientDataJSON."""
|
||||
if not client_data_b64:
|
||||
raise HTTPException(status_code=400, detail="Missing clientDataJSON")
|
||||
try:
|
||||
data = json.loads(base64url_to_bytes(client_data_b64).decode('utf-8'))
|
||||
chal_bytes = base64url_to_bytes(data.get("challenge", ""))
|
||||
except Exception:
|
||||
raise HTTPException(status_code=400, detail="Invalid clientDataJSON")
|
||||
|
||||
entry = _challenges.pop(chal_bytes, None)
|
||||
if not entry or time.time() - entry > 300:
|
||||
raise HTTPException(status_code=400, detail="Challenge expired or invalid")
|
||||
return chal_bytes
|
||||
|
||||
@router.post("/passkey/register/options")
|
||||
async def passkey_register_options(current_user = Depends(auth.get_current_user)):
|
||||
"""Generate WebAuthn registration options for adding a new passkey."""
|
||||
existing = auth.load_passkey_credentials()
|
||||
exclude = [
|
||||
PublicKeyCredentialDescriptor(id=base64url_to_bytes(c["credential_id"]))
|
||||
for c in existing
|
||||
]
|
||||
options = generate_registration_options(
|
||||
rp_id=config.WEBAUTHN_RP_ID,
|
||||
rp_name="NAS Dashboard",
|
||||
user_id=current_user.username.encode(),
|
||||
user_name=current_user.username,
|
||||
user_display_name=current_user.username,
|
||||
exclude_credentials=exclude,
|
||||
)
|
||||
_store_challenge(options.challenge)
|
||||
return JSONResponse(content=json.loads(options_to_json(options)))
|
||||
|
||||
@router.post("/passkey/register/verify")
|
||||
async def passkey_register_verify(request: Request, current_user = Depends(auth.get_current_user)):
|
||||
"""Verify and save a new passkey registration."""
|
||||
body = await request.json()
|
||||
challenge = _get_challenge(body.get("response", {}).get("clientDataJSON", ""))
|
||||
try:
|
||||
verification = verify_registration_response(
|
||||
credential=body,
|
||||
expected_challenge=challenge,
|
||||
expected_rp_id=config.WEBAUTHN_RP_ID,
|
||||
expected_origin=config.WEBAUTHN_ORIGINS,
|
||||
)
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
|
||||
auth.save_passkey_credential({
|
||||
"credential_id": bytes_to_base64url(verification.credential_id),
|
||||
"public_key": bytes_to_base64url(verification.credential_public_key),
|
||||
"sign_count": verification.sign_count,
|
||||
"name": body.get("name", "Passkey"),
|
||||
"created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||||
})
|
||||
return {"ok": True}
|
||||
|
||||
@router.post("/passkey/login/options")
|
||||
@limiter.limit("10/minute")
|
||||
async def passkey_login_options(request: Request):
|
||||
"""Generate WebAuthn authentication options for passkey login."""
|
||||
creds = auth.load_passkey_credentials()
|
||||
if not creds:
|
||||
raise HTTPException(status_code=404, detail="No passkeys registered")
|
||||
|
||||
allow = [
|
||||
PublicKeyCredentialDescriptor(id=base64url_to_bytes(c["credential_id"]))
|
||||
for c in creds
|
||||
]
|
||||
options = generate_authentication_options(
|
||||
rp_id=config.WEBAUTHN_RP_ID,
|
||||
allow_credentials=allow,
|
||||
user_verification=UserVerificationRequirement.PREFERRED,
|
||||
)
|
||||
_store_challenge(options.challenge)
|
||||
return JSONResponse(content=json.loads(options_to_json(options)))
|
||||
|
||||
@router.post("/passkey/login/verify")
|
||||
@limiter.limit("10/minute")
|
||||
async def passkey_login_verify(request: Request):
|
||||
"""Verify passkey authentication and issue tokens."""
|
||||
body = await request.json()
|
||||
challenge = _get_challenge(body.get("response", {}).get("clientDataJSON", ""))
|
||||
creds = auth.load_passkey_credentials()
|
||||
cred_id_b64 = body.get("id", "")
|
||||
matched = next((c for c in creds if c["credential_id"] == cred_id_b64), None)
|
||||
if not matched:
|
||||
raise HTTPException(status_code=400, detail="Unknown credential")
|
||||
|
||||
try:
|
||||
verification = verify_authentication_response(
|
||||
credential=body,
|
||||
expected_challenge=challenge,
|
||||
expected_rp_id=config.WEBAUTHN_RP_ID,
|
||||
expected_origin=config.WEBAUTHN_ORIGINS,
|
||||
credential_public_key=base64url_to_bytes(matched["public_key"]),
|
||||
credential_current_sign_count=matched["sign_count"],
|
||||
)
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
|
||||
# Update sign count
|
||||
matched["sign_count"] = verification.new_sign_count
|
||||
data = auth._load_auth_data()
|
||||
data["passkey_credentials"] = creds
|
||||
auth._save_auth_data(data)
|
||||
|
||||
username = config.ADMIN_USER
|
||||
access_token = auth.create_access_token(
|
||||
data={"sub": username, "role": "admin"},
|
||||
expires_delta=timedelta(minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES),
|
||||
)
|
||||
refresh_token = auth.create_refresh_token(data={"sub": username, "role": "admin"})
|
||||
return {
|
||||
"access_token": access_token,
|
||||
"refresh_token": refresh_token,
|
||||
"token_type": "bearer",
|
||||
"user": username,
|
||||
"role": "admin"
|
||||
}
|
||||
|
||||
@router.get("/passkey/list")
|
||||
async def passkey_list(current_user = Depends(auth.get_current_user)):
|
||||
"""List all registered passkeys for the current user."""
|
||||
creds = auth.load_passkey_credentials()
|
||||
return {
|
||||
"passkeys": [
|
||||
{
|
||||
"id": c["credential_id"],
|
||||
"name": c.get("name", "Passkey"),
|
||||
"created_at": c.get("created_at", "")
|
||||
}
|
||||
for c in creds
|
||||
]
|
||||
}
|
||||
|
||||
@router.post("/passkey/delete")
|
||||
async def passkey_delete(request: Request, current_user = Depends(auth.get_current_user)):
|
||||
"""Delete a specific passkey by credential ID."""
|
||||
body = await request.json()
|
||||
cred_id = body.get("id")
|
||||
data = auth._load_auth_data()
|
||||
creds = data.get("passkey_credentials", [])
|
||||
data["passkey_credentials"] = [c for c in creds if c["credential_id"] != cred_id]
|
||||
auth._save_auth_data(data)
|
||||
return {"ok": True}
|
||||
|
||||
@router.post("/passkey/clear")
|
||||
async def passkey_clear(current_user = Depends(auth.get_current_user)):
|
||||
"""Clear all passkeys for the current user."""
|
||||
auth.clear_passkey_credentials()
|
||||
return {"ok": True}
|
||||
Reference in New Issue
Block a user