xtr-security-oauth2-server-core

The OAuth 2.0 authorization-server library: grants, servers, entities, response types and the resource server.

Why?

This is the framework-neutral half of xtr-security-oauth2-server: the authorization server, the token server, the grants, the entities and their interfaces, the response types, the proof-key and redirect-URI validators, the resource server and the protocol errors. An application builds them by hand, so nothing here reaches for a web framework, a container or an ORM — it speaks the HTTP message contracts, and the caller hands it a request and a response.

What it decides is the protocol; what it leaves to whoever builds it is storage. Every store is a Protocol in repositories/ — clients, scopes, access tokens, codes, refresh tokens, device codes, users — with async methods, and every entity a store hands back is an interface in entities/, composed from the traits in entities/traits/. Authorization codes and refresh tokens are sealed (JWE dir + A256GCM), so storage only has to remember whether one was spent; access tokens are signed RFC 9068 tokens, read back by the resource server without a round trip.

The bundle that wires all of this into an application on xtr-dependency-injection — with stores, persistence, controllers, a router, commands and a firewall authenticator — lives in xtr-security-oauth2-server, which depends on this package. To use the server in an application, follow its Use in an application section.

Install

uv add xtr-security-oauth2-server-core

Most applications install xtr-security-oauth2-server instead, which brings this along. Building the servers by hand also needs an event dispatcher and HTTP messages: xtr-event-dispatcher and xtr-http-message are what the Quick start uses.

Quick start

An AuthorizationServer issuing a token to a service by client credentials, and a ResourceServer reading it back — built from constructors, over stores written here in memory.

import json
import secrets
from collections.abc import Sequence
from datetime import timedelta

import anyio
from xtr_clock import SystemClock
from xtr_event_dispatcher import EventDispatcher
from xtr_http_message import Response, ServerRequest
from xtr_security_jwt.services.jws_provider import JoserfcJwsProvider
from xtr_security_jwt.services.key_loader import RawKeyLoader, generate_signing_key, signing_key_id
from xtr_security_oauth2_server_core.authorization_server import AuthorizationServer
from xtr_security_oauth2_server_core.authorization_validators.bearer_token_validator import (
    BearerTokenValidator,
)
from xtr_security_oauth2_server_core.entities.access_token_entity_interface import (
    AccessTokenEntityInterface,
)
from xtr_security_oauth2_server_core.entities.access_token_signer import AccessTokenSigner
from xtr_security_oauth2_server_core.entities.client_entity_interface import ClientEntityInterface
from xtr_security_oauth2_server_core.entities.scope_entity_interface import ScopeEntityInterface
from xtr_security_oauth2_server_core.entities.traits.access_token_trait import AccessTokenTrait
from xtr_security_oauth2_server_core.entities.traits.client_trait import ClientTrait
from xtr_security_oauth2_server_core.entities.traits.entity_trait import EntityTrait
from xtr_security_oauth2_server_core.entities.traits.scope_trait import ScopeTrait
from xtr_security_oauth2_server_core.entities.traits.token_entity_trait import TokenEntityTrait
from xtr_security_oauth2_server_core.grant.client_credentials_grant import ClientCredentialsGrant
from xtr_security_oauth2_server_core.repositories.access_token_repository_interface import (
    AccessTokenRepositoryInterface,
)
from xtr_security_oauth2_server_core.repositories.client_repository_interface import (
    ClientRepositoryInterface,
)
from xtr_security_oauth2_server_core.repositories.scope_repository_interface import (
    ScopeRepositoryInterface,
)
from xtr_security_oauth2_server_core.resource_server import ResourceServer

ISSUER = "https://authorization.example.com"


# Entities: the traits hold the state, the interface comes last.
class Client(EntityTrait, ClientTrait, ClientEntityInterface):
    def __init__(self, identifier: str, name: str) -> None:
        self.set_identifier(identifier)
        self._name = name
        self._redirect_uri = ()
        self._is_confidential = True


class Scope(EntityTrait, ScopeTrait, ScopeEntityInterface):
    def __init__(self, identifier: str) -> None:
        self.set_identifier(identifier)


class AccessToken(EntityTrait, TokenEntityTrait, AccessTokenTrait, AccessTokenEntityInterface):
    pass


# Repositories: where the server reads clients and scopes and keeps tokens.
class Clients(ClientRepositoryInterface):
    def __init__(self) -> None:
        self._clients = {"billing": (Client("billing", "Billing"), "billing-secret")}

    async def get_client_entity(self, client_identifier: str) -> ClientEntityInterface | None:
        found = self._clients.get(client_identifier)
        return found[0] if found is not None else None

    async def validate_client(
        self,
        client_identifier: str,
        client_secret: str | None,
        grant_type: str | None,
    ) -> bool:
        found = self._clients.get(client_identifier)
        # A real store keeps a hash and verifies it; never compare with ==.
        return found is not None and secrets.compare_digest(found[1], client_secret or "")


class Scopes(ScopeRepositoryInterface):
    async def get_scope_entity_by_identifier(self, identifier: str) -> ScopeEntityInterface | None:
        return Scope(identifier) if identifier in {"invoices:read", "invoices:write"} else None

    async def finalize_scopes(
        self,
        scopes: Sequence[ScopeEntityInterface],
        grant_type: str,
        client_entity: ClientEntityInterface,
        user_identifier: str | None = None,
        auth_code_id: str | None = None,
    ) -> Sequence[ScopeEntityInterface]:
        return scopes


class AccessTokens(AccessTokenRepositoryInterface):
    def __init__(self) -> None:
        self._revoked: set[str] = set()

    async def get_new_token(
        self,
        client_entity: ClientEntityInterface,
        scopes: Sequence[ScopeEntityInterface],
        user_identifier: str | None = None,
    ) -> AccessTokenEntityInterface:
        token = AccessToken()
        token.set_client(client_entity)
        for scope in scopes:
            token.add_scope(scope)
        if user_identifier is not None:
            token.set_user_identifier(user_identifier)
        return token

    async def persist_new_access_token(
        self, access_token_entity: AccessTokenEntityInterface
    ) -> None:
        pass  # keep it, so it can be revoked

    async def revoke_access_token(self, token_id: str) -> None:
        self._revoked.add(token_id)

    async def is_access_token_revoked(self, token_id: str) -> bool:
        return token_id in self._revoked


async def main() -> None:
    clock = SystemClock()
    dispatcher = EventDispatcher()
    key_loader = RawKeyLoader(generate_signing_key("RS256").as_pem(private=True).decode(), None)
    jws_provider = JoserfcJwsProvider(key_loader, "RS256", None, 0, clock)
    signer = AccessTokenSigner(
        jws_provider,
        signing_key_id(key_loader, "RS256"),  # the kid the published key set names
        ISSUER,
        clock,
    )
    encryption_key = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8"  # base64url of 32 bytes
    clients, scopes, access_tokens = Clients(), Scopes(), AccessTokens()

    server = AuthorizationServer(
        clients, access_tokens, scopes, signer, encryption_key, "key", dispatcher
    )
    server.enable_grant_type(ClientCredentialsGrant(dispatcher, clock), timedelta(minutes=15))

    # The token endpoint: the caller hands in the request and the response to answer on.
    request = ServerRequest("POST", f"{ISSUER}/token").with_parsed_body(
        {
            "grant_type": "client_credentials",
            "client_id": "billing",
            "client_secret": "billing-secret",
            "scope": "invoices:read",
        },
    )
    answer = await server.respond_to_access_token_request(request, Response())
    issued = json.loads(await answer.get_body().get_contents())
    print(answer.get_status_code(), issued["token_type"], issued["expires_in"])

    # The API: the resource server reads the bearer token back.
    resource_server = ResourceServer(
        access_tokens,
        BearerTokenValidator(access_tokens, jws_provider, ISSUER, clock),
    )
    call = ServerRequest(
        "GET",
        "https://api.example.com/invoices",
        {"Authorization": f"Bearer {issued['access_token']}"},
    )
    read = await resource_server.validate_authenticated_request(call)
    print(read.get_attribute("oauth_client_id"), read.get_attribute("oauth_scopes"))


anyio.run(main)

200 Bearer 899
billing ['invoices:read']

• A refusal is an OAuthServerError: catch it around a server call and answer with await error.generate_http_response(response). Every answer here writes a body, so every generate_http_response is awaited.
• The entities compose traits with the interface last among the bases: an interface listed first would shadow the traits' accessors with its own empty stubs.
• encryption_key_type is "key" (the base64url of 32 random bytes) or "plain" (a password of at least 32 characters, stretched once with PBKDF2 when the server is built).

Grants

Each grant takes the event dispatcher and the clock, plus the stores only it needs; the server hands it the shared ones — client, access-token and scope repositories, the signer, the encryption key — when enable_grant_type(grant, access_token_ttl) enables it.

OAuth2Grants holds the identifiers. The authorization code grant requires a proof key of a public client unless disable_require_code_challenge_for_public_clients() is called, refuses a verifier for a code issued without a challenge, and judges the challenge by the method the code carries. A grant of your own subclasses AbstractGrant (or AbstractAuthorizeGrant for one reached through /authorize); AuthCodeGrant keeps two seams for one that carries something through the code — create_authorization_request() and get_auth_code_payload().

AuthorizationServer answers validate_authorization_request / complete_authorization_request, respond_to_access_token_request, respond_to_device_authorization_request and complete_device_authorization_request(device_code, user_id, user_approved). TokenServer answers respond_to_token_revocation_request (RFC 7009) and respond_to_token_introspection_request (RFC 7662), with the same client authentication.

Response types

AuthorizationServer's last argument is the ResponseTypeInterface it answers token requests with — a BearerTokenResponse when left out. Each request is answered on a shallow copy, so the services a response type holds are shared and per-request state is not.

Events

The grants and servers announce through the dispatcher they were built with, each event under its own name:

from xtr_event_dispatcher import EventDispatcher
from xtr_security_oauth2_server_core.request_access_token_event import RequestAccessTokenEvent
from xtr_security_oauth2_server_core.request_event import RequestEvent


async def record(event: RequestAccessTokenEvent) -> None:
    print("issued", event.get_access_token().get_identifier())


dispatcher = EventDispatcher()
dispatcher.add_listener(RequestEvent.ACCESS_TOKEN_ISSUED, record)

Errors

Everything the library raises derives from OAuth2Error (xtr_security_oauth2_server_core.exception):

A refresh token that is invalid, expired or spent is invalid_grant with status 400, as RFC 6749 §5.2 names it.

Layout

xtr_security_oauth2_server_core/
├── authorization_server.py         AuthorizationServer
├── token_server.py                 TokenServer: revocation and introspection
├── resource_server.py              ResourceServer
├── abstract_handler.py             what grants and token handlers share
├── crypt_trait.py                  sealing codes and refresh tokens
├── oauth2_grants.py                the grant identifiers
├── bearer_challenge.py             the WWW-Authenticate challenge a bearer refusal carries
├── redirect_uri.py                 carrying parameters to where a browser is sent
├── request_event.py, request_access_token_event.py, request_refresh_token_event.py
├── grant/                          the grants and their interface
├── handlers/                       the revocation and introspection handlers
├── entities/                       the entity interfaces, AccessTokenSigner, traits/
├── repositories/                   the store interfaces
├── response_types/                 bearer, device code, redirect and introspection answers
├── request_types/                  AuthorizationRequest
├── authorization_validators/       BearerTokenValidator
├── code_challenge_verifiers/       S256 and plain
├── redirect_uri_validators/        exact matching, loopback ports aside
└── exception/                      OAuth2Error and the errors under it

Development

Developed in the python-xtr monorepo, under packages/xtr-security-oauth2-server-core; run the commands below from there.

uv sync --all-packages --all-extras
uv run ruff check && uv run ruff format --check && uv run basedpyright && uv run ty check && uv run pytest --cov

License

MIT — see LICENSE.