63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Protocol
|
|
from uuid import uuid4
|
|
|
|
import jwt
|
|
|
|
from app.core.config import Settings
|
|
from app.core.contracts import RequestContext
|
|
from app.core.errors import UnauthorizedAgentError
|
|
|
|
|
|
class RevocationStore(Protocol):
|
|
def is_revoked(self, jti: str) -> bool: ...
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EmptyRevocationStore:
|
|
def is_revoked(self, jti: str) -> bool:
|
|
return False
|
|
|
|
|
|
class JwtAuthenticator:
|
|
def __init__(self, settings: Settings, revocation_store: RevocationStore | None = None) -> None:
|
|
self._settings = settings
|
|
self._revocation_store = revocation_store or EmptyRevocationStore()
|
|
self._public_key = self._load_public_key()
|
|
|
|
def _load_public_key(self) -> str:
|
|
path = Path(self._settings.jwt_public_key_path)
|
|
if not path.is_absolute():
|
|
path = Path.cwd() / path
|
|
try:
|
|
return path.read_text(encoding="utf-8")
|
|
except OSError as exc:
|
|
raise RuntimeError(f"JWT public key cannot be read: {path}") from exc
|
|
|
|
def authenticate(self, token: str) -> RequestContext:
|
|
if not token:
|
|
raise UnauthorizedAgentError("missing bearer token")
|
|
try:
|
|
claims = jwt.decode(
|
|
token,
|
|
self._public_key,
|
|
algorithms=[self._settings.jwt_algorithm],
|
|
issuer=self._settings.jwt_issuer,
|
|
audience=self._settings.jwt_audience,
|
|
leeway=self._settings.jwt_clock_skew_seconds,
|
|
options={"require": ["sub", "iss", "aud", "exp", "nbf", "jti"]},
|
|
)
|
|
except jwt.PyJWTError as exc:
|
|
raise UnauthorizedAgentError("invalid bearer token") from exc
|
|
|
|
jti = str(claims["jti"])
|
|
if self._revocation_store.is_revoked(jti):
|
|
raise UnauthorizedAgentError("revoked bearer token")
|
|
subject = claims["sub"]
|
|
if (not isinstance(subject, str) or not subject.isascii()
|
|
or not subject.isdecimal() or len(subject) > 20
|
|
or not 0 < int(subject) <= 18446744073709551615):
|
|
raise UnauthorizedAgentError("invalid subject")
|
|
return RequestContext(user_id=str(claims["sub"]), trace_id=str(uuid4()))
|