"""DEV / TEST ONLY — a local issuer and a stand-in for the Ignite gateway, for running the LIVE runner against a
composition served on a laptop (tier 1 against a real DataK3 bucket) when the bucket has no user pool.
Deployed, a person's token comes from the composition's dodil-appid pool and the Ignite gateway verifies it
(EdDSA, `iss` AND `aud` mandatory), strips any inbound identity headers and injects X-Dodil-User /
X-Dodil-User-Jwt / X-Dodil-Auth-Source. Here a LocalIssuer mints the personas' tokens from the role
catalogue and LocalGateway does the gateway's job for bearers, with the same rules: issuer and audience are
required, the expiry is checked, inbound identity headers are always stripped, and no bearer means no identity
(the app answers 401).
HS256 with a key from the secrets directory, not EdDSA: this stands in for a gateway to exercise the APP;
it is not a model of appid's keys, and it adds no dependency. Never serve it in front of a deployment.
"""
import base64
import hashlib
import hmac
import json
import secrets
import time
from typing import Any
_STRIP = {b"x-dodil-user", b"x-dodil-user-jwt", b"x-dodil-auth-source", b"authorization"}
def _b64(raw: bytes) -> str:
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
def _unb64(text: str) -> bytes:
return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))
class InvalidToken(Exception):
pass
class LocalIssuer:
"""Mints and verifies the personas' tokens (HS256) for one issuer and audience."""
def __init__(self, key: bytes, *, issuer: str, audience: str, ttl: int = 3600):
if len(key) < 32:
raise ValueError("a signing key of at least 32 bytes")
self._key, self.issuer, self.audience, self.ttl = key, issuer, audience, ttl
@staticmethod
def new_key() -> bytes:
return secrets.token_urlsafe(48).encode()
@classmethod
def from_pem(cls, raw: bytes, **kw) -> "LocalIssuer":
"""The key file's content (any text of ≥ 32 bytes; whitespace trimmed)."""
return cls(raw.strip(), **kw)
def mint(self, claims: dict) -> str:
now = int(time.time())
payload = {**claims, "iss": self.issuer, "aud": self.audience, "iat": now, "exp": now + self.ttl}
head = _b64(json.dumps({"alg": "HS256", "typ": "JWT"}).encode())
body = _b64(json.dumps(payload).encode())
sig = _b64(hmac.new(self._key, f"{head}.{body}".encode(), hashlib.sha256).digest())
return f"{head}.{body}.{sig}"
def verify(self, token: str) -> dict:
try:
head, body, sig = token.split(".")
header = json.loads(_unb64(head))
claims = json.loads(_unb64(body))
except Exception: # noqa: BLE001
raise InvalidToken("malformed token") from None
if header.get("alg") != "HS256":
raise InvalidToken("unexpected algorithm")
want = _b64(hmac.new(self._key, f"{head}.{body}".encode(), hashlib.sha256).digest())
if not hmac.compare_digest(want, sig):
raise InvalidToken("bad signature")
for k in ("iss", "aud", "exp", "sub"):
if k not in claims:
raise InvalidToken(f"no {k}")
if claims["iss"] != self.issuer or claims["aud"] != self.audience:
raise InvalidToken("wrong issuer or audience")
if claims["exp"] <= time.time():
raise InvalidToken("expired")
return claims
class LocalGateway:
"""ASGI middleware doing the gateway's job for bearer tokens a LocalIssuer minted."""
def __init__(self, app: Any, issuer: LocalIssuer):
self.app, self.issuer = app, issuer
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
return await self.app(scope, receive, send)
headers = [(k, v) for k, v in scope["headers"] if k not in _STRIP]
auth = dict(scope["headers"]).get(b"authorization", b"").decode()
if auth.lower().startswith("bearer "):
token = auth[7:].strip()
try:
claims = self.issuer.verify(token)
except InvalidToken as exc:
body = json.dumps({"detail": f"invalid token: {exc}"}).encode()
await send({"type": "http.response.start", "status": 401,
"headers": [(b"content-type", b"application/json"), (b"www-authenticate", b"Bearer")]})
return await send({"type": "http.response.body", "body": body})
user = {"sub": claims["sub"], "email": claims.get("email", ""), "connection": claims.get("connection"),
"app_roles": claims.get("app_roles", [])}
headers += [(b"x-dodil-user", json.dumps(user).encode()), (b"x-dodil-user-jwt", token.encode()),
(b"x-dodil-auth-source", b"pool")]
await self.app({**scope, "headers": headers}, receive, send)