API key come hash SHA-256, una per sessione, TTL rinnovabile (#41)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-06 11:31:29 +02:00
parent 1d93eded27
commit 35f46926a9
14 changed files with 343 additions and 110 deletions
+21 -4
View File
@@ -1,10 +1,17 @@
"""API Key authentication dependency for FastAPI."""
from datetime import timedelta
from fastapi import Depends, HTTPException, Request, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.backend.database import get_db
from src.backend.config import settings
from src.backend.models.orm.api_key import ApiKey
from src.backend.models.orm.user import User
from src.backend.services.auth_service import hash_api_key, utcnow
_RENEW_INTERVAL = timedelta(minutes=5)
async def get_current_user(
@@ -24,15 +31,25 @@ async def get_current_user(
)
result = await db.execute(
select(User).where(User.api_key == api_key, User.active == True)
select(ApiKey, User)
.join(User, User.id == ApiKey.user_id)
.where(ApiKey.key_hash == hash_api_key(api_key), User.active == True)
)
user = result.scalar_one_or_none()
row = result.first()
now = utcnow()
if user is None:
if row is None or row[0].expires_at <= now:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or inactive API key",
detail="Invalid or expired API key",
)
key_row, user = row
# Sliding TTL, written at most once per _RENEW_INTERVAL to spare the DB.
if key_row.last_used_at is None or now - key_row.last_used_at > _RENEW_INTERVAL:
key_row.last_used_at = now
key_row.expires_at = now + timedelta(hours=settings.api_key_ttl_hours)
await db.flush()
# Store user_id in request state for access logging middleware
request.state.user_id = user.id