masonry-auth-sdk 0.1.2

Python SDK for integrating with Masonry-Auth (OAuth2/OIDC) servers.

masonry-auth-sdk

Python SDK for integrating applications with a Masonry-Auth deployment over OAuth2 / OpenID Connect.

Wraps Authlib with an opinionated surface for the Authorization Code + PKCE flow against the Hydra/Kratos backend that powers Masonry-Auth. Ships both sync and async clients with identical APIs.

Install

pip install masonry-auth-sdk

Requires Python 3.10+.

Quick start

from masonry_auth_sdk import MasonryAuthClient

client = MasonryAuthClient(
    auth_host="https://auth.themasonry.com",
    client_id="my-client-id",
    client_secret="my-client-secret",
    redirect_uri="https://myapp.com/oauth/callback",
)

The async variant is a drop-in replacement with the same constructor:

from masonry_auth_sdk import AsyncMasonryAuthClient

client = AsyncMasonryAuthClient(
    auth_host="https://auth.themasonry.com",
    client_id="my-client-id",
    client_secret="my-client-secret",
    redirect_uri="https://myapp.com/oauth/callback",
)

OIDC discovery and JWKS are fetched lazily and cached in-process, so a single client can safely be kept as a module-level singleton.

State storage

login_url() returns an AuthState object containing the state, nonce, and PKCE code_verifier that were generated for the login attempt. The SDK is storage-agnostic — you are responsible for stashing AuthState in whatever session store your framework uses (signed cookie, Redis, Flask session, FastAPI SessionMiddleware, etc.) and passing the same object back into exchange_code() on the callback request.

Flask (sync) example

from dataclasses import asdict
from flask import Flask, redirect, request, session, url_for
from masonry_auth_sdk import MasonryAuthClient, AuthState

app = Flask(__name__)
app.secret_key = "..."

auth = MasonryAuthClient(
    auth_host="https://auth.themasonry.com",
    client_id="my-client-id",
    client_secret="my-client-secret",
    redirect_uri="https://myapp.com/oauth/callback",
)


@app.route("/login")
def login():
    url, auth_state = auth.login_url()
    session["auth_state"] = asdict(auth_state)
    return redirect(url)


@app.route("/oauth/callback")
def callback():
    stored = session.pop("auth_state", None)
    if not stored:
        return "No pending login", 400

    tokens = auth.exchange_code(
        code=request.args["code"],
        returned_state=request.args["state"],
        auth_state=AuthState(**stored),
    )
    session["access_token"] = tokens.access_token
    session["id_token"] = tokens.id_token
    session["refresh_token"] = tokens.refresh_token

    info = auth.userinfo(tokens.access_token)
    session["user"] = {"sub": info.subject, "email": info.email}
    return redirect(url_for("home"))


@app.route("/logout")
def logout():
    id_token = session.pop("id_token", None)
    session.clear()
    return redirect(
        auth.logout_url(
            id_token_hint=id_token,
            post_logout_redirect_uri=url_for("home", _external=True),
        )
    )


@app.route("/register")
def register():
    return redirect(auth.register_url(return_to=url_for("login", _external=True)))

FastAPI (async) example

from dataclasses import asdict
from fastapi import FastAPI, Request
from fastapi.responses import RedirectResponse
from starlette.middleware.sessions import SessionMiddleware
from masonry_auth_sdk import AsyncMasonryAuthClient, AuthState

app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key="...")

auth = AsyncMasonryAuthClient(
    auth_host="https://auth.themasonry.com",
    client_id="my-client-id",
    client_secret="my-client-secret",
    redirect_uri="https://myapp.com/oauth/callback",
)


@app.get("/login")
async def login(request: Request):
    url, auth_state = await auth.login_url()
    request.session["auth_state"] = asdict(auth_state)
    return RedirectResponse(url)


@app.get("/oauth/callback")
async def callback(request: Request, code: str, state: str):
    stored = request.session.pop("auth_state", None)
    if not stored:
        return {"error": "No pending login"}

    tokens = await auth.exchange_code(
        code=code,
        returned_state=state,
        auth_state=AuthState(**stored),
    )
    request.session["access_token"] = tokens.access_token
    request.session["id_token"] = tokens.id_token

    info = await auth.userinfo(tokens.access_token)
    request.session["user"] = {"sub": info.subject, "email": info.email}
    return RedirectResponse("/")


@app.get("/logout")
async def logout(request: Request):
    id_token = request.session.pop("id_token", None)
    request.session.clear()
    url = await auth.logout_url(
        id_token_hint=id_token,
        post_logout_redirect_uri="https://myapp.com/",
    )
    return RedirectResponse(url)


@app.get("/register")
async def register():
    return RedirectResponse(
        auth.register_url(return_to="https://myapp.com/login")
    )

API summary

Method	Purpose
login_url(*, extra_params=None)	Build /oauth2/auth URL; returns (url, AuthState).
exchange_code(*, code, returned_state, auth_state)	Verify state, exchange code, validate ID token.
refresh(refresh_token)	Exchange a refresh token for new tokens.
userinfo(access_token)	Fetch the OIDC userinfo document.
logout_url(*, id_token_hint, post_logout_redirect_uri, state=None)	Build Hydra RP-initiated logout URL.
register_url(*, return_to=None)	Build URL into the Kratos registration UI.
revoke(token, *, token_type_hint=None)	Revoke a token via RFC 7009 revocation endpoint.
close()	Release the underlying httpx client.

All methods have identical signatures on MasonryAuthClient and AsyncMasonryAuthClient; the async client's methods are coroutines except for register_url, which is purely local.

Exceptions

All errors inherit from MasonryAuthError:

• DiscoveryError — discovery document or JWKS fetch failed.
• StateMismatchError — returned state did not match what was issued.
• TokenExchangeError — token, refresh, or revocation call failed.
• IDTokenError — ID token signature or claim validation failed.
• UserInfoError — userinfo endpoint returned an error.

Development

Tests run in the venv at sdk/env/ (create a new one if this is the first setup) and hit a mocked authorization server (via respx), so no real Hydra is required:

# One-time setup (from repo root)
sdk/env/bin/pip install -e sdk/python[dev]

# Run the tests
sdk/env/bin/pytest sdk/python/tests

Publishing

A convenience script handles building, validation, and upload. Run from the repo root:

./scripts/publish_sdk_python.sh --build-only  # build + validate without uploading
./scripts/publish_sdk_python.sh --test         # upload to TestPyPI
./scripts/publish_sdk_python.sh               # upload to PyPI

Before your first upload, configure a PyPI API token via ~/.pypirc or TWINE_USERNAME / TWINE_PASSWORD environment variables. To test the full flow safely, upload to TestPyPI first:

./scripts/publish_sdk_python.sh --test
pip install --index-url https://test.pypi.org/simple/ masonry-auth-sdk