FastAPI Auth: API Key Authentication for Machine Clients

import hashlib
import secrets


def generate_api_key() -> tuple[str, str]:
    """Generate an API key and its hash.

    Returns:
        (plaintext_key, hashed_key) — show plaintext to the user ONCE,
        store only the hash.
    """
    # 32 bytes = 256 bits of entropy, URL-safe encoding
    raw_key = secrets.token_urlsafe(32)

    # Prefix for easy identification in logs and key rotation
    plaintext = f"cdk_{raw_key}"  # "cdk" = CloudDocs Key

    # Hash for storage — SHA-256 is fine here (not a password)
    hashed = hashlib.sha256(plaintext.encode()).hexdigest()

    return plaintext, hashed

from __future__ import annotations
from datetime import datetime
from uuid import UUID, uuid4

from sqlalchemy import ForeignKey, String, DateTime, JSON, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

from src.db.base import Base


class APIKey(Base):
    __tablename__ = "api_keys"

    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
    name: Mapped[str] = mapped_column(String(100))  # Human-readable label
    hashed_key: Mapped[str] = mapped_column(String(64), unique=True, index=True)
    key_prefix: Mapped[str] = mapped_column(String(12))  # First 8 chars for display: "cdk_a1b2..."
    scopes: Mapped[dict] = mapped_column(JSON, default=list)  # ["documents:read", "documents:write"]
    owner_id: Mapped[UUID] = mapped_column(ForeignKey("users.id"))
    is_active: Mapped[bool] = mapped_column(default=True)
    expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    last_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

    owner: Mapped["User"] = relationship(back_populates="api_keys")

from fastapi import APIRouter, Depends, HTTPException, status
from src.core.api_keys import generate_api_key
from src.api.deps import get_current_user

router = APIRouter(prefix="/api-keys", tags=["api-keys"])

VALID_SCOPES = {
    "documents:read", "documents:write", "documents:delete",
    "users:read", "users:write",
    "analytics:read",
}


@router.post("/", status_code=status.HTTP_201_CREATED)
async def create_api_key(
    body: CreateAPIKeyRequest,
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    # Validate scopes
    invalid = set(body.scopes) - VALID_SCOPES
    if invalid:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Invalid scopes: {invalid}",
        )

    plaintext, hashed = generate_api_key()

    api_key = APIKey(
        name=body.name,
        hashed_key=hashed,
        key_prefix=plaintext[:12],
        scopes=body.scopes,
        owner_id=user.id,
        expires_at=body.expires_at,
    )
    db.add(api_key)
    await db.commit()

    # ⚠️ This is the ONLY time the full key is returned
    return {
        "id": str(api_key.id),
        "name": api_key.name,
        "key": plaintext,  # Show once, never again
        "scopes": api_key.scopes,
        "expires_at": api_key.expires_at,
        "warning": "Store this key securely. It will not be shown again.",
    }

import hashlib
from fastapi import Depends, HTTPException, Security, status
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)


async def get_api_key_user(
    api_key: str | None = Security(api_key_header),
    db: AsyncSession = Depends(get_db),
) -> tuple[User, APIKey]:
    if not api_key:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing API key",
        )

    # Hash the provided key and look up
    hashed = hashlib.sha256(api_key.encode()).hexdigest()

    result = await db.execute(
        select(APIKey)
        .where(APIKey.hashed_key == hashed, APIKey.is_active == True)
        .options(joinedload(APIKey.owner))
    )
    key_record = result.scalar_one_or_none()

    if not key_record:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key")

    # Check expiry
    if key_record.expires_at and key_record.expires_at < datetime.now(timezone.utc):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="API key expired")

    # Update last_used_at (fire and forget — do not block the request)
    key_record.last_used_at = datetime.now(timezone.utc)
    await db.commit()

    return key_record.owner, key_record

import redis.asyncio as redis
from fastapi import HTTPException, status

# Default limits — override per key in the database
DEFAULT_RATE_LIMIT = 100   # requests
DEFAULT_RATE_WINDOW = 60   # per 60 seconds


async def check_api_key_rate_limit(
    key_id: str,
    redis_client: redis.Redis,
    limit: int = DEFAULT_RATE_LIMIT,
    window: int = DEFAULT_RATE_WINDOW,
):
    redis_key = f"api_rate:{key_id}"
    current = await redis_client.incr(redis_key)

    if current == 1:
        await redis_client.expire(redis_key, window)

    if current > limit:
        ttl = await redis_client.ttl(redis_key)
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Rate limit exceeded",
            headers={
                "Retry-After": str(ttl),
                "X-RateLimit-Limit": str(limit),
                "X-RateLimit-Remaining": "0",
                "X-RateLimit-Reset": str(ttl),
            },
        )

    return {
        "limit": limit,
        "remaining": max(0, limit - current),
        "reset": await redis_client.ttl(redis_key),
    }