Metadata-Version: 2.4
Name: tynari-auth-sdk
Version: 0.2.0
Summary: JWT verification SDK for Tynari microservices — JWKS fetching, caching, RBAC, and FastAPI middleware
Project-URL: Homepage, https://github.com/tynari/tynari-auth-sdk
Project-URL: Issues, https://github.com/tynari/tynari-auth-sdk/issues
Author-email: Tynari <admin@printeast.com>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Security
Requires-Python: >=3.10
Requires-Dist: httpx>=0.24
Requires-Dist: pydantic>=2.0
Requires-Dist: python-jose[cryptography]>=3.3
Requires-Dist: starlette>=0.27
Provides-Extra: dev
Requires-Dist: fastapi>=0.100; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=7.0; extra == 'dev'
Requires-Dist: respx>=0.20; extra == 'dev'
Description-Content-Type: text/markdown

# 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

```bash
pip install tynari-auth-sdk
```

## Quick Start

```python
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

```python
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:

```python
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:

```python
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:

```python
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`:

```json
{ "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`:

```python
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:

```python
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`:

```python
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:

```python
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:

```python
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.

```python
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

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