93 lines
2.2 KiB
Python
93 lines
2.2 KiB
Python
"""
|
|
OIDC authorization code management.
|
|
|
|
Authorization codes are short-lived (60 seconds) and stored in-memory only.
|
|
Similar to remote auth, these are not persisted to the database.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import secrets
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import msgspec
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
# Auth codes expire after this duration
|
|
AUTH_CODE_LIFETIME = timedelta(seconds=60)
|
|
|
|
|
|
class AuthCode(msgspec.Struct):
|
|
"""A pending authorization code for OIDC or native auth."""
|
|
|
|
session_key: str
|
|
created: datetime
|
|
oidc: OIDC | None = None # OIDC-specific fields (None for native auth)
|
|
|
|
|
|
class OIDC(msgspec.Struct):
|
|
"""OIDC verification data carried authenticate->token that is not stored in Session."""
|
|
|
|
redirect_uri: str
|
|
scope: str
|
|
nonce: str
|
|
code_challenge: str
|
|
code_challenge_method: str
|
|
|
|
|
|
# Public interface - auth codes in-memory store
|
|
codes: dict[str, AuthCode] = {}
|
|
|
|
# Background cleanup task
|
|
_cleanup_task: asyncio.Task | None = None
|
|
|
|
|
|
async def start():
|
|
"""Start the cleanup background task."""
|
|
global _cleanup_task
|
|
if _cleanup_task is None:
|
|
_cleanup_task = asyncio.create_task(_cleanup_loop())
|
|
|
|
|
|
async def stop():
|
|
"""Stop the cleanup background task."""
|
|
global _cleanup_task
|
|
if _cleanup_task:
|
|
_cleanup_task.cancel()
|
|
try:
|
|
await _cleanup_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
_cleanup_task = None
|
|
|
|
|
|
async def _cleanup_loop():
|
|
while True:
|
|
try:
|
|
await asyncio.sleep(30) # Check every 30 seconds
|
|
_cleanup_expired()
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception:
|
|
_logger.exception("Error in auth code cleanup loop")
|
|
|
|
|
|
def _cleanup_expired():
|
|
oldest = datetime.now(UTC) - AUTH_CODE_LIFETIME
|
|
for code, auth_code in list(codes.items()):
|
|
if auth_code.created < oldest:
|
|
del codes[code]
|
|
|
|
|
|
def store(auth_code: AuthCode) -> str:
|
|
"""Store an authorization code and return the code string.
|
|
|
|
Caller must construct AuthCode with their own timestamp.
|
|
"""
|
|
code = secrets.token_urlsafe(12)
|
|
codes[code] = auth_code
|
|
return code
|