Metadata-Version: 2.4
Name: latchvector-sso
Version: 1.0.2
Summary: Official Python SDK for Latch Vector SSO — token verification and authentication.
License-Expression: MIT
Requires-Python: >=3.9
Requires-Dist: httpx>=0.25
Requires-Dist: pyjwt[crypto]>=2.8
Provides-Extra: django
Requires-Dist: django>=4.2; extra == 'django'
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.100; extra == 'fastapi'
Provides-Extra: flask
Requires-Dist: flask>=2.2; extra == 'flask'
Provides-Extra: sqlalchemy
Requires-Dist: sqlalchemy>=2.0; extra == 'sqlalchemy'
Description-Content-Type: text/markdown

# latchvector-sso

Python SDK for Latch Vector SSO. Python 3.9+.

```bash
pip install latchvector-sso
```

Framework integrations live in submodules and pull nothing in unless you
import them:

```bash
pip install "latchvector-sso[fastapi]"   # or [flask], [django]
```

---

## Contents

- [Protecting an API — the common case](#protecting-an-api--the-common-case)
- [What `audience` is for](#what-audience-is-for)
- [The principal](#the-principal)
- [Logging users in](#logging-users-in)
- [Password reset](#password-reset)
- [Device sessions (mobile)](#device-sessions-mobile)
- [Machine-to-machine (API clients)](#machine-to-machine-api-clients)
- [Multitenancy](#multitenancy)
- [Errors](#errors)
- [Configuration](#configuration)
- [Smoke test](#smoke-test)
- [Before you go live](#before-you-go-live)
- [Management API](#management-api)
- [Webhooks](#webhooks)
- [Migrating from your current system](#migrating-from-your-current-system)

## Protecting an API — the common case

Most integrations only need this. Your API verifies tokens locally; it does
not call the SSO service on every request.

```python
from latchvector_sso import TokenVerifier

# Build once at startup — it caches the discovery document and signing keys.
verifier = TokenVerifier(
    issuer="https://sso.yourdomain.com",
    audience="https://api.yourcompany.com",   # your registered identifier
)

principal = verifier.verify_authorization_header(request.headers.get("Authorization"))
```

### FastAPI

```python
from fastapi import Depends, FastAPI
from latchvector_sso import Principal, TokenVerifier
from latchvector_sso.fastapi import SsoAuth

auth = SsoAuth(TokenVerifier(issuer=..., audience=...))
app = FastAPI()

@app.get("/invoices")
def list_invoices(user: Principal = Depends(auth.required)):
    return {"owner": user.uid}

@app.post("/invoices/{invoice_id}/approve")
def approve(invoice_id: int, user: Principal = Depends(auth.requires("invoice.approve"))):
    ...
```

### Flask

```python
from latchvector_sso.flask import SsoAuth, current_principal

auth = SsoAuth(app, TokenVerifier(issuer=..., audience=...))

@app.get("/invoices")
@auth.required
def list_invoices():
    return {"owner": current_principal().uid}

@app.post("/invoices/<int:invoice_id>/approve")
@auth.requires("invoice.approve")
def approve(invoice_id):
    ...
```

### Django

```python
# settings.py
LATCHVECTOR_SSO = {
    "ISSUER": "https://sso.yourdomain.com",
    "AUDIENCE": "https://api.yourcompany.com",
}
MIDDLEWARE = [..., "latchvector_sso.django.SsoAuthenticationMiddleware"]
```

```python
# views.py
from latchvector_sso.django import sso_required, sso_requires

@sso_required
def invoices(request):
    return JsonResponse({"owner": request.principal.uid})

@sso_requires("invoice.approve")
def approve(request, pk):
    ...
```

The middleware attaches `request.principal` when a valid token is present
and leaves it `None` otherwise; the decorators do the rejecting. That keeps
authentication and authorisation separable, so a view can be deliberately
public without having to bypass the middleware.

---

## What `audience` is for

It is your application's registered identifier, and it is **required** —
there is no option to turn the check off.

A token issued for a *different* application is still validly signed by a
trusted issuer. If you check the signature but not the audience, you accept
it, which means you accept one from every user of every application on the
platform. This is the single most common way an SSO integration is
compromised, and it is why the parameter has no default.

## The principal

```python
principal.uid          # 4711 — key your records on this
principal.email        # display only, see below
principal.org_id       # 57
principal.tenant_id    # 1
principal.org_path     # "/1/57/"
principal.permissions  # frozenset({"invoice.approve"})
principal.expires_at   # timezone-aware datetime

principal.has("invoice.approve")
principal.has_any("invoice.approve", "invoice.admin")
principal.has_all("invoice.read", "invoice.approve")
principal.can_reach("/1/57/903/")   # does their granted scope cover this node?
```

**Key your own tables on `uid`, never on the email.** Addresses change, and
a GDPR erasure request scrubs the address while `uid` survives. Rows keyed
on email lose the link to their own user the first time either happens.

**Do not cache `permissions` past `expires_at`.** They are a snapshot from
issue time; a revoked role takes effect on the next token, which is why
access tokens last only 15 minutes.

---

## Logging users in

Only whatever actually handles the password needs this — a login backend, a
BFF, a mobile gateway. Your resource APIs do not.

```python
from latchvector_sso import SsoClient, MfaRequired

sso = SsoClient(
    issuer="https://sso.yourdomain.com",
    audience="https://api.yourcompany.com",
)

result = sso.login(email, password)

if isinstance(result, MfaRequired):
    code = prompt_user_for_code()               # TOTP or recovery code
    tokens = sso.verify_mfa(result.pending_token, code)
else:
    tokens = result
```

`login()` returns a union, not an object with an empty `access_token`. You
have to branch before you can reach a token, so the MFA path cannot be
quietly skipped — a customer will enable MFA eventually, and the failure
mode of the nullable version is an empty token in production.

### Social login

```python
result = sso.social_login("google", google_id_token)
```

The user must already exist. Accounts are provisioned by an administrator;
a first-time social login for an unknown email is refused rather than
silently creating an account.

### Refresh

```python
fresh = sso.refresh(stored_refresh_token)
save_refresh_token(fresh.refresh_token)   # before you use it
```

Refresh tokens rotate: **the old one is dead the moment `refresh()`
returns.** Persist the new one first.

```python
from latchvector_sso import RefreshTokenError, RefreshTokenReusedError

try:
    return sso.refresh(stored)
except RefreshTokenReusedError:
    destroy_session()
    alert_security_team(user_id)      # this is a security event
    raise
except RefreshTokenError:
    return redirect_to_login()        # expired or unknown — ordinary
```

`RefreshTokenReusedError` is deliberately *not* a subclass of
`RefreshTokenError`, so an `except` that only meant "refresh or re-login"
cannot swallow a compromise signal.

### Logout

```python
sso.logout(refresh_token)
```

This revokes the refresh token. The current access token stays valid for
the rest of its 15 minutes — it is a signed bearer token, not a session.
For immediate cut-off, have an administrator disable the account.

---

## Machine-to-machine (API clients)

For a backend job that acts as itself, not a user — the OAuth2
`client_credentials` grant. An admin registers an API client (secret shown
once) bound to an application; the job exchanges the credentials for a
short-lived token.

Each framework has a machine counterpart of the user helpers. They verify with
`verify_client`, so a user access token is rejected there just as a machine
token is rejected by the user helpers — the two never cross.

**FastAPI**

```python
from latchvector_sso import ClientPrincipal
from latchvector_sso.fastapi import SsoAuth

auth = SsoAuth(TokenVerifier(issuer=..., audience=...))

@app.post("/reports/sync")
def sync(client: ClientPrincipal = Depends(auth.requires_scope("reports.write"))):
    return {"org": client.org_id, "client": client.client_id}
```

**Flask**

```python
from latchvector_sso.flask import SsoAuth, current_client

@app.post("/reports/sync")
@auth.requires_scope("reports.write")
def sync():
    return {"org": current_client().org_id}
```

**Django**

```python
from latchvector_sso.django import sso_requires_scope

@sso_requires_scope("reports.write")
def sync(request):
    return JsonResponse({"org": request.client.org_id})
```

**Calling another service** (your app is the job) — obtain and cache a token:

```python
machine = sso.client_credentials(client_id, client_secret, ["reports.write"])
httpx.post(url, headers={"authorization": f"Bearer {machine.access_token}"})
# machine.expires_in_seconds ~ 900; no refresh — cache and re-fetch on expiry.
```

---

## Multitenancy

Verifying a token tells you *who* is calling; multitenancy is about what data
they may touch. The SDK ties your models to the tenant in the verified token, so
a query cannot read or write another tenant's rows even if you forget the
filter. It works on either ORM — Django's, or SQLAlchemy for Flask and FastAPI.

### Django

Inherit `BelongsToTenant` (it adds a `tenant_id` column and the scoping):

```python
from latchvector_sso.django import BelongsToTenant

class Invoice(BelongsToTenant):
    amount = models.IntegerField()
```

Behind `SsoAuthenticationMiddleware`, that is all:

```python
Invoice.objects.all()            # only the caller's tenant
Invoice.objects.create(amount=5) # tenant_id stamped automatically
Invoice.all_tenants.all()        # escape hatch: every tenant, used deliberately
```

Configure it in `settings.py`:

```python
LATCHVECTOR_SSO = {
    "ISSUER": "...",
    "AUDIENCE": "...",
    "TENANT": {
        "ENABLED": True,                      # False in a sandbox / dev
        "COLUMN": "tenant_id",
        "BYPASS_PERMISSIONS": ["PLATFORM_ADMIN"],  # see across tenants
    },
}
```

- **Bypass** — a caller with a `BYPASS_PERMISSIONS` code (a platform operator)
  is unconstrained; an org admin is still bound to their tenant.
- **Sandbox** — set `"ENABLED": False` so dev data and tests aren't confined.
- **Commands & tasks** — with no request there is no tenant, so the scope is
  inert. In a task that must be tenant-bound, set it: `TenantContext.set(id)`.

The `TenantContext` primitive is importable directly
(`from latchvector_sso import TenantContext`) if you integrate a different ORM.

### Flask & FastAPI (SQLAlchemy)

Those frameworks use SQLAlchemy, so the model layer is a mixin plus one call at
startup. The SSO auth (below) sets the tenant context from the verified token;
the rest is identical to Django.

```python
from latchvector_sso.sqlalchemy import BelongsToTenant, install_tenant_scoping

class Invoice(Base, BelongsToTenant):          # gains a tenant_id column
    __tablename__ = "invoices"
    id = mapped_column(Integer, primary_key=True)
    amount = mapped_column(Numeric)

install_tenant_scoping()                       # once, after your models import
```

That is all — reads gain `WHERE tenant_id = <current>` (through joins and
relationship loads too) and inserts are stamped:

```python
session.scalars(select(Invoice)).all()   # only the caller's tenant
session.add(Invoice(amount=5))            # tenant_id stamped on flush
```

The auth integrations set the context automatically once you tell them who may
see across tenants:

```python
# FastAPI
auth = SsoAuth(verifier, bypass_permissions=["PLATFORM_ADMIN"])
# Flask
auth = SsoAuth(app, verifier, bypass_permissions=["PLATFORM_ADMIN"])
```

- **Bypass** — a caller holding a bypass permission is unconstrained; an org
  admin stays bound to their tenant.
- **Async-safe** — the context lives in a `ContextVar`, correct under both
  thread-per-request (Flask) and many concurrent requests on one event loop
  (async FastAPI).
- **Sandbox / other sessions** — `install_tenant_scoping(enabled=False)` turns
  scoping off; pass your own `Session`/`sessionmaker` class as the `session`
  argument if you don't use the default.

### Confining to a sub-tree

`tenant_id` is the hard wall *between* customers. *Within* one customer, an
admin of a sub-org should often see only their slice of the org tree, not the
whole tenant. Opt a model into **subtree mode** and it is narrowed to exactly
the org paths the caller's token grants:

```python
# Django
class Chart(BelongsToTenant):
    tenant_scope_mode = "subtree"          # default is "tenant"
    org_id = models.BigIntegerField(editable=False, null=True)
    org_path = models.TextField(editable=False, null=True)  # index it: see below
    title = models.CharField(max_length=120)

# SQLAlchemy (Flask / FastAPI) — a ready-made mixin with both columns
from latchvector_sso.sqlalchemy import BelongsToTenantSubtree

class Chart(Base, BelongsToTenantSubtree):   # adds org_id + org_path
    __tablename__ = "charts"
    id = mapped_column(Integer, primary_key=True)
    title = mapped_column(String)
```

Alongside `tenant_id`, a subtree model needs an `org_id` and an `org_path` column
(a materialized path like `/1/57/903/`). New rows are stamped with the writer's
own node; reads are confined to:

- **SUBTREE** grants — the caller's node *and everything below it* (a
  left-anchored `org_path__startswith`);
- **SELF** grants — that node *only* (an exact match).

Which applies is decided by the caller's roles at token-issue time and carried
in the `scope_subtree` / `scope_self` claims — you write nothing. A machine
(client-credentials) token has no org reach, so a subtree model falls back to
tenant-wide for it — still leak-safe across customers.

> The trailing slash matters. Paths are stored `/1/57/` (not `/1/57`), so the
> prefix `/1/57/` can never leak into a sibling like `/1/570/`.

### Multitenancy at scale

For tables that will hold billions of rows, three columns and the right indexes
keep every scoped query a range scan, never a table scan:

| Column | Type | Why |
|---|---|---|
| `tenant_id` | `bigint` | the hard customer wall; on every tenant-aware table |
| `org_id` | `bigint` | the owning node — subtree tables only |
| `org_path` | `text` | materialized path `/1/57/903/`, trailing slash — subtree only |

Index **tenant-leading**, so the tenant predicate drives the scan:

```sql
-- every tenant-aware table
CREATE INDEX ON invoices (tenant_id, created_at DESC);

-- subtree tables: prefix scans on org_path within the tenant
CREATE INDEX ON charts (tenant_id, org_path text_pattern_ops);
```

`text_pattern_ops` is what makes `org_path LIKE '/1/57/%'` an index range scan
under any collation. For the largest tenants, partition or shard by `tenant_id`
(Postgres declarative partitioning, or Citus/Nile-style distribution): the
tenant-leading key means a query already touches only its own partition.

---

## Errors

Every error is an `SsoError` with `.code` and `.status`.

| Class | Codes |
|---|---|
| `AuthenticationError` | `invalid_credentials`, `invalid_code`, `invalid_id_token`, `invalid_token`, `invalid_token_use`, `invalid_or_expired_pending_token` |
| `RefreshTokenError` | `invalid_refresh_token`, `refresh_token_expired` |
| `RefreshTokenReusedError` | `refresh_token_reused` |
| `AccountNotActiveError` | `account_not_active` |
| `AccountLockedError` | `account_locked` |
| `AccessDeniedError` | `access_denied` |
| `ValidationError` | `validation_failed` (with `.fields`) |
| `RateLimitError` | `too_many_requests` (with `.retry_after_seconds`) |
| `ConfigurationError` | `unknown_audience`, discovery failures |

`429` is retried automatically with exponential backoff and jitter (twice
by default, `max_rate_limit_retries` to change it). **Nothing else is
retried**, and `error.retryable` is `False` for everything but
`RateLimitError`. A `403` is a decision the service already made; retrying
it produces a stream of `ACCESS_DENIED` audit entries that a compliance
officer will eventually ask you about.

`access_denied` does not distinguish "forbidden" from "does not exist" —
telling them apart would let anyone enumerate records across tenants.

---

## Configuration

You configure **one** URL. The JWKS endpoint is resolved from
`{issuer}/.well-known/openid-configuration` and cached, so the SDK keeps
working if it ever moves.

| Argument | Default | |
|---|---|---|
| `issuer` | — | required |
| `audience` | — | required, cannot be disabled |
| `leeway_seconds` | `30` | skew allowance; keep NTP running regardless |
| `jwks_cache_seconds` | `600` | |
| `timeout_seconds` | `10.0` | |
| `max_rate_limit_retries` | `2` | client only |

`TokenVerifier` is thread-safe and intended to be shared.

---

## Smoke test

```bash
SSO_ISSUER=http://localhost:9000 SSO_AUDIENCE=http://localhost:9000 \
SSO_EMAIL=… SSO_PASSWORD=… python examples/smoke.py
```

Beyond the happy path it asserts that a token minted for a different
audience is rejected and that a tampered signature is rejected — the two
checks whose absence turns a working integration into an open door.

---

## Before you go live

- [ ] `audience` is set to your identifier, not ours
- [ ] Your tables key on `uid`, not email
- [ ] `TokenVerifier` is constructed once, not per request
- [ ] `RefreshTokenReusedError` is handled as a compromise, not retried
- [ ] The `MfaRequired` branch is implemented and tested
- [ ] Tokens are never written to logs, URLs, or error reports

## Password reset

The invite / forgot-password flow (the token comes from the emailed link
or an admin-issued setup link):

```python
sso.forgot_password(email)                # emails a one-time link (no account oracle)
sso.reset_password(token, new_password)   # redeem the link's token
```

## Device sessions (mobile)

A mobile app gets a **longer-lived, device-bound** session by passing a
`device` at login (also on `verifyMfa`/`socialLogin`). The service returns a
stable **`deviceId`** — store it in secure storage and resend it so the same
device is reused. The refresh token lives far longer than the web one and
slides on every use; the user can list and revoke devices via
`GET`/`DELETE /api/users/me/devices` (also on the ManagementClient).

```python
from latchvector_sso import DeviceInfo
# Presence of a device ⇒ a longer-lived, device-bound session.
r = sso.login(email, password, DeviceInfo(name="Ana's iPhone", platform="ios"))
save_to_secure_store(r.device_id)   # store it; resend as DeviceInfo(device_id=…) next launch
```

## Management API

Everything the console does, in code — users, organizations, roles, applications,
API clients, webhooks, audit, bulk import, GDPR. Authenticated with a management
token (log in with `audience` equal to the issuer):

```python
from latchvector_sso import SsoClient, ManagementClient

sso = SsoClient(issuer=issuer, audience=issuer)      # management token
tokens = sso.login(email, password)
mgmt = ManagementClient(issuer, tokens.access_token)
mgmt.users.create(organizationId=org_id, email=email, fullName=name)
mgmt.request("POST", "/api/anything", body={"x": 1})  # every endpoint, incl. new ones
```

**→ [Management API guide](docs/management.md)** — every resource, the token model,
and the generic `request()` escape hatch.

## Webhooks

Get notified the moment a user's access changes — a role assigned or revoked, a
role's permissions changed, an account disabled or erased — so you can clear
caches or force a refresh instead of waiting for the next failed call. Every
delivery is HMAC-signed and timestamped, and this SDK ships a one-call verifier.

**→ [Webhooks guide](docs/webhooks.md)** — events, payload, the signature scheme,
and a verified handler example.

## Migrating from your current system

Bring an existing estate — organizations, users, roles, permissions — across in
one validated pass. Records reference each other by *your own ids*, bcrypt
passwords carry over (everyone else is invited), and re-runs are safe.

**→ [Migration guide](docs/migration.md)** — the two-step validate/commit flow,
the full payload schema, and a worked example.
