File size: 2,831 Bytes
a579d44 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | """
OmniDiag — API Key Authentication
====================================
Provides an `X-API-Key` header as an alternative auth method for
programmatic integrations (scripts, dashboards, third-party systems).
Design:
- Keys are generated as 32-byte random hex strings (256-bit entropy)
- Only the bcrypt hash is stored in the DB — the plain key is shown
once at creation and never retrievable again
- Each key has an optional expiry date
- A valid API key grants the same permissions as the owning user's roles
- API key lookup goes through the same `get_current_active_user` path
so RBAC enforcement is unchanged
"""
import secrets
from datetime import datetime, timezone
from typing import Optional
from fastapi import Header, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from backend.auth.hashing import verify_password
from backend.db_models.user import User
def generate_api_key() -> str:
"""Return a cryptographically random 64-hex-char API key."""
return secrets.token_hex(32)
async def get_user_by_api_key(
api_key: str,
db: AsyncSession,
) -> Optional[User]:
"""
Look up the user whose stored api_key_hash matches *api_key*.
Returns None if no match or the key has expired.
Brute-force is mitigated by bcrypt's cost factor.
"""
# Fetch all users that have an api_key_hash set (index scan)
result = await db.execute(
select(User).where(User.api_key_hash.isnot(None))
)
candidates = result.scalars().all()
for user in candidates:
if verify_password(api_key, user.api_key_hash):
# Check expiry
if user.api_key_expires_at is not None:
if user.api_key_expires_at < datetime.now(timezone.utc):
return None
return user
return None
async def get_current_user_via_api_key(
x_api_key: Optional[str] = Header(None, alias="X-API-Key"),
db: AsyncSession = None, # injected by caller
) -> Optional[User]:
"""
FastAPI dependency: resolve a User from X-API-Key header.
Returns None if header absent — callers can chain with JWT auth.
Raises 401 if header is present but invalid/expired.
"""
if not x_api_key:
return None
user = await get_user_by_api_key(x_api_key, db)
if user is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"error": "Invalid or expired API key", "code": "INVALID_API_KEY"},
headers={"WWW-Authenticate": "ApiKey"},
)
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={"error": "Account is disabled", "code": "ACCOUNT_DISABLED"},
)
return user
|