Metadata-Version: 2.4
Name: cloudlet-apis
Version: 0.3.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.1; 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, 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. See docs/DESIGN.md - Stream tickets.

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

A `Principal`'s `groups` pairs each **normalized** group name with the SSO
spelling it came from:

```python
{"payments-team": "/ggd-1234-Payments_Team"}
```

The key is what authorization compares; the value is what normalization destroys
- the name a user recognizes, the one an SSO admin can search for, the one
another system may key on.

The idioms that matter read the keys, so the shape costs nothing at the call
site:

```python
"payments-team" in principal.groups     # membership
for group in principal.groups:          # iteration, in the order SSO issued them
list(principal.groups)                  # the flat, normalized list
principal.groups["payments-team"]       # ...and the SSO spelling when you want it
```

Constructing one by hand takes a plain list - `Principal(groups=["payments"])`
becomes `{"payments": "payments"}`, the right reading when nothing was
normalized. **Only the keys are safe to compare against a request:** a value is a
name the identity provider chose and no validator ever saw.

One key, one value, 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 (the key is identical either way), and raising instead would lock
every member of one badly-named group out of the API.

## 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_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="MYAPI_", env_nested_delimiter="__")
    sso: SSOConfig  # MYAPI_SSO__ISSUER, MYAPI_SSO__AUDIENCE, MYAPI_SSO__ADMIN_GROUPS, ...
    admin_api_key: str = ""  # empty disables key auth
    stream_ticket_key: str = ""  # empty disables tickets; >=32 bytes when set
    auth_enabled: bool = True


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

Only `sso.issuer` is required; discovery, JWKS, and the token endpoints are
derived from it. 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`).

```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()

    app = FastAPI(
        title="My API",
        docs_url=None,  # the vendored offline docs replace both routes
        redoc_url=None,
        lifespan=lifespan,
    )

    # 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)  # /docs and /redoc from vendored assets
    wire_sso_login(app, settings.sso)  # Swagger's "Authorize" via SSO + PKCE

    app.include_router(health_router)  # /healthz, /readyz
    app.include_router(things.router, prefix="/api/v1")
    return app
```

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.

### 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=`. Mint behind normal auth; on the stream route, take
the ticket first and fall back to the header, so CLI clients need no ticket:

```python
# myapi/deps.py (continued)
from cloudlet_apis.auth import StreamTickets
from cloudlet_apis.errors import UnauthenticatedError
from fastapi import Query


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


def require_stream_auth(
    request: Request,
    ticket: Annotated[str | None, Query()] = None,
) -> Principal:
    if ticket is not None:
        # A caller that sent a ticket is a browser; no header fallback, or a
        # bad ticket turns into a misleading 401 about a missing header.
        return get_tickets().verify(ticket, request.url.path)
    return require_auth(request)


StreamUser = Annotated[Principal, Depends(require_stream_auth)]
```

```python
# myapi/routers/tickets.py
@router.post("/stream-tickets")
async def create_stream_ticket(body: TicketRequest, user: CurrentUser) -> TicketResponse:
    ticket = get_tickets().mint(user, body.path)  # the path is inside the signature
    return TicketResponse(ticket=ticket.value, expiresAt=ticket.expires_at)
```

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). Which paths are worth minting
for is your call, not this package's.

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