60 lines
2.1 KiB
Python
60 lines
2.1 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
import hashlib
|
|
import hmac
|
|
|
|
from fastapi import HTTPException, Request
|
|
from sqlalchemy import delete, func, select, text
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .models import AdminAuthAttempt
|
|
from .submission_security import RateLimitConfig, client_address
|
|
|
|
|
|
class AdminAuthConfig(RateLimitConfig):
|
|
admin_token: str
|
|
admin_auth_attempt_limit: int
|
|
admin_auth_window_seconds: int
|
|
|
|
|
|
def verify_admin(
|
|
request: Request,
|
|
db: Session,
|
|
authorization: str | None,
|
|
config: AdminAuthConfig,
|
|
) -> str:
|
|
now = datetime.now(timezone.utc)
|
|
cutoff = now - timedelta(seconds=config.admin_auth_window_seconds)
|
|
client = client_address(request, config.trusted_proxy_cidrs)
|
|
client_hash = hmac.new(
|
|
config.rate_limit_secret.encode(), f"admin:{client}".encode(), hashlib.sha256
|
|
).hexdigest()
|
|
if db.get_bind().dialect.name == "postgresql":
|
|
lock_key = int(client_hash[:16], 16) & 0x7FFF_FFFF_FFFF_FFFF
|
|
db.execute(text("SELECT pg_advisory_xact_lock(:lock_key)"), {"lock_key": lock_key})
|
|
db.execute(delete(AdminAuthAttempt).where(AdminAuthAttempt.created_at < now - timedelta(days=1)))
|
|
failures = db.scalar(
|
|
select(func.count()).select_from(AdminAuthAttempt).where(
|
|
AdminAuthAttempt.client_hash == client_hash,
|
|
AdminAuthAttempt.created_at >= cutoff,
|
|
)
|
|
) or 0
|
|
if failures >= config.admin_auth_attempt_limit:
|
|
db.commit()
|
|
raise HTTPException(
|
|
status_code=429,
|
|
detail="too many admin authentication attempts",
|
|
headers={"Retry-After": str(config.admin_auth_window_seconds)},
|
|
)
|
|
expected = f"Bearer {config.admin_token}"
|
|
if not authorization or not hmac.compare_digest(authorization, expected):
|
|
db.add(AdminAuthAttempt(client_hash=client_hash, created_at=now))
|
|
db.commit()
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail="invalid admin token",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
db.execute(delete(AdminAuthAttempt).where(AdminAuthAttempt.client_hash == client_hash))
|
|
db.commit()
|
|
return "admin"
|