Metadata-Version: 2.4
Name: cloudlet-apis
Version: 0.4.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, 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. 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` 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


@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`).

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

    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.

#### 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 /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,
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=`. 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 .
```
