v6ole
51 lines
1.4 KiB
Python
51 lines
1.4 KiB
Python
"""Auth utilities — API key verification."""
|
|
|
|
import hmac
|
|
from typing import Optional
|
|
|
|
from fastapi import Header, HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.database import get_db
|
|
from app.models.project import Project
|
|
|
|
|
|
async def verify_api_key(
|
|
authorization: Optional[str] = Header(None),
|
|
) -> str:
|
|
"""Extract and verify the API key from the Authorization header.
|
|
|
|
Returns the project_id on success.
|
|
"""
|
|
if not authorization:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Missing Authorization header",
|
|
)
|
|
|
|
scheme, _, key = authorization.partition(" ")
|
|
if scheme.lower() != "bearer" or not key.strip():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid Authorization format. Use: Bearer <api_key>",
|
|
)
|
|
|
|
return key.strip()
|
|
|
|
|
|
async def resolve_project_by_api_key(
|
|
db: AsyncSession, api_key: str
|
|
) -> Project:
|
|
"""Look up a project by its API key."""
|
|
result = await db.execute(
|
|
select(Project).where(Project.api_key == api_key, Project.is_active == True)
|
|
)
|
|
project = result.scalar_one_or_none()
|
|
if not project:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid or inactive API key",
|
|
)
|
|
return project
|