API reference¶

Generated from the apogee-auth-runtime source with mkdocstrings. Every symbol below is exported from apogee_auth_runtime, so it is part of the supported public surface.

Domain¶

JwtKeyPair `dataclass` ¶

Python

JwtKeyPair(active_kid: str, active_secret: str, legacy_secrets: dict[str, str] = None)

Active signing key + previously valid keys (rotation).

kid (key id) is added to JWT headers so clients holding tokens signed with older keys can still validate during the rotation window.

active_kid `instance-attribute` ¶

Python

active_kid: str

active_secret `instance-attribute` ¶

Python

active_secret: str

legacy_secrets `class-attribute` `instance-attribute` ¶

Python

legacy_secrets: dict[str, str] = None

secret_for ¶

Python

secret_for(kid: str) -> str | None

Source code in apogee_auth_runtime/domain/value_objects/jwt_keypair.py

Python

def secret_for(self, kid: str) -> str | None:
    if kid == self.active_kid:
        return self.active_secret
    return self.legacy_secrets.get(kid)

Domain · Exceptions¶

AuthError ¶

Bases: Exception

Base for apogee-auth-runtime errors.

InvalidTokenError ¶

Bases: AuthError

TokenRevokedError ¶

Bases: AuthError

Domain · Protocols (ports)¶

IJwtService ¶

Bases: Protocol

name `instance-attribute` ¶

Python

name: str

sign ¶

Python

sign(subject: str, ttl_s: int = 900, token_type: str = 'access', extra: dict[str, Any] | None = None) -> str

Source code in apogee_auth_runtime/domain/services/i_jwt_service.py

Python

def sign(
    self,
    subject: str,
    ttl_s: int = 900,
    token_type: str = "access",
    extra: dict[str, Any] | None = None,
) -> str: ...

verify ¶

Python

verify(token: str, expected_type: str | None = None) -> dict[str, Any]

Source code in apogee_auth_runtime/domain/services/i_jwt_service.py

Python

def verify(
    self, token: str, expected_type: str | None = None
) -> dict[str, Any]: ...

revoke `async` ¶

Python

revoke(token: str) -> None

Source code in apogee_auth_runtime/domain/services/i_jwt_service.py

Python

async def revoke(self, token: str) -> None: ...

IPasswordHasher ¶

Bases: Protocol

name `instance-attribute` ¶

Python

name: str

hash ¶

Python

hash(password: str) -> str

Source code in apogee_auth_runtime/domain/services/i_password_hasher.py

Python

def hash(self, password: str) -> str: ...

verify ¶

Python

verify(password: str, hashed: str) -> bool

Source code in apogee_auth_runtime/domain/services/i_password_hasher.py

Python

def verify(self, password: str, hashed: str) -> bool: ...

ITokenBlacklist ¶

Bases: Protocol

add `async` ¶

Python

add(token_id: str, ttl_s: int) -> None

Source code in apogee_auth_runtime/domain/services/i_token_blacklist.py

Python

async def add(self, token_id: str, ttl_s: int) -> None: ...

is_revoked `async` ¶

Python

is_revoked(token_id: str) -> bool

Source code in apogee_auth_runtime/domain/services/i_token_blacklist.py

Python

async def is_revoked(self, token_id: str) -> bool: ...

clear `async` ¶

Python

clear() -> None

Source code in apogee_auth_runtime/domain/services/i_token_blacklist.py

Python

async def clear(self) -> None: ...

Infrastructure¶

Argon2PasswordHasher ¶

Python

Argon2PasswordHasher(time_cost: int = 3, memory_cost: int = 65536, parallelism: int = 2, hash_len: int = 32)

Source code in apogee_auth_runtime/infrastructure/hashers/argon2_password_hasher.py

Python

def __init__(
    self,
    time_cost: int = 3,
    memory_cost: int = 65_536,  # 64 MiB
    parallelism: int = 2,
    hash_len: int = 32,
) -> None:
    self._time_cost = time_cost
    self._memory_cost = memory_cost
    self._parallelism = parallelism
    self._hash_len = hash_len
    self._hasher = None

name `class-attribute` `instance-attribute` ¶

Python

name = 'argon2id'

hash ¶

Python

hash(password: str) -> str

Source code in apogee_auth_runtime/infrastructure/hashers/argon2_password_hasher.py

Python

def hash(self, password: str) -> str:
    return self._ensure_hasher().hash(password)

verify ¶

Python

verify(password: str, hashed: str) -> bool

Source code in apogee_auth_runtime/infrastructure/hashers/argon2_password_hasher.py

Python

def verify(self, password: str, hashed: str) -> bool:
    try:
        return bool(self._ensure_hasher().verify(hashed, password))
    except Exception:  # noqa: BLE001
        return False

BcryptPasswordHasher ¶

Python

BcryptPasswordHasher(rounds: int = 12)

Source code in apogee_auth_runtime/infrastructure/hashers/bcrypt_password_hasher.py

Python

def __init__(self, rounds: int = 12) -> None:
    self._rounds = rounds
    self._bcrypt = None

name `class-attribute` `instance-attribute` ¶

Python

name = 'bcrypt'

hash ¶

Python

hash(password: str) -> str

Source code in apogee_auth_runtime/infrastructure/hashers/bcrypt_password_hasher.py

Python

def hash(self, password: str) -> str:
    bcrypt = self._ensure()
    salt = bcrypt.gensalt(rounds=self._rounds)
    return bcrypt.hashpw(password.encode("utf-8"), salt).decode("utf-8")

verify ¶

Python

verify(password: str, hashed: str) -> bool

Source code in apogee_auth_runtime/infrastructure/hashers/bcrypt_password_hasher.py

Python

def verify(self, password: str, hashed: str) -> bool:
    bcrypt = self._ensure()
    try:
        return bool(bcrypt.checkpw(password.encode("utf-8"), hashed.encode("utf-8")))
    except Exception:  # noqa: BLE001
        return False

HS256JwtService ¶

Python

HS256JwtService(keypair: JwtKeyPair, issuer: str = 'apogee-auth-runtime', blacklist: Any | None = None, clock: Any | None = None)

JWT signer/verifier using HS256 with key rotation + optional blacklist.

Source code in apogee_auth_runtime/infrastructure/jwt/hs256_jwt_service.py

Python

def __init__(
    self,
    keypair: JwtKeyPair,
    issuer: str = "apogee-auth-runtime",
    blacklist: Any | None = None,
    clock: Any | None = None,
) -> None:
    self._keys = keypair
    self._issuer = issuer
    self._blacklist = blacklist
    self._now = clock or time.time

name `class-attribute` `instance-attribute` ¶

Python

name = 'hs256'

sign ¶

Python

sign(subject: str, ttl_s: int = 900, token_type: str = 'access', extra: dict[str, Any] | None = None) -> str

Source code in apogee_auth_runtime/infrastructure/jwt/hs256_jwt_service.py

Python

def sign(
    self,
    subject: str,
    ttl_s: int = 900,
    token_type: str = "access",
    extra: dict[str, Any] | None = None,
) -> str:
    if not subject:
        raise ValueError("subject cannot be empty")
    if ttl_s <= 0:
        raise ValueError("ttl_s must be positive")
    now = int(self._now())
    payload: dict[str, Any] = {
        "iss": self._issuer,
        "sub": subject,
        "iat": now,
        "exp": now + ttl_s,
        "jti": uuid.uuid4().hex,
        "type": token_type,
    }
    if extra:
        payload.update(extra)
    return _hs256.encode(payload, self._keys.active_secret, kid=self._keys.active_kid)

verify ¶

Python

verify(token: str, expected_type: str | None = None) -> dict[str, Any]

Source code in apogee_auth_runtime/infrastructure/jwt/hs256_jwt_service.py

Python

def verify(self, token: str, expected_type: str | None = None) -> dict[str, Any]:
    try:
        header = _hs256.decode_unsafe_header(token)
    except Exception as exc:
        raise InvalidTokenError(f"malformed token: {exc}") from exc

    kid = header.get("kid")
    secret = self._keys.secret_for(kid) if kid else self._keys.active_secret
    if secret is None:
        raise InvalidTokenError(f"unknown kid: {kid!r}")

    try:
        payload = _hs256.decode(token, secret)
    except Exception as exc:
        raise InvalidTokenError(str(exc)) from exc

    # exp check
    exp = payload.get("exp")
    if exp is not None and int(self._now()) >= int(exp):
        raise InvalidTokenError("token expired")

    if expected_type is not None and payload.get("type") != expected_type:
        raise InvalidTokenError(
            f"expected token_type={expected_type!r}, got {payload.get('type')!r}"
        )

    if self._blacklist is not None:
        jti = payload.get("jti")
        # blacklist may be sync (in-memory) or async (Redis).
        # we accept both; sync path used here for verify().
        is_revoked = self._blacklist.is_revoked_sync(jti) if hasattr(self._blacklist, "is_revoked_sync") else None
        if is_revoked:
            raise TokenRevokedError("token revoked")
    return payload

revoke `async` ¶

Python

revoke(token: str) -> None

Source code in apogee_auth_runtime/infrastructure/jwt/hs256_jwt_service.py

Python

async def revoke(self, token: str) -> None:
    if self._blacklist is None:
        return
    try:
        payload = self.verify(token)
    except InvalidTokenError:
        return
    jti = payload.get("jti")
    if jti is None:
        return
    ttl_s = max(int(payload.get("exp", self._now())) - int(self._now()), 1)
    await self._blacklist.add(jti, ttl_s)

InMemoryTokenBlacklist ¶

Python

InMemoryTokenBlacklist()

Source code in apogee_auth_runtime/infrastructure/blacklist/in_memory_blacklist.py

Python

def __init__(self) -> None:
    self._revoked: dict[str, _Entry] = {}

name `class-attribute` `instance-attribute` ¶

Python

name = 'in_memory'

add `async` ¶

Python

add(token_id: str, ttl_s: int) -> None

Source code in apogee_auth_runtime/infrastructure/blacklist/in_memory_blacklist.py

Python

async def add(self, token_id: str, ttl_s: int) -> None:
    self._revoked[token_id] = _Entry(expires_at=time.monotonic() + ttl_s)

is_revoked `async` ¶

Python

is_revoked(token_id: str) -> bool

Source code in apogee_auth_runtime/infrastructure/blacklist/in_memory_blacklist.py

Python

async def is_revoked(self, token_id: str) -> bool:
    entry = self._revoked.get(token_id)
    if entry is None:
        return False
    if not self._is_alive(entry):
        del self._revoked[token_id]
        return False
    return True

is_revoked_sync ¶

Python

is_revoked_sync(token_id: str) -> bool

Sync façade used by HS256JwtService.verify (which is sync).

Source code in apogee_auth_runtime/infrastructure/blacklist/in_memory_blacklist.py

Python

def is_revoked_sync(self, token_id: str) -> bool:
    """Sync façade used by HS256JwtService.verify (which is sync)."""

    entry = self._revoked.get(token_id)
    if entry is None:
        return False
    if not self._is_alive(entry):
        del self._revoked[token_id]
        return False
    return True

clear `async` ¶

Python

clear() -> None

Source code in apogee_auth_runtime/infrastructure/blacklist/in_memory_blacklist.py

Python

async def clear(self) -> None:
    self._revoked.clear()

JoseJwtService ¶

Python

JoseJwtService(keypair: JwtKeyPair, algorithm: str = 'HS256', issuer: str = 'apogee-auth-runtime', blacklist: Any | None = None)

Wraps PyJWT (preferred) or joserfc.

Supports HS256 (default) and asymmetric algos via private/public keys when caller passes them as active_secret/legacy_secrets.

Source code in apogee_auth_runtime/infrastructure/jwt/jose_jwt_service.py

Python

def __init__(
    self,
    keypair: JwtKeyPair,
    algorithm: str = "HS256",
    issuer: str = "apogee-auth-runtime",
    blacklist: Any | None = None,
) -> None:
    self._keys = keypair
    self._algorithm = algorithm
    self._issuer = issuer
    self._blacklist = blacklist
    self._jwt = None

name `class-attribute` `instance-attribute` ¶

Python

name = 'jose'

sign ¶

Python

sign(subject: str, ttl_s: int = 900, token_type: str = 'access', extra: dict[str, Any] | None = None) -> str

Source code in apogee_auth_runtime/infrastructure/jwt/jose_jwt_service.py

Python

def sign(
    self,
    subject: str,
    ttl_s: int = 900,
    token_type: str = "access",
    extra: dict[str, Any] | None = None,
) -> str:
    import time
    import uuid

    pyjwt = self._ensure_jwt()
    now = int(time.time())
    payload: dict[str, Any] = {
        "iss": self._issuer,
        "sub": subject,
        "iat": now,
        "exp": now + ttl_s,
        "jti": uuid.uuid4().hex,
        "type": token_type,
    }
    if extra:
        payload.update(extra)
    return pyjwt.encode(
        payload,
        self._keys.active_secret,
        algorithm=self._algorithm,
        headers={"kid": self._keys.active_kid},
    )

verify ¶

Python

verify(token: str, expected_type: str | None = None) -> dict[str, Any]

Source code in apogee_auth_runtime/infrastructure/jwt/jose_jwt_service.py

Python

def verify(self, token: str, expected_type: str | None = None) -> dict[str, Any]:
    pyjwt = self._ensure_jwt()
    try:
        header = pyjwt.get_unverified_header(token)
    except Exception as exc:
        raise InvalidTokenError(str(exc)) from exc
    kid = header.get("kid")
    secret = self._keys.secret_for(kid) if kid else self._keys.active_secret
    if secret is None:
        raise InvalidTokenError(f"unknown kid: {kid!r}")
    try:
        payload = pyjwt.decode(
            token, secret, algorithms=[self._algorithm], issuer=self._issuer
        )
    except Exception as exc:
        raise InvalidTokenError(str(exc)) from exc
    if expected_type is not None and payload.get("type") != expected_type:
        raise InvalidTokenError(
            f"expected token_type={expected_type!r}, got {payload.get('type')!r}"
        )
    return payload

revoke `async` ¶

Python

revoke(token: str) -> None

Source code in apogee_auth_runtime/infrastructure/jwt/jose_jwt_service.py

Python

async def revoke(self, token: str) -> None:
    if self._blacklist is None:
        return
    try:
        payload = self.verify(token)
    except InvalidTokenError:
        return
    jti = payload.get("jti")
    if jti is None:
        return
    import time

    ttl_s = max(int(payload.get("exp", time.time())) - int(time.time()), 1)
    await self._blacklist.add(jti, ttl_s)

PlainPasswordHasher ¶

SHA-256 with random salt. NOT recommended for production.

name `class-attribute` `instance-attribute` ¶

Python

name = 'plain'

hash ¶

Python

hash(password: str) -> str

Source code in apogee_auth_runtime/infrastructure/hashers/plain_password_hasher.py

Python

def hash(self, password: str) -> str:
    salt = os.urandom(16).hex()
    digest = hashlib.sha256(f"{salt}:{password}".encode("utf-8")).hexdigest()
    return f"plain${salt}${digest}"

verify ¶

Python

verify(password: str, hashed: str) -> bool

Source code in apogee_auth_runtime/infrastructure/hashers/plain_password_hasher.py

Python

def verify(self, password: str, hashed: str) -> bool:
    try:
        scheme, salt, expected = hashed.split("$", 2)
    except ValueError:
        return False
    if scheme != "plain":
        return False
    actual = hashlib.sha256(f"{salt}:{password}".encode("utf-8")).hexdigest()
    return hmac.compare_digest(actual, expected)

RedisTokenBlacklist ¶

Python

RedisTokenBlacklist(client: Any | None = None, url: str = 'redis://localhost:6379/0', namespace: str = 'auth:revoked')

Source code in apogee_auth_runtime/infrastructure/blacklist/redis_blacklist.py

Python

def __init__(
    self,
    client: Any | None = None,
    url: str = "redis://localhost:6379/0",
    namespace: str = "auth:revoked",
) -> None:
    self._url = url
    self._namespace = namespace
    self._client = client

name `class-attribute` `instance-attribute` ¶

Python

name = 'redis'

add `async` ¶

Python

add(token_id: str, ttl_s: int) -> None

Source code in apogee_auth_runtime/infrastructure/blacklist/redis_blacklist.py

Python

async def add(self, token_id: str, ttl_s: int) -> None:
    client = await self._ensure_client()
    await client.set(self._key(token_id), b"1", ex=int(ttl_s))

is_revoked `async` ¶

Python

is_revoked(token_id: str) -> bool

Source code in apogee_auth_runtime/infrastructure/blacklist/redis_blacklist.py

Python

async def is_revoked(self, token_id: str) -> bool:
    client = await self._ensure_client()
    return bool(await client.exists(self._key(token_id)))

clear `async` ¶

Python

clear() -> None

Source code in apogee_auth_runtime/infrastructure/blacklist/redis_blacklist.py

Python

async def clear(self) -> None:
    client = await self._ensure_client()
    async for key in client.scan_iter(f"{self._namespace}:*"):
        await client.delete(key)

API reference¶

Domain¶

JwtKeyPair dataclass ¶

active_kid instance-attribute ¶

active_secret instance-attribute ¶

legacy_secrets class-attribute instance-attribute ¶

secret_for ¶

Domain · Exceptions¶

AuthError ¶

InvalidTokenError ¶

TokenRevokedError ¶

Domain · Protocols (ports)¶

IJwtService ¶

name instance-attribute ¶

sign ¶

verify ¶

revoke async ¶

IPasswordHasher ¶

name instance-attribute ¶

hash ¶

verify ¶

ITokenBlacklist ¶

add async ¶

is_revoked async ¶

clear async ¶

Infrastructure¶

Argon2PasswordHasher ¶

name class-attribute instance-attribute ¶

hash ¶

verify ¶

BcryptPasswordHasher ¶

name class-attribute instance-attribute ¶

hash ¶

verify ¶

HS256JwtService ¶

name class-attribute instance-attribute ¶

sign ¶

verify ¶

revoke async ¶

InMemoryTokenBlacklist ¶

name class-attribute instance-attribute ¶

add async ¶

is_revoked async ¶

is_revoked_sync ¶

clear async ¶

JoseJwtService ¶

name class-attribute instance-attribute ¶

sign ¶

verify ¶

revoke async ¶

PlainPasswordHasher ¶

name class-attribute instance-attribute ¶

hash ¶

verify ¶

RedisTokenBlacklist ¶

name class-attribute instance-attribute ¶

add async ¶

is_revoked async ¶

clear async ¶

JwtKeyPair `dataclass` ¶

active_kid `instance-attribute` ¶

active_secret `instance-attribute` ¶

legacy_secrets `class-attribute` `instance-attribute` ¶

name `instance-attribute` ¶

revoke `async` ¶

name `instance-attribute` ¶

add `async` ¶

is_revoked `async` ¶

clear `async` ¶

name `class-attribute` `instance-attribute` ¶

name `class-attribute` `instance-attribute` ¶

name `class-attribute` `instance-attribute` ¶

revoke `async` ¶

name `class-attribute` `instance-attribute` ¶

add `async` ¶

is_revoked `async` ¶

clear `async` ¶

name `class-attribute` `instance-attribute` ¶

revoke `async` ¶

name `class-attribute` `instance-attribute` ¶

name `class-attribute` `instance-attribute` ¶

add `async` ¶

is_revoked `async` ¶

clear `async` ¶