Class tigrbl_auth.jwtoken.JWTCoder

tigrbl_auth.jwtoken.JWTCoder

JWTCoder(arg1, arg2)

Stateless JWT helper backed by JWTTokenService.

JWTCoder historically accepted a private/public key pair and constructed its own :class:JWTTokenService. Recent refactoring switched the constructor to require an already configured service instance. The tests for RFC 9068 still rely on the original behaviour, so the initializer now supports both invocation styles:

JWTCoder(service, kid) -- use the provided service directly.

JWTCoder(private_key_pem, public_key_pem) -- build an ephemeral service from the PEM encoded Ed25519 key pair.

Source code in tigrbl_auth/jwtoken.py

def __init__(self, arg1: JWTTokenService | bytes, arg2: str | bytes):
    if isinstance(arg1, JWTTokenService) and isinstance(arg2, str):
        self._svc = arg1
        self._kid = arg2
        return

    if isinstance(arg1, (bytes, bytearray)) and isinstance(
        arg2, (bytes, bytearray)
    ):
        kp = LocalKeyProvider()
        spec = KeySpec(
            klass=KeyClass.asymmetric,
            alg=KeyAlg.ED25519,
            uses=(KeyUse.SIGN, KeyUse.VERIFY),
            export_policy=ExportPolicy.SECRET_WHEN_ALLOWED,
            label="jwt_ed25519",
        )
        ref = _run(kp.import_key(spec, arg1, public=arg2))
        self._svc = JWTTokenService(kp)
        self._kid = ref.kid
        return

    raise TypeError(
        "JWTCoder requires (JWTTokenService, kid) or (private_pem, public_pem)"
    )

default classmethod

default()

Source code in tigrbl_auth/jwtoken.py

@classmethod
def default(cls) -> "JWTCoder":
    svc, kid = _svc()
    return cls(svc, kid)

async_sign async

async_sign(
    *,
    sub,
    tid=None,
    ttl=_ACCESS_TTL,
    typ="access",
    issuer=None,
    audience=None,
    cert_thumbprint=None,
    **extra,
)

Source code in tigrbl_auth/jwtoken.py

async def async_sign(
    self,
    *,
    sub: str,
    tid: Optional[str] = None,
    ttl: timedelta = _ACCESS_TTL,
    typ: str = "access",
    issuer: Optional[str] = None,
    audience: Optional[Iterable[str] | str] = None,
    cert_thumbprint: Optional[str] = None,
    **extra: Any,
) -> str:
    now = datetime.now(timezone.utc)
    payload: Dict[str, Any] = {
        "sub": sub,
        "typ": typ,
        "iat": int(now.timestamp()),
        "exp": int((now + ttl).timestamp()),
        **extra,
    }
    if tid is not None:
        payload["tid"] = tid
    if settings.enable_rfc8705:
        if cert_thumbprint is None:
            raise ValueError(
                "cert_thumbprint required when RFC 8705 support is enabled",
            )
        payload["cnf"] = {"x5t#S256": cert_thumbprint}
    if settings.enable_rfc9068:
        if issuer is None or audience is None:
            raise ValueError(
                "issuer and audience required when RFC 9068 support is enabled",
            )
        from .rfc.rfc9068 import add_rfc9068_claims

        payload = add_rfc9068_claims(payload, issuer=issuer, audience=audience)
    token = await self._svc.mint(
        payload,
        alg=JWAAlg.EDDSA,
        kid=self._kid,
        lifetime_s=int(ttl.total_seconds()),
        subject=sub,
        issuer=issuer,
        audience=audience,
    )
    if settings.enable_rfc7662:
        claims = {"sub": sub, "kind": typ}
        if tid is not None:
            claims["tid"] = tid
        register_token(token, claims)
    return token

sign

sign(
    *,
    sub,
    tid=None,
    ttl=_ACCESS_TTL,
    typ="access",
    issuer=None,
    audience=None,
    cert_thumbprint=None,
    **extra,
)

Source code in tigrbl_auth/jwtoken.py

def sign(
    self,
    *,
    sub: str,
    tid: Optional[str] = None,
    ttl: timedelta = _ACCESS_TTL,
    typ: str = "access",
    issuer: Optional[str] = None,
    audience: Optional[Iterable[str] | str] = None,
    cert_thumbprint: Optional[str] = None,
    **extra: Any,
) -> str:
    return _run(
        self.async_sign(
            sub=sub,
            tid=tid,
            ttl=ttl,
            typ=typ,
            issuer=issuer,
            audience=audience,
            cert_thumbprint=cert_thumbprint,
            **extra,
        ),
    )

async_sign_pair async

async_sign_pair(*, sub, tid, cert_thumbprint=None, **extra)

Source code in tigrbl_auth/jwtoken.py

async def async_sign_pair(
    self, *, sub: str, tid: str, cert_thumbprint: Optional[str] = None, **extra: Any
) -> Tuple[str, str]:
    access = await self.async_sign(
        sub=sub, tid=tid, cert_thumbprint=cert_thumbprint, **extra
    )
    refresh = await self.async_sign(
        sub=sub,
        tid=tid,
        ttl=_REFRESH_TTL,
        typ="refresh",
        cert_thumbprint=cert_thumbprint,
        **extra,
    )
    return access, refresh

sign_pair

sign_pair(*, sub, tid, cert_thumbprint=None, **extra)

Source code in tigrbl_auth/jwtoken.py

def sign_pair(
    self, *, sub: str, tid: str, cert_thumbprint: Optional[str] = None, **extra: Any
) -> Tuple[str, str]:
    return _run(
        self.async_sign_pair(
            sub=sub,
            tid=tid,
            cert_thumbprint=cert_thumbprint,
            **extra,
        ),
    )

async_decode async

async_decode(
    token,
    verify_exp=True,
    issuer=None,
    audience=None,
    cert_thumbprint=None,
)

Source code in tigrbl_auth/jwtoken.py

async def async_decode(
    self,
    token: str,
    verify_exp: bool = True,
    issuer: Optional[str] = None,
    audience: Optional[Iterable[str] | str] = None,
    cert_thumbprint: Optional[str] = None,
) -> Dict[str, Any]:
    if is_revoked(token):
        raise InvalidTokenError("token has been revoked")
    if _header_alg(token) in {"", "none"}:
        raise InvalidTokenError("unsigned JWTs are not accepted")
    try:
        payload = await self._svc.verify(token, issuer=issuer, audience=audience)
    except Exception as exc:
        # Delegate any verification issues to our own domain specific error
        raise InvalidTokenError("unable to verify token") from exc
    if verify_exp:
        exp = payload.get("exp")
        if exp is not None and int(exp) < int(
            datetime.now(timezone.utc).timestamp()
        ):
            raise InvalidTokenError("token is expired")
    if settings.enable_rfc8705:
        if cert_thumbprint is None:
            raise ValueError(
                "certificate thumbprint required for mTLS per RFC 8705",
            )
        validate_certificate_binding(payload, cert_thumbprint)
    if settings.enable_rfc9068:
        if issuer is None or audience is None:
            raise ValueError(
                "issuer and audience required for JWT access tokens per RFC 9068",
            )
        from .rfc.rfc9068 import validate_rfc9068_claims

        validate_rfc9068_claims(payload, issuer=issuer, audience=audience)
    return payload

decode

decode(
    token,
    verify_exp=True,
    issuer=None,
    audience=None,
    cert_thumbprint=None,
)

Source code in tigrbl_auth/jwtoken.py

def decode(
    self,
    token: str,
    verify_exp: bool = True,
    issuer: Optional[str] = None,
    audience: Optional[Iterable[str] | str] = None,
    cert_thumbprint: Optional[str] = None,
) -> Dict[str, Any]:
    return _run(
        self.async_decode(
            token,
            verify_exp=verify_exp,
            issuer=issuer,
            audience=audience,
            cert_thumbprint=cert_thumbprint,
        ),
    )

refresh

refresh(refresh_token)

Source code in tigrbl_auth/jwtoken.py

def refresh(self, refresh_token: str) -> Tuple[str, str]:
    payload = self.decode(refresh_token)
    if payload.get("typ") != "refresh":
        raise ValueError("token is not a refresh token")
    base_claims = {
        k: v for k, v in payload.items() if k not in {"iat", "exp", "typ"}
    }
    return self.sign_pair(**base_claims)