Metadata-Version: 2.4
Name: cloudlet-apis
Version: 0.6.0
Summary: Shared building blocks for the platform's FastAPI services: errors, logging, request ids, web wiring, name rules and SSO auth.
Requires-Python: >=3.12
Description-Content-Type: text/markdown
Requires-Dist: pydantic>=2.7
Provides-Extra: web
Requires-Dist: fastapi>=0.110; extra == "web"
Provides-Extra: auth
Requires-Dist: fastapi>=0.110; extra == "auth"
Requires-Dist: httpx>=0.27; extra == "auth"
Requires-Dist: pyjwt[crypto]>=2.8; extra == "auth"
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-cov>=5.0; extra == "dev"
Requires-Dist: ruff==0.16.2; extra == "dev"
Requires-Dist: pip-audit>=2.7; extra == "dev"
Requires-Dist: build>=1.0; extra == "dev"
Requires-Dist: twine>=5.0; extra == "dev"

# cloudlet-apis

Building blocks every API on this platform repeats, published once: the error
envelope, the log format, request correlation, the shared FastAPI wiring, the
name/group rules, and SSO auth.

Extracted from `serverless`, which is its first consumer.

## Install

```
pip install "cloudlet-apis[web,auth]"
```

Python 3.12+. The extras mirror the layering, so you take only the layers you
need:

| install | you get | it costs you |
| --- | --- | --- |
| `cloudlet-apis` | `errors`, `names`, `logging`, `requestid` | pydantic 2.7+ |
| `cloudlet-apis[web]` | `+ web` | FastAPI 0.110+ |
| `cloudlet-apis[auth]` | `+ auth` | FastAPI 0.110+, httpx 0.27+, pyjwt 2.8+ |

Those floors are low on purpose, and they are **tested, not assumed**: a bound
here is a bound on every service that installs this, and nothing in the package
needs a recent release of anything. CI's `floor` job pins every `>=` to exactly
its lower bound, on the oldest supported interpreter, and runs the whole suite
there - because `pip install .` always resolves to the *newest* version
satisfying a bound, so an ordinary test run never touches the versions
pyproject claims to support. The `test` job covers every Python in the declared
range, not only its ends.

A service that serves no HTTP - a controller, a worker - installs it bare and
still gets the error types and the log format without acquiring a web stack.
`tests/test_layering.py` enforces that, and CI re-checks it against the built
wheel in a venv where the extras genuinely are not installed.

## What is here

**`errors`** - `APIError` and its subclasses, as plain data (`status_code`,
`code`). No web framework, so they can be raised anywhere. Add your service's own
failures by subclassing `APIError` in your own tree; `error_catalog()` walks
subclasses at call time, so a locally-defined error is published without editing
this package.

**`names`** - `normalize_group`, `validate_name`, `validate_group`, and the
`Name` / `Group` annotated types that put them in a request model. Group
normalization is an authorization concern, not a formatting one: it is applied to
both a token's groups and a request-supplied one, so a membership check compares
canonical forms. Two APIs normalizing differently is a security bug, which is why
the rule is published rather than copied.

**`logging`** / **`requestid`** - one line format carrying an `X-Request-ID`
correlation id, adopted from the ingress when present and minted otherwise, and
readable from `request.state`, the response header, and every log record.

**`web`** - `/healthz` and `/readyz`, Swagger/ReDoc served from vendored assets
(no CDN, for airgap), and the exception handlers rendering everything - domain
errors, request validation, framework HTTP errors, and anything unanticipated -
as one response envelope.

**`auth`** - SSO OIDC token validation (discovery, JWKS cache, JWT verification),
a static admin-key fallback, claims-to-`Principal` mapping, the FastAPI
dependency that wires them together, the Swagger "Authorize" wiring (with a
token proxy for deployments whose SSO forbids public clients), and
`StreamTickets`.

`StreamTickets` is for the one client that cannot authenticate normally: a
browser consuming Server-Sent Events. `EventSource` sends no `Authorization`
header and there is no API to give it one, so the app mints a short-lived signed
ticket over an ordinary POST and the browser opens the stream with `?ticket=`.
It is worth almost nothing on purpose - one path, about a minute, an identity
the caller already had - and it is signed rather than stored, so any replica can
verify one any replica minted. Which paths are worth minting for is the app's
call, not this package's. The HTTP around the signer ships here too:
`ticket_mint_router` (the POST that spends a header on a ticket) and
`stream_auth` (the dependency a stream route wraps in `Depends`) - see
"Stream tickets" under Wiring an API, and docs/DESIGN.md - Stream tickets.

```python
tickets = StreamTickets(settings.stream_ticket_key)  # empty disables minting
ticket = tickets.mint(principal, "/api/myapi/v1/.../logs/pods/some-pod")
principal = tickets.verify(ticket.value, request.url.path)
```

A `Principal`'s `groups` is a tuple of `SSOGroup`, each carrying both spellings of
one group:

```python
SSOGroup(name="payments-team", sso_name="/ggd-1234-Payments_Team")
```

`name` is normalized and is what authorization compares; `sso_name` is what
normalization destroyed - the name a user recognizes, the one an SSO admin can
search for, the one another system may key on. One object rather than two
parallel collections, because the two spellings are one fact and cannot then
drift.

```python
principal.can_access_group("payments-team")  # membership - use this
principal.group_names  # ("payments-team", ...) normalized
principal.sso_group_names  # ("/ggd-1234-Payments_Team", ...)

for group in principal.groups:  # in the order SSO issued them
    group.name, group.sso_name, str(group)  # str(group) is the normalized name
```

> **`"payments-team" in principal.groups` is always False** - it compares a
> string against `SSOGroup` objects. Use `can_access_group`. It fails closed (a
> stale check denies rather than grants), but it is silent, so it is worth
> knowing about.

Constructing one by hand takes a plain list of SSO names -
`Principal(groups=["payments"])` becomes `SSOGroup(name="payments",
sso_name="payments")`, the right reading when nothing was normalized.
`groups_from_sso_names(...)` is the same pairing on its own, and
`SSOGroup.from_sso_name(...)` does one. **Only `name` is safe to compare against a
request:** `sso_name` is a name the identity provider chose and no validator
ever saw.

One normalized name to one SSO spelling, because our SSO groups are spelled so
that no two normalize to the same name. That is a property of the directory
rather than of `normalize_group`, which is many-to-one in principle - so a clash
is survived, not trusted away: the first spelling wins and a warning is logged.
Authorization is unaffected (`name` is identical either way), and raising instead
would lock every member of one badly-named group out of the API.

> Not to be confused with `cloudlet_apis.names.Group`, which is the annotated
> **string** type for a group in a request model. That one is the wire form a
> caller sends; `cloudlet_apis.auth.SSOGroup` is a group the token resolved to.

## Wiring an API

The whole assembly, in the order the pieces go together: settings, the auth
component, the app, error handling, and (if you stream to browsers) tickets.
The `serverless` API is the canonical consumer if you want a full working
example next to this outline.

### 1. Settings

Configuration stays in your repository. This package never reads the
environment: you own your `BaseSettings` - its env prefix, its `.env`, its
fields - and hand the auth component the resolved values.

```python
# myapi/core/config.py
from functools import lru_cache

from cloudlet_apis.auth import SSOConfig
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="MYAPI_", env_nested_delimiter="__")
    # MYAPI_SSO__ISSUER, MYAPI_SSO__AUDIENCE, MYAPI_SSO__SWAGGER_CLIENT_SECRET, ...
    # Every SSOConfig field has a default, so no subclass is needed to declare one -
    # but SET MYAPI_SSO__ISSUER: its default is the platform's realm, not yours.
    sso: SSOConfig = Field(default_factory=SSOConfig)
    admin_api_key: str = ""  # empty disables key auth
    stream_ticket_key: str = ""  # empty disables tickets; >=32 bytes when set
    auth_enabled: bool = True
    # Where this API is served, when it shares a host with others (step 3).
    # Empty is the root mount. Normalize it once - everything is concatenated
    # onto it, so "/api/myapi/" and "/api/myapi" would give two sets of URLs.
    base_path: str = ""


@lru_cache
def get_settings() -> Settings:
    return Settings()
```

Discovery, JWKS, and the token endpoints are all derived from `sso.issuer`, so
it is the one value that matters - **and its default is the platform's realm,
not yours.** Nothing fails to start if you leave it: the service simply trusts
that realm, so set `MYAPI_SSO__ISSUER` in every deployment
(docs/DESIGN.md - The issuer's default is a liability).

An empty `sso.audience` (the default) skips the `aud` check, so tokens work
without a Keycloak audience mapper - set it when you want tokens scoped to this
service only.

### 2. The auth component

One `SSOAuth` per app, holding that app's JWKS cache. Build it behind an
`lru_cache` getter (so settings resolve once, not per request) and keep
`require_auth` a module-level function of your own - that function is the key
`dependency_overrides` uses, so tests override *your* callable without knowing
this package exists.

```python
# myapi/deps.py  -- bind CurrentUser at MODULE level, see the note below
from functools import lru_cache
from typing import Annotated

from cloudlet_apis.auth import Principal, SSOAuth
from fastapi import Depends, Request

from myapi.core.config import get_settings


@lru_cache
def get_auth() -> SSOAuth:
    settings = get_settings()
    return SSOAuth(
        settings.sso,
        admin_api_key=settings.admin_api_key,
        enabled=settings.auth_enabled,
    )


def require_auth(request: Request) -> Principal:
    return get_auth().require_auth(request)


CurrentUser = Annotated[Principal, Depends(require_auth)]
```

> **Bind `CurrentUser` at module level.** Under `from __future__ import
> annotations` a route's annotations are strings FastAPI resolves against module
> globals, so a name local to a factory function is one it cannot find - the
> route then reads as an unannotated parameter and answers 400 instead of
> authenticating.

`require_auth` accepts two credentials on one `Authorization: Bearer` header: a
structural JWT is validated against SSO, anything else is matched against
`admin_api_key` in constant time. With `enabled=False` every request resolves
to a dev principal and nothing is validated - local development only.

### 3. The app

Order matters twice here: middleware added *last* wraps everything added
before it, and `mount_offline_docs` must find no framework `/docs` route in its
way (it enforces `docs_url=None`).

If your API shares a host with the platform's others, it is served under a base
path of its own - `/api/myapi`, say - and **everything** goes beneath it: the
routes, the docs and their assets, the OpenAPI document, the token proxy. Pass
it to the two helpers that register routes for you, and put it on
`openapi_url` and your own `include_router` calls. Leave it empty and you get
the root mount, which is what a local run wants, so it is one setting rather
than a code path per environment.

```python
# myapi/main.py
import asyncio
from contextlib import asynccontextmanager

from cloudlet_apis.auth import wire_sso_login
from cloudlet_apis.logging import configure_logging, get_logger
from cloudlet_apis.requestid import RequestIDMiddleware
from cloudlet_apis.web import health_router, mount_offline_docs, register_exception_handlers
from fastapi import FastAPI

from myapi.core.config import get_settings
from myapi.deps import get_auth
from myapi.routers import things

logger = get_logger(__name__)


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Resolve OIDC discovery now, off the event loop, so the first request
    # doesn't pay for it. Best-effort by design: a down IdP is retried lazily
    # on first use, so startup never depends on it - log, don't raise.
    if get_settings().auth_enabled:
        try:
            await asyncio.wait_for(asyncio.to_thread(get_auth().warmup), timeout=5.0)
        except Exception as exc:
            logger.warning("SSO warmup failed (retried on first use): %s", exc)
    yield


def create_app() -> FastAPI:
    configure_logging()
    settings = get_settings()
    base_path = settings.base_path  # e.g. "/api/myapi"; empty serves at the root

    app = FastAPI(
        title="My API",
        docs_url=None,  # the vendored offline docs replace both routes
        redoc_url=None,
        lifespan=lifespan,
        openapi_url=f"{base_path}/openapi.json",
    )

    # Last added = outermost. The request id must be minted before anything
    # else runs, so every log line and every error body can carry it.
    app.add_middleware(RequestIDMiddleware)

    register_exception_handlers(app)  # every error becomes the one envelope
    mount_offline_docs(app, base_path=base_path)  # its /docs, /redoc and assets
    # Swagger's "Authorize" via SSO + PKCE. With sso.swagger_client_secret set,
    # the token exchange is proxied so the client can be confidential (see below).
    wire_sso_login(app, settings.sso, base_path=base_path)

    app.include_router(health_router)  # /healthz, /readyz - NOT under base_path
    app.include_router(things.router, prefix=f"{base_path}/v1")
    return app
```

The probes stay off the base path deliberately: the kubelet reaches the pod
directly, so they never travel through whatever serves the API to everyone else.

Two things follow from serving this way, and both bite in production rather than
in tests. **Whatever fronts the app must forward the path whole** - no HAProxy
`rewrite-target`, no equivalent - because the routes are registered at the paths
clients call and a router that strips the leading segments leaves nothing that
matches. And **the Swagger redirect URI moves with the base path**, to
`https://{host}{base_path}/docs/oauth2-redirect`, so the Keycloak client has to list
that exact value or "Authorize" fails with `invalid_redirect_uri` after the user
has already logged in.

#### `base_path` or `root_path`, and why not both

FastAPI already has a way to say where an app is served: `root_path`. The two
describe the same address and disagree about where the prefix lives.

| | `root_path="/api/myapi"` | `base_path="/api/myapi"` |
|---|---|---|
| Routes registered at | `/v1/ping` | `/api/myapi/v1/ping` |
| `/api/myapi/v1/ping` | 200 | 200 |
| `/v1/ping` | **200** - a second address | 404 |
| `request.url.path` | whichever the caller used | always the complete one |
| OpenAPI | `paths: /v1/...`, `servers: /api/myapi` | `paths: /api/myapi/v1/...` |

`root_path` is built for the classic reverse proxy: it strips the prefix, tells
the app about it, and the app writes URLs pointing back at itself. `base_path`
puts the prefix into the registrations, so the complete path is the only path.

Prefer `base_path`, for two reasons that only show up in a deployment:

- **The short path stays live under `root_path`.** Two addresses for one
  endpoint means `request.url.path` can be either string - and anything that
  signs or compares a path (a stream ticket, below) then has two answers for
  the same request.
- **A `Mount` does not follow the same rule as a route.** The docs assets are
  one, and it answers on the full path only. So even under `root_path` - the
  arrangement whose whole premise is that something in front strips the prefix -
  the assets 404 if anything actually strips it. Either way the path has to
  arrive whole, which is the one thing `root_path` was supposed to spare you.

**Pick one per app; they do not compose.** With `root_path` set, Starlette
strips it before matching, so a route registered at `/api/myapi/docs` is looked
up as `/docs` and never found. Serving the API through `root_path` while
registering the docs through `base_path` leaves the docs answering at neither
address - 404 on `/api/myapi/docs` *and* on `/docs`. If `root_path` is set,
every route must be registered short, including the ones these helpers add
(pass no `base_path` and they read `root_path` instead).

Don't call `warmup()` from module scope or `SSOAuth.__init__`-time code: the
component is built at import, and a blocking discovery round trip at import
time would slow (or, IdP down, break) startup in every worker. The lifespan is
the right place - on the event loop's clock, bounded, and non-fatal.

#### When your SSO team forbids public clients

Swagger UI logs in with Authorization Code + PKCE, which needs no secret - and
that is exactly what makes its Keycloak client a **public** one. A policy against
public clients therefore blocks the "Authorize" button, not the API: this package
only *validates* tokens, so a service registers no OAuth client to serve traffic.

Set `sso.swagger_client_secret` and the client can be confidential instead. The browser
still runs the authorization leg against Keycloak with PKCE, but posts the code
to `POST {base_path}/auth/token` on your app, which adds the secret and completes
the exchange server-side:

```
MYAPI_SSO__SWAGGER_CLIENT_SECRET=<from your secret store>
```

Nothing else changes - the route is mounted, the published `tokenUrl` points at
it, and both happen in that one call so they cannot disagree. Leave the secret
empty (local dev) and you get today's public-client flow, so the switch is
configuration rather than a code path per environment.

On the Keycloak client: enable Standard Flow, register the Swagger redirect URI
(the one carrying your base path, above),
set *Proof Key for Code Exchange Code Challenge Method* to `S256` so PKCE is
required rather than merely offered, and **disable Service Accounts and Direct
Access Grants** - the proxy refuses every grant but `authorization_code` and
`refresh_token`, and that setting is the same lock on Keycloak's side.

Worth being straight with your SSO team about what this does and does not do: it
keeps the *client secret* server-side, which is what the public-client rule is
about. The user's own tokens still reach the browser, because Swagger UI has to
call your API with them. It is a secret-hiding proxy, not a BFF holding tokens in
a server-side session.

### 4. Routes and errors

```python
# myapi/routers/things.py
from cloudlet_apis.errors import APIError, ForbiddenError, NotFoundError
from cloudlet_apis.names import Group
from fastapi import APIRouter

from myapi.deps import CurrentUser

router = APIRouter()


class ThingFrozenError(APIError):  # your own failures: subclass, done
    status_code = 409
    code = "THING_FROZEN"


@router.get("/groups/{group}/things")
async def list_things(group: Group, user: CurrentUser) -> list[Thing]:
    if not user.can_access_group(group):  # both sides already normalized
        raise ForbiddenError(f"not a member of {group}")
    ...
```

Raise `APIError` subclasses from anywhere - routes, services, background code
that surfaces through a route - and the handlers render them all as one
envelope, request id included:

```json
{"error": {"status": 403, "code": "FORBIDDEN", "message": "not a member of payments",
           "details": [], "requestId": "d0a1..."}}
```

Locally-defined subclasses are picked up by `error_catalog()` too, so an
`/info`-style endpoint can publish your service's error codes without this
package knowing about them.

### 5. Stream tickets (only if you serve SSE to browsers)

A browser's `EventSource` cannot send an `Authorization` header, so the app
mints a short-lived signed ticket over an ordinary POST and the browser opens
the stream with `?ticket=`. Both ends ship here as factories - you supply the
signer, your auth, your path allowlist, and your API's timezone:

```python
# myapi/deps.py (continued)
from cloudlet_apis.auth import StreamTickets, stream_auth
from fastapi import Depends


@lru_cache
def get_tickets() -> StreamTickets:
    return StreamTickets(get_settings().stream_ticket_key)


# Ticket first, no header fallback on a bad one: a caller that sent a ticket
# is a browser, and a 401 about a missing header would mislead whoever debugs
# it. A caller with no ticket still authenticates off the header, so CLI
# clients need none.
require_stream_auth = stream_auth(
    get_tickets, optional_auth, mint_path_hint="/api/myapi/v1/stream-tickets"
)
StreamUser = Annotated[Principal, Depends(require_stream_auth)]
```

```python
# myapi/main.py (with the other include_router calls)
from cloudlet_apis.auth import ticket_mint_router


def valid_stream_path(path: str) -> str:
    """Your allowlist: raise ValidationError unless `path` names one of YOUR
    streaming endpoints. A ticket is a bearer credential in a URL - what it
    can open is your call, not this package's."""
    ...


app.include_router(
    ticket_mint_router(get_tickets, require_auth, valid_stream_path, expiry_tz=MY_TZ),
    prefix=api_base(settings),  # the mint lands at {base}/stream-tickets
)
```

`stream_auth` wants your *optional* header auth (returns `None` when no
credential was sent, still raises on a bad one); `ticket_mint_router` wants the
*required* one - minting spends the caller's real credential. `expiry_tz` is
whatever timezone your API renders every other timestamp in, so `expiresAt`
matches them.

The key must be the same value in every replica and site - a ticket is
verified by whichever process the stream lands on - and at least 32 bytes
(`StreamTickets` refuses less; a ticket asserts an identity, admin flag
included, so a guessable key must not start). Enumerate the allowlist anchored
at your base path, so mint and `verify` compare the same complete string that
`request.url.path` holds.

## Releasing

`ci.yml` validates every PR and branch push: lint, tests on every Python in the
supported range, the `floor` job described above, gitleaks + pip-audit, and a
package build that verifies the wheel carries the vendored assets and that a bare
install really has no web or auth dependencies.

`release.yml` is a manual `workflow_dispatch`. Enter `X.Y.Z` (or
`X.Y.Z-{alpha|beta|rc}.N`) and it stamps the version onto a `vX.Y.Z` tag - the
branch is left unchanged - runs the same checks, then builds, cosign-signs and
publishes the sdist and wheel, and cuts a GitHub Release carrying both plus their
signature bundles.

It publishes to **PyPI with trusted publishing** - there is no stored token,
here or in repository secrets. GitHub mints a short-lived OIDC token for the run,
PyPI checks its claims against the publisher registered for the project, and
returns an API token that expires in minutes.

PyPI matches on three things, and all three must agree with the trusted
publisher configured at
[pypi.org](https://pypi.org/manage/account/publishing/):

| claim | value |
| --- | --- |
| Repository | `black-cloudlet/cloudlet-apis` |
| Workflow | `release.yml` - the file name, **no leading dot** |
| Environment | `pypi_token` - the `environment:` on the release jobs |

A mismatch means PyPI refuses to mint. The `prepare` job therefore *rehearses*
the exchange and throws the token away, **before** the tag is pushed, so a
misconfiguration fails without leaving a stray tag behind. It also prints the
claims the run presented, so a refusal can be read against the publisher config
rather than guessed at.

The `pypi_token` environment has to exist in the repository
(**Settings -> Environments**). Referencing it is what puts the `environment`
claim in the token; it is also where to add **required reviewers** if a release
should wait for a human.

The exchange is two `curl` calls rather than `pypa/gh-action-pypi-publish`,
because every action here is SHA-pinned and that one could not be pinned when
this was written. Swapping it in later is a small change.

Two things to know before the first release. A PyPI version is **permanent**: it
can be yanked but never replaced, and the project name is claimed globally by the
first upload. And an airgapped cluster cannot reach pypi.org, so installing this
there still goes through a pull-through mirror.

Consumers should pin a compatible range (`cloudlet-apis[web,auth]>=1.2,<2`) so a
patch here does not need a coordinated bump in every API.

## Design notes

The reasoning behind the decisions that are not obvious from the code - the
layering, why auth is an object, the `Principal.groups` shape and its collision
rule, why the annotated types document without constraining - is in
[docs/DESIGN.md](docs/DESIGN.md). The docstrings stay short and point there.

## Development

```
pip install -e ".[web,auth,dev]"
ruff check . && ruff format --check .
pytest -q --cov=cloudlet_apis
```

That installs the newest of everything. To reproduce the `floor` job - what a
consumer with an older index gets - pin each `>=` to its lower bound:

```
python -c "
import tomllib
from packaging.requirements import Requirement
p = tomllib.load(open('pyproject.toml','rb'))['project']
for raw in p['dependencies'] + p['optional-dependencies']['web'] + p['optional-dependencies']['auth']:
    r = Requirement(raw)
    if lo := [s.version for s in r.specifier if s.operator == '>=']:
        e = f\"[{','.join(sorted(r.extras))}]\" if r.extras else ''
        print(f'{r.name}{e}=={min(lo)}')
" | sort -u | pip install -r /dev/stdin
pip install --no-deps -e .
```
