Metadata-Version: 2.5
Name: educate-custodian-consumer
Version: 0.2.1
Summary: Typed runtime config and secrets for Educate! apps, backed by Custodian
Project-URL: Repository, https://github.com/experienceeducate/educate-secrets-manager
License-Expression: MIT
License-File: LICENSE
Keywords: configuration,fastapi,pydantic,secrets
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: Pydantic :: 2
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.12
Requires-Dist: httpx>=0.27
Requires-Dist: nats-py>=2.7
Requires-Dist: pydantic-settings>=2.3
Requires-Dist: pydantic>=2.7
Provides-Extra: dev
Requires-Dist: mypy>=1.10; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Requires-Dist: respx>=0.21; extra == 'dev'
Requires-Dist: ruff>=0.5; extra == 'dev'
Description-Content-Type: text/markdown

# educate-custodian-consumer

Typed runtime config and secrets for Educate! apps, backed by
[Custodian](https://github.com/experienceeducate/educate-secrets-manager).

Declare what your app needs as ordinary typed fields. This package fetches
them at startup, keeps them current when they change, and gets out of the
way — application code reads a plain attribute and never knows that one
value came from a `.env` file and another from an encrypted store over the
network, or that the second one silently rotated twenty minutes into the
process's life.

```python
from typing import Any
from educate_custodian_consumer import CustodianSettings, from_config, from_secret

class Settings(CustodianSettings):
    clerk_secret_key: str = from_secret()
    database_url: str = from_secret("postgresql://localhost:5432/sample_dev")
    max_upload_size_mb: int = from_config(10)
    checkout_rollout: dict[str, Any] = from_config({})
    registration_flow_enabled: bool = from_config(False, name="ENABLE_NEW_REGISTRATION_FLOW")

    log_level: str = "INFO"   # plain env config, read identically
```

```python
from contextlib import asynccontextmanager
from educate_custodian_consumer import CustodianClient
from fastapi import FastAPI

settings = Settings()
client = CustodianClient(settings)

@asynccontextmanager
async def lifespan(app: FastAPI):
    await client.start()          # fetch, assert fully loaded, subscribe
    try:
        yield
    finally:
        await client.close()

app = FastAPI(lifespan=lifespan)

@app.get("/limits")
async def limits():
    return {"max_upload_size_mb": settings.max_upload_size_mb}   # always current
```

FastAPI is the example, not a dependency: the client is `httpx` + `nats-py`
+ pydantic, and works the same from a worker or a script.

## Install

```bash
pip install educate-custodian-consumer
```

Python 3.12+.

## Configuration

Every setting below is read from the environment (or `.env`), and only from
there. They say *where* to fetch and *with what*, so they can't themselves
come from Custodian.

| Env var | Default | |
|---|---|---|
| `CUSTODIAN_ORG_SLUG` | — | required |
| `CUSTODIAN_APP_SLUG` | — | required |
| `CUSTODIAN_ENV_SLUG` | — | required |
| `CUSTODIAN_API_KEY` | — | the `sk_live_...` service-account key. Required *unless* a projected Kubernetes token is available — see below |
| `CUSTODIAN_TOKEN_PATH` | `/var/run/secrets/custodian/token` | the pod's projected ServiceAccount token; `""` disables |
| `CUSTODIAN_CREDENTIAL_MODE` | `auto` | `auto`, `workload_identity` or `api_key` |
| `CUSTODIAN_BASE_URL` | `http://localhost:8080` | |
| `CUSTODIAN_UPDATE_TRANSPORT` | `auto` | `auto`, `watch`, `nats` or `none` |
| `CUSTODIAN_NATS_URL` | `nats://localhost:4222` | only used by the `nats` transport; empty disables it |
| `CUSTODIAN_FETCH_MODE` | `auto` | `auto`, `batch` or `per_name` |
| `CUSTODIAN_TOKEN_REFRESH_MARGIN_SECONDS` | `60` | |
| `CUSTODIAN_HTTP_TIMEOUT_SECONDS` | `10.0` | |
| `CUSTODIAN_WATCH_MAX_BACKOFF_SECONDS` | `30.0` | reconnect backoff ceiling for the watch stream |

## Credentials: in Kubernetes, there is nothing to copy

In a pod, this package authenticates with the **projected ServiceAccount
token the kubelet already writes to disk** — so no `sk_live_...` key needs
to exist for that workload at all. Nothing to mint, nothing for a human to
paste into a `Secret`, nothing anybody has to remember to rotate: the
kubelet replaces that file roughly hourly on its own.

All it takes on the app's side is the volume:

```yaml
serviceAccountName: sifa                 # the identity, mapped in Custodian
volumes:
  - name: custodian-token
    projected:
      sources:
        - serviceAccountToken:
            audience: custodian:<cluster-slug>   # per-cluster, and a security check
            expirationSeconds: 3600
            path: token
containers:
  - name: sifa
    volumeMounts:
      - name: custodian-token
        mountPath: /var/run/secrets/custodian
        readOnly: true
```

An admin maps `(cluster, namespace, serviceAccountName)` to one
`(app, environment)` in Custodian once, and that mapping is the grant.
Deleting it revokes access immediately.

**`CUSTODIAN_API_KEY` keeps working and is not deprecated.** The choice is
negotiated, not assumed: if a readable token file is present it is used,
otherwise the key is, and the decision is logged once at startup. A CI job,
a VM or a laptop has no projected token and will keep using a key
permanently — that is what keys are for.

Set `CUSTODIAN_CREDENTIAL_MODE` explicitly when you want a mistake to be
loud rather than papered over: `workload_identity` refuses to fall back to
a key that happens to still exist, which is what you want in a pod that is
*supposed* to have retired its.

## How values resolve

Three layers, each overriding the one before:

1. **Field default** — what you get with no configuration anywhere.
2. **Environment** — `.env` and real env vars, pydantic-settings' normal job.
3. **Custodian** — fetched at startup, re-fetched on change.

**Custodian wins.** Env is the bootstrap and offline-dev layer; Custodian is
authoritative at runtime.

A field marked `from_secret()` / `from_config()` with no default *must* be
supplied by one of the layers: `start()` raises rather than let the app
serve traffic half-configured. Give a field a real default and it becomes a
genuine fallback instead.

Names match case-insensitively, so a `clerk_secret_key` field resolves the
`CLERK_SECRET_KEY` the console shows. Pass `name=` only when a field can't
be named after its secret at all.

## Live updates

`start()` subscribes to this app/environment's change events and re-fetches
the affected field when one arrives — the event says *that* something
changed, never what it changed to, so the value always comes from the API.
Rotating a secret in the console updates every running replica within about
a second, with no restart and no polling — for code that reads the attribute
when it needs it. Anything your app built *from* a value once doesn't follow
it; see [Values your app captured once](#values-your-app-captured-once).

Deleting a value in Custodian reverts the field to whatever env or the field
default supplied, rather than leaving it stale.

### Two transports, negotiated

`auto` (the default) uses an authenticated stream on the API itself
(`GET .../watch`), authorized by the same short-lived token this client
already holds. If the Custodian it's pointed at predates that endpoint, it
falls back to NATS for the life of the process and logs the downgrade once.

**Prefer the default, and know why it exists.** Custodian's NATS is
in-cluster only, so a pod running on any *other* cluster could never reach
it — and this package's behaviour with no NATS is to hold its boot values.
The result was that rotating a secret silently failed to reach those apps
until someone triggered a rollout. The watch stream is served by the API
every app can already reach, so it needs no URL, no new credential and no
new network path.

`watch` refuses to fall back. `nats` pins the old behaviour. `none` turns
live updates off entirely — supported, and logged loudly every time,
because it means the process holds the values it booted with until it
restarts.

The stream is deliberately closed by the server when the authorizing token
expires; reconnecting with a fresh one is a normal part of the loop, and the
reconnect carries a cursor so a rotation landing in that gap is still
delivered.

## Values your app captured once

"Always current" is a promise about the **attribute**: `settings.x` returns
the latest value on every read. It can't reach a copy. Anything built from a
value once — at import, at startup, or on first use — keeps the value it was
built with, and rotating that value in the console changes nothing until
the process restarts. The usual suspects:

- a database engine or connection pool (`create_engine(settings.database_url)`
  at import);
- middleware whose options are passed in once (CORS origins, a session
  middleware's signing key);
- an SDK client held in a module global or behind `@lru_cache`;
- `x = settings.x` at module level, or a default argument
  `def f(key=settings.x)`.

And one more that is easy to miss: **a process that never calls `start()`
never sees Custodian at all** — a migration run, an entrypoint script, a
cron job. It runs on env and field defaults, whatever Custodian holds.

None of this needs the pod restarted. Pick whichever of these fits.

### 1. Read at use time

The default, and the fix for most cases. Read `settings.x` inside the
function that needs it, not into a module-level name.

```python
# Stale after a rotation:
JWT_SECRET = settings.jwt_secret
def sign(payload): return jwt.encode(payload, JWT_SECRET)

# Current:
def sign(payload): return jwt.encode(payload, settings.jwt_secret)
```

### 2. Rebuild when the value changes

When building the thing is expensive enough that you want to keep it, key
it on the value it was built from and rebuild when that differs. The check
runs on use, so nothing has to be told about the change.

A cached SDK client — cache per *key*, not once per process:

```python
from functools import lru_cache

def clerk_client() -> Clerk:
    return _clerk_client_for(settings.clerk_secret_key)

@lru_cache(maxsize=1)
def _clerk_client_for(secret_key: str) -> Clerk:
    return Clerk(bearer_auth=secret_key)
```

A SQLAlchemy engine — swap it, and `dispose()` the old one. Sessions already
open on the old engine finish normally; new sessions get the new one.

```python
import threading
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

_lock = threading.Lock()
_engine = None
_engine_url = None

def get_engine():
    global _engine, _engine_url
    url = settings.database_url
    if url != _engine_url:
        with _lock:
            if url != _engine_url:
                old, _engine = _engine, create_engine(url, pool_pre_ping=True)
                _engine_url = url
                if old is not None:
                    old.dispose()
    return _engine

SessionLocal = sessionmaker(autoflush=False)   # no bind at import

def get_db():
    db = SessionLocal(bind=get_engine())
    try:
        yield db
    finally:
        db.close()
```

**Rotate credentials with an overlap.** Work already running on the old
engine still authenticates with the old password, so create the new
credential, put it in Custodian, and revoke the old one only after
in-flight work has had time to finish. Revoking first turns a rotation into
an outage.

Middleware configured once — wrap it so it rebuilds when its input changes.
An ASGI wrapper, shown here with Starlette's CORS middleware:

```python
from starlette.middleware.cors import CORSMiddleware

class RebuildOnChange:
    def __init__(self, app, *, build, key):
        self._app, self._build, self._key = app, build, key
        self._built_for, self._inner = None, None

    async def __call__(self, scope, receive, send):
        current = self._key()
        if self._inner is None or current != self._built_for:
            self._inner, self._built_for = self._build(self._app), current
        await self._inner(scope, receive, send)

app.add_middleware(
    RebuildOnChange,
    build=lambda inner: CORSMiddleware(
        inner, allow_origins=settings.cors_origins_list, allow_methods=["*"]
    ),
    key=lambda: settings.cors_origins,
)
```

The same wrapper works for a session middleware's signing key. Rotating
that key signs everyone out, which is usually what rotating it is for.

### 3. One-shot fetch, for processes that don't serve

A migration, an entrypoint step or a script doesn't need live updates, but
it does need the values. Fetch once, check, close — no `start()`, so no
watch stream is opened:

```python
import asyncio
from educate_custodian_consumer import CustodianClient

async def load_once(settings) -> None:
    client = CustodianClient(settings)
    try:
        await client.refresh_all()          # unreadable secrets still fail
        settings.assert_fully_loaded()
    finally:
        await client.close()

asyncio.run(load_once(settings))
# ...then run the migration with settings.database_url
```

For Alembic, call it at the top of `env.py` before reading the URL. Do it
only when Custodian is configured for that process, so a laptop with no
`CUSTODIAN_*` set still runs migrations on its `.env`.

## `missing` vs. `unreadable`

The distinction the client exists to get right. "No value here" means fall
back to the layer underneath. "A value exists and Custodian could not
produce it" means **keep what you have and never fall back** — falling back
asserts that no value exists, and for that name one does.

So an unreadable secret fails startup, even when an env fallback would have
satisfied the field. A runtime refresh does the opposite: it logs the name
and Custodian's `request_id`, and keeps serving on the value it has.

## `fetch_mode`

`auto` (the default) fetches everything in two requests where the Custodian
it's pointed at supports batch endpoints, and falls back to one request per
field — for the life of the process, logged once — where it doesn't. That
makes this package installable against a deployment that hasn't taken the
batch endpoints yet, and it picks up the improvement the day that deployment
does.

`batch` refuses to fall back. `per_name` pins the one-request-per-field
behavior.

## What this package will never do

- **No writes.** No create, no rotate, no approvals, no admin surface. The
  `-consumer` in the name is the promise.
- **No secret value in any log line, `repr`, or exception message.** Errors
  carry the name and scope; `loaded_from_custodian` returns names only, and
  is safe in a health check.
- **No on-disk cache.** Nothing is written to a file, a temp dir, or a
  `~/.custodian` anything.
- **No env var read** other than the `CUSTODIAN_*` set above.

## License

MIT — see [LICENSE](LICENSE).
