100 lines
2.7 KiB
Python
100 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta
|
|
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from govoplan_access.backend.auth.tokens import generate_secret, hash_secret, verify_secret
|
|
from govoplan_access.backend.db.models import ApiKey, User
|
|
from govoplan_core.security.time import ensure_aware_utc, utc_now
|
|
|
|
API_KEY_PREFIX_LENGTH = 12
|
|
API_KEY_RANDOM_BYTES = 32
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class CreatedApiKey:
|
|
model: ApiKey
|
|
secret: str
|
|
|
|
|
|
def hash_api_key(secret: str) -> str:
|
|
return hash_secret(secret)
|
|
|
|
|
|
def verify_api_key(secret: str, expected_hash: str) -> bool:
|
|
return verify_secret(secret, expected_hash)
|
|
|
|
|
|
def generate_api_key_secret() -> str:
|
|
return generate_secret("mm_", random_bytes=API_KEY_RANDOM_BYTES)
|
|
|
|
|
|
def api_key_prefix(secret: str) -> str:
|
|
# Prefix is only a lookup helper and must not be enough to authenticate.
|
|
return secret[:API_KEY_PREFIX_LENGTH]
|
|
|
|
|
|
def create_api_key(
|
|
session: Session,
|
|
*,
|
|
user: User,
|
|
name: str,
|
|
scopes: list[str],
|
|
secret: str | None = None,
|
|
expires_at: datetime | None = None,
|
|
) -> CreatedApiKey:
|
|
secret = secret or generate_api_key_secret()
|
|
model = ApiKey(
|
|
tenant_id=user.tenant_id,
|
|
user_id=user.id,
|
|
name=name,
|
|
prefix=api_key_prefix(secret),
|
|
key_hash=hash_api_key(secret),
|
|
scopes=scopes,
|
|
expires_at=expires_at,
|
|
)
|
|
session.add(model)
|
|
session.flush()
|
|
return CreatedApiKey(model=model, secret=secret)
|
|
|
|
|
|
def authenticate_api_key(
|
|
session: Session,
|
|
secret: str,
|
|
*,
|
|
touch_interval_seconds: int = 5 * 60,
|
|
) -> ApiKey | None:
|
|
prefix = api_key_prefix(secret)
|
|
candidates = (
|
|
session.query(ApiKey)
|
|
.options(joinedload(ApiKey.user).joinedload(User.account))
|
|
.filter(
|
|
ApiKey.prefix == prefix,
|
|
ApiKey.revoked_at.is_(None),
|
|
)
|
|
.all()
|
|
)
|
|
now = utc_now()
|
|
for candidate in candidates:
|
|
expires_at = ensure_aware_utc(candidate.expires_at)
|
|
if expires_at and expires_at < now:
|
|
continue
|
|
if verify_api_key(secret, candidate.key_hash):
|
|
last_used_at = ensure_aware_utc(candidate.last_used_at)
|
|
if (
|
|
touch_interval_seconds <= 0
|
|
or last_used_at is None
|
|
or now - last_used_at >= timedelta(seconds=touch_interval_seconds)
|
|
):
|
|
candidate.last_used_at = now
|
|
session.add(candidate)
|
|
return candidate
|
|
return None
|
|
|
|
|
|
def has_scope(api_key: ApiKey, required_scope: str) -> bool:
|
|
scopes = set(api_key.scopes or [])
|
|
return "*" in scopes or required_scope in scopes
|