Codex Lead Engineer
80 lines
3.1 KiB
Python
80 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
from dataclasses import dataclass
|
|
from typing import Mapping, Protocol
|
|
|
|
from .errors import AuthenticationError
|
|
from .models import Principal
|
|
|
|
|
|
class AuthProvider(Protocol):
|
|
def authenticate(self, authorization_header: str | None) -> Principal:
|
|
"""Authenticate a request without trusting actor parameters."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TokenIdentity:
|
|
actor_id: str
|
|
auth_subject: str
|
|
display_name: str
|
|
|
|
|
|
class StaticTokenAuthProvider:
|
|
"""Small deployment auth adapter intended for local and controlled use.
|
|
|
|
Tokens are compared using SHA-256 digests and constant-time comparison.
|
|
Production deployments should replace this adapter with an OIDC/JWT or
|
|
mTLS-backed implementation and keep the same ``AuthProvider`` contract.
|
|
"""
|
|
|
|
def __init__(self, token_identities: Mapping[str, TokenIdentity]):
|
|
if not token_identities:
|
|
raise ValueError("At least one authentication token is required")
|
|
if any(len(token) < 24 for token in token_identities):
|
|
raise ValueError("Every authentication token must contain at least 24 characters")
|
|
self._digests = {
|
|
hashlib.sha256(token.encode("utf-8")).digest(): identity
|
|
for token, identity in token_identities.items()
|
|
}
|
|
|
|
@classmethod
|
|
def from_environment(cls) -> "StaticTokenAuthProvider":
|
|
raw = os.environ.get("PROJECT_BUS_TOKENS_JSON")
|
|
bootstrap = os.environ.get("PROJECT_BUS_BOOTSTRAP_TOKEN")
|
|
if not bootstrap:
|
|
raise ValueError("PROJECT_BUS_BOOTSTRAP_TOKEN is required")
|
|
if len(bootstrap) < 24:
|
|
raise ValueError("PROJECT_BUS_BOOTSTRAP_TOKEN must contain at least 24 characters")
|
|
entries: dict[str, TokenIdentity] = {
|
|
bootstrap: TokenIdentity("system", "system", "Bootstrap system")
|
|
}
|
|
if raw:
|
|
decoded = json.loads(raw)
|
|
if not isinstance(decoded, dict):
|
|
raise ValueError("PROJECT_BUS_TOKENS_JSON must be a JSON object")
|
|
for token, identity in decoded.items():
|
|
entries[token] = TokenIdentity(
|
|
actor_id=identity["actor_id"],
|
|
auth_subject=identity.get("auth_subject", identity["actor_id"]),
|
|
display_name=identity.get("display_name", identity["actor_id"]),
|
|
)
|
|
return cls(entries)
|
|
|
|
def authenticate(self, authorization_header: str | None) -> Principal:
|
|
if not authorization_header or not authorization_header.startswith("Bearer "):
|
|
raise AuthenticationError("A Bearer token is required")
|
|
token = authorization_header[7:]
|
|
candidate = hashlib.sha256(token.encode("utf-8")).digest()
|
|
for expected, identity in self._digests.items():
|
|
if hmac.compare_digest(candidate, expected):
|
|
return Principal(
|
|
actor_id=identity.actor_id,
|
|
auth_subject=identity.auth_subject,
|
|
display_name=identity.display_name,
|
|
)
|
|
raise AuthenticationError("Invalid Bearer token")
|