tynari-auth-sdk 0.1.3

JWT verification SDK for Tynari microservices — JWKS fetching, caching, RBAC, and FastAPI middleware

tynari-auth-sdk

JWT verification SDK for Tynari microservices. Drop-in middleware that handles JWKS fetching, public key caching, automatic key rotation, RS256 token verification, and role-based access control.

Your auth service signs tokens with a private key. Every other microservice uses this SDK to verify those tokens with the public key — fetched automatically from the JWKS endpoint. No shared secrets, no key files to distribute.

Installation

pip install tynari-auth-sdk

Quick Start

from fastapi import FastAPI, Depends
from auth_sdk import AuthMiddleware, get_current_user, require_roles, TokenUser

app = FastAPI()

app.add_middleware(
    AuthMiddleware,
    jwks_url="https://auth.tynari.com/.well-known/jwks.json",
)

@app.get("/me")
async def me(user: TokenUser = Depends(get_current_user)):
    return {"user_id": user.sub, "role": user.role}

That's it. The middleware fetches the public key, caches it, verifies every incoming Authorization: Bearer <token> header, and attaches the decoded user to the request.

---

Middleware Configuration

AuthMiddleware

app.add_middleware(
    AuthMiddleware,
    jwks_url="https://auth.tynari.com/.well-known/jwks.json",
    cache_ttl=3600.0,
    exclude_paths={"/health", "/docs", "/openapi.json"},
    exclude_prefixes={"/public/", "/webhooks/"},
    auto_error=True,
)

Parameter	Type	Default	Description
jwks_url	str	required	Full URL to the auth service JWKS endpoint
cache_ttl	float	3600.0	How long (in seconds) to cache the public key before re-fetching
exclude_paths	set[str]	set()	Exact paths that skip authentication entirely
exclude_prefixes	set[str]	set()	Path prefixes that skip authentication (e.g. "/public/" matches /public/anything)
auto_error	bool	True	If True, returns 401 on missing/invalid token. If False, sets request.state.user = None and continues

Path Exclusion

Use exclude_paths for exact matches and exclude_prefixes for wildcard-style prefix matching:

app.add_middleware(
    AuthMiddleware,
    jwks_url="https://auth.tynari.com/.well-known/jwks.json",
    exclude_paths={"/", "/health", "/docs", "/openapi.json", "/redoc"},
    exclude_prefixes={"/stream/", "/library/", "/public/"},
)

With this config:

• /health — skipped (exact match)
• /stream/abc-123 — skipped (prefix match)
• /library/fonts/search — skipped (prefix match)
• /api/orders — authenticated

---

Accessing the Current User

After the middleware runs, the verified user is available on request.state.user. Use the get_current_user dependency to access it cleanly:

from auth_sdk import get_current_user, TokenUser

@app.get("/profile")
async def profile(user: TokenUser = Depends(get_current_user)):
    return {
        "user_id": user.sub,
        "role": user.role,
    }

If no valid token was provided, get_current_user raises a 401 Unauthorized.

TokenUser Fields

Field	Type	Description
sub	str	User ID (UUID string from the auth service)
role	str	User role (e.g. "artist", "seller", "individual_buyer")
scope	str	Token scope — always "access" after verification
exp	int	Token expiration timestamp (unix epoch)

---

Role-Based Access Control (RBAC)

Use require_roles to restrict endpoints to specific roles. The microservice decides which roles are allowed — the SDK just enforces it:

from auth_sdk import require_roles

# Single role
@app.get("/studio", dependencies=[Depends(require_roles(["artist"]))])
async def artist_studio():
    return {"message": "Welcome to your studio"}

# Multiple roles
@app.get("/orders", dependencies=[Depends(require_roles(["seller", "individual_buyer"]))])
async def list_orders():
    return {"orders": []}

# Get the user object AND enforce roles
@app.get("/dashboard")
async def dashboard(user: TokenUser = Depends(require_roles(["artist", "seller"]))):
    return {"user_id": user.sub, "role": user.role}

If the token's role is not in the allowed list, the SDK returns 403 Forbidden:

{ "detail": "Insufficient permissions" }

---

Optional Authentication

Some endpoints need to work for both authenticated and unauthenticated users (e.g. a product page that shows extra info for logged-in users). Set auto_error=False:

app.add_middleware(
    AuthMiddleware,
    jwks_url="https://auth.tynari.com/.well-known/jwks.json",
    auto_error=False,
)

@app.get("/product/{id}")
async def product(id: str, request: Request):
    user = request.state.user  # TokenUser or None

    product = get_product(id)
    if user:
        product["is_favorited"] = check_favorite(user.sub, id)

    return product

When auto_error=False, the middleware sets request.state.user = None instead of returning 401 for missing or invalid tokens. Note that get_current_user will still raise 401 if used on these endpoints — check request.state.user directly instead.

---

JWKS Client (Advanced)

For non-FastAPI applications or custom verification flows, use JWKSClient and verify_access_token directly:

from auth_sdk import JWKSClient, verify_access_token

client = JWKSClient(
    jwks_url="https://auth.tynari.com/.well-known/jwks.json",
    cache_ttl=3600.0,
)

async def verify(token: str):
    jwk = await client.get_key()
    user = verify_access_token(token, jwk)
    print(user.sub, user.role)

# Clean up when shutting down
await client.close()

Key Caching and Rotation

The JWKSClient handles key lifecycle automatically:

1. First request — fetches the JWKS from the auth service and caches it
2. Subsequent requests — serves from cache (no network call)
3. Cache expiry — re-fetches after cache_ttl seconds (default: 1 hour)
4. Unknown kid — if a token has a kid not in the cache, triggers an immediate re-fetch (handles key rotation)
5. Thundering herd protection — concurrent cache misses are coalesced into a single fetch via asyncio.Lock

---

Error Handling

The SDK defines three exception types, all inheriting from AuthSDKError:

from auth_sdk import AuthSDKError, JWKSFetchError, TokenVerificationError

Exception	When
JWKSFetchError	JWKS endpoint is unreachable, returns invalid data, or has no keys
TokenVerificationError	JWT signature is invalid, token is expired, scope is not "access", or required claims are missing
AuthSDKError	Base class — catch this to handle any SDK error

When using the middleware, these are caught automatically and returned as 401 responses. When using JWKSClient / verify_access_token directly, handle them yourself:

from auth_sdk import JWKSClient, verify_access_token, JWKSFetchError, TokenVerificationError

async def check_token(token: str):
    try:
        jwk = await client.get_key()
        user = verify_access_token(token, jwk)
        return {"valid": True, "user_id": user.sub}
    except JWKSFetchError:
        return {"valid": False, "reason": "Could not reach auth service"}
    except TokenVerificationError as e:
        return {"valid": False, "reason": str(e)}

---

Full Example — Microservice Setup

A complete FastAPI microservice using the SDK:

from fastapi import FastAPI, Depends
from auth_sdk import AuthMiddleware, get_current_user, require_roles, TokenUser

app = FastAPI(title="Orders Service")

# Auth middleware — all routes require a valid token unless excluded
app.add_middleware(
    AuthMiddleware,
    jwks_url="https://auth.tynari.com/.well-known/jwks.json",
    cache_ttl=3600.0,
    exclude_paths={"/health"},
    exclude_prefixes={"/webhooks/"},
)


@app.get("/health")
async def health():
    return {"status": "ok"}


@app.get("/me")
async def me(user: TokenUser = Depends(get_current_user)):
    return {"user_id": user.sub, "role": user.role}


@app.get("/orders")
async def list_orders(user: TokenUser = Depends(require_roles(["seller", "individual_buyer"]))):
    # user.sub is the authenticated user's UUID
    orders = await fetch_orders_for_user(user.sub)
    return {"orders": orders}


@app.post("/orders")
async def create_order(user: TokenUser = Depends(require_roles(["individual_buyer"]))):
    order = await create_order_for_user(user.sub)
    return {"order_id": order.id}


@app.get("/admin/stats", dependencies=[Depends(require_roles(["seller"]))])
async def admin_stats():
    return {"total_orders": 42}

---

How It Works

Client Request
    │
    ▼
┌─────────────────────────────────────────────────┐
│  AuthMiddleware                                  │
│                                                  │
│  1. Extract Authorization: Bearer <token>        │
│  2. Read `kid` from JWT header                   │
│  3. Fetch public key from JWKS (cached)          │
│  4. Verify RS256 signature + expiration          │
│  5. Enforce scope == "access"                    │
│  6. Attach TokenUser to request.state.user       │
└─────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────┐
│  Route Handler                                   │
│                                                  │
│  get_current_user()  → TokenUser                 │
│  require_roles(...)  → TokenUser (or 403)        │
└─────────────────────────────────────────────────┘

---

API Reference

AuthMiddleware

Starlette middleware that verifies JWTs on every request.

AuthMiddleware(
    app: ASGIApp,
    jwks_url: str,
    cache_ttl: float = 3600.0,
    exclude_paths: set[str] | None = None,
    exclude_prefixes: set[str] | None = None,
    auto_error: bool = True,
)

get_current_user(request: Request) -> TokenUser

FastAPI dependency. Returns the authenticated user or raises 401.

require_roles(allowed_roles: list[str]) -> Depends

FastAPI dependency factory. Returns the authenticated user if their role is in allowed_roles, otherwise raises 403.

verify_access_token(token: str, jwk_dict: dict) -> TokenUser

Low-level function. Verifies an RS256 JWT against a JWK dict. Raises TokenVerificationError on failure.

JWKSClient(jwks_url: str, cache_ttl: float = 3600.0)

Async JWKS fetcher with caching.

• await client.get_key(kid=None) — returns the JWK dict for the given kid
• await client.close() — closes the underlying HTTP client

TokenUser

Pydantic model with fields: sub, role, scope, exp.

---

Publishing

pip install build twine
python -m build
twine upload dist/*