fix: harden public api token storage

This commit is contained in:
Christian Krakau-Louis
2026-05-23 17:47:13 +02:00
parent ee663afffb
commit e2b95f183d
6 changed files with 23 additions and 11 deletions
+1 -2
View File
@@ -240,8 +240,7 @@ def require_api_token_scope(required_scope: str) -> Callable:
token = find_api_token(db, api_key)
if token is None:
suffix = api_key[-8:] if len(api_key) >= 8 else "invalid"
logger.warning("Invalid public API token attempt: ...%s", suffix)
logger.warning("Invalid public API token attempt")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid API token",
+1 -1
View File
@@ -12,7 +12,7 @@ class APIToken(Base):
id = Column(Integer, primary_key=True, index=True)
name = Column(String(120), nullable=False)
key_hash = Column(String(64), unique=True, nullable=False, index=True)
key_hash = Column(String(255), unique=True, nullable=False, index=True)
key_prefix = Column(String(16), nullable=False, index=True)
scopes = Column(Text, nullable=False)
active = Column(Boolean, default=True, nullable=False, index=True)
+17 -5
View File
@@ -2,12 +2,12 @@
from __future__ import annotations
import hashlib
import secrets
from dataclasses import dataclass
from datetime import datetime
from typing import Iterable, List, Optional, Set
import bcrypt
from sqlalchemy.orm import Session
from app.models.api_token import APIToken
@@ -59,7 +59,15 @@ def generate_public_api_key() -> str:
def hash_api_key(secret: str) -> str:
"""Hash an API token for database storage."""
return hashlib.sha256(secret.encode("utf-8")).hexdigest()
return bcrypt.hashpw(secret.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
def verify_api_key_secret(secret: str, hashed_secret: str) -> bool:
"""Return True when a raw API token matches the stored hash."""
try:
return bcrypt.checkpw(secret.encode("utf-8"), hashed_secret.encode("utf-8"))
except ValueError:
return False
def create_api_token(db: Session, *, name: str, scopes: Iterable[str]) -> CreatedAPIToken:
@@ -85,11 +93,15 @@ def find_api_token(db: Session, secret: str) -> Optional[APIToken]:
"""Return the active token row matching *secret*, if any."""
if not secret:
return None
return (
candidates = (
db.query(APIToken)
.filter(APIToken.key_hash == hash_api_key(secret), APIToken.active == True) # noqa: E712
.first()
.filter(APIToken.key_prefix == secret[:12], APIToken.active == True) # noqa: E712
.all()
)
for token in candidates:
if verify_api_key_secret(secret, token.key_hash):
return token
return None
def record_api_token_use(db: Session, token: APIToken, *, ip_address: Optional[str]) -> None: