63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import Depends, Header, HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from .api_keys_service import KEY_PREFIX, parse_api_token, verify_api_token
|
|
from .database import get_db
|
|
from .models import ApiKey, UserAccount
|
|
|
|
|
|
def _extract_api_key(
|
|
authorization: str | None,
|
|
x_api_key: str | None,
|
|
) -> str | None:
|
|
if x_api_key:
|
|
return x_api_key.strip()
|
|
if authorization and authorization.lower().startswith("bearer "):
|
|
return authorization[7:].strip()
|
|
return None
|
|
|
|
|
|
def current_api_key(
|
|
authorization: str | None = Header(default=None),
|
|
x_api_key: str | None = Header(default=None, alias="X-Api-Key"),
|
|
db: Session = Depends(get_db),
|
|
) -> ApiKey:
|
|
token = _extract_api_key(authorization, x_api_key)
|
|
if not token or not token.startswith(KEY_PREFIX):
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Valid API key required")
|
|
|
|
parsed = parse_api_token(token)
|
|
if parsed:
|
|
key_id, _ = parsed
|
|
api_key = db.scalar(
|
|
select(ApiKey)
|
|
.options(joinedload(ApiKey.user))
|
|
.where(ApiKey.id == key_id, ApiKey.revoked_at.is_(None))
|
|
)
|
|
if api_key and verify_api_token(token, api_key.key_hash):
|
|
api_key.last_used_at = datetime.now(timezone.utc)
|
|
db.commit()
|
|
return api_key
|
|
|
|
active_keys = db.scalars(
|
|
select(ApiKey)
|
|
.options(joinedload(ApiKey.user))
|
|
.where(ApiKey.revoked_at.is_(None))
|
|
).all()
|
|
for api_key in active_keys:
|
|
if verify_api_token(token, api_key.key_hash):
|
|
api_key.last_used_at = datetime.now(timezone.utc)
|
|
db.commit()
|
|
return api_key
|
|
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key")
|
|
|
|
|
|
def current_api_user(api_key: ApiKey = Depends(current_api_key)) -> UserAccount:
|
|
if api_key.user.status != "active":
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Account unavailable")
|
|
return api_key.user
|