cr-service

WorkOS authentication, logging and observability for Climate Resource FastAPI services.

Services generated from copier-python-service each carried their own copy of the logging, Sentry, tracing, metrics and profiling modules, and each service that needed WorkOS wrote its own token verifier. This package holds one version of all of it, so a fix lands everywhere with a dependency bump.

What it provides:

• WorkOS access-token verification, with the production and staging issuers and JWKS URLs built in.
• FastAPI dependencies for the caller, required permissions and feature flags.
• Machine-to-machine (client credentials) tokens, allowed per client id.
• A request-scoped log context that the caller's ids are added to.
• The wide-event access log, correlation headers, Sentry, OpenTelemetry, Prometheus, Pyroscope and health probes.
• A token factory that signs real tokens, so tests go through the production verifier.

The infrastructure observability docs cover where the logs, metrics, traces and profiles go.

Quick start

import fastapi
from fastapi.middleware.cors import CORSMiddleware

import cr_service
from cr_service.auth import CurrentPrincipal, require_permission

from my_service import __version__

SERVICE = cr_service.ServiceInfo(name="my-service", version=__version__)

# Capture startup logs before uvicorn applies its own config.
cr_service.configure_logging(SERVICE)


class Settings(cr_service.ServiceSettings):
    cors_allow_origins: tuple[str, ...] = ()


def build_app() -> fastapi.FastAPI:
    settings = Settings()
    app = fastapi.FastAPI(title="My Service", version=__version__)
    app.add_middleware(CORSMiddleware, allow_origins=settings.cors_allow_origins, allow_headers=["*"])
    cr_service.setup(app, service=SERVICE, settings=settings)

    @app.get("/v1/me")
    def me(principal: CurrentPrincipal) -> dict[str, str]:
        return {"id": principal.id}

    @app.post("/v1/things", dependencies=[fastapi.Depends(require_permission("things:write"))])
    def create_thing() -> None: ...

    return app

setup installs, in order:

• logging and Sentry, from the settings,
• the Pyroscope profiler, when PYROSCOPE_SERVER_ADDRESS is set, tagging profiles like log lines and each request with its route template as endpoint and its method (per thread, so approximate under concurrent async requests),
• the wide-event, route-tag and forwarded-prefix middleware,
• /livez and /readyz, running any readiness_checks passed in,
• /metrics,
• an OpenTelemetry tracer provider, when OTEL_EXPORTER_OTLP_ENDPOINT is set, which switches on FastAPI's native request spans, with /livez, /readyz and /metrics left out,
• auth, unless auth=None is passed.

Add the service's own middleware, such as CORS, before calling setup, so the wide event and the trace wrap it.

Configuration

Settings come from the environment, or from .env. Field names map straight to variable names.

The OTEL_, PYROSCOPE_, SENTRY_RELEASE and GIT_COMMIT variables are read from the process environment, not from .env.

WORKOS_API_KEY is the only secret, so it alone goes in chamber, as Secrets and Secret management describe. Every other variable is public deploy configuration, which Deploying applications places.

WorkOS environments

ENVIRONMENT=production verifies tokens from the production WorkOS environment. Every other environment, local included, uses staging. WORKOS_ENVIRONMENT overrides the mapping.

Each environment's issuers, JWKS URLs and AuthKit domain are built into cr_service/workos.py, so a service never sets them. WorkOS core configuration lists the user-token issuer and JWKS, and explains why the issuer names the environment's default application. Machine tokens come from the environment's AuthKit domain, with the default application's client id as their audience.

The bookshelf CLI signs people in with its own application in each environment:

settings.public_auth_config() returns what a browser needs to sign in, for a /config endpoint.

Authentication

CurrentPrincipal answers 401 without a valid token. OptionalPrincipal is None without a token, and still answers 401 for an invalid one. require_permission(...) and require_feature_flag(...) answer 403. await try_authenticate(request) returns the caller or None and never refuses, for code that decides access itself, such as a GraphQL context. Guard on permissions, never on role names.

A verified token becomes a Principal:

• kind is user, machine, agent, local (from the local or fake provider), or anonymous for a caller let through by shadow mode. Only a service's own authenticator produces agent.
• id is the WorkOS user id, the machine client id, an organisation API key's id, or an agent's id.
• delegated_user_id is the person an agent acts for.
• credential is access_token, api_key, or none when nothing was checked, as for AUTH_PROVIDER=local.
• organization_id, permissions, feature_flags, role and roles come from the token.
• email, first_name, last_name and organization_name come from the Climate Resource JWT template.
• claims holds every verified claim.

User tokens must be RS256, signed by the environment key, carry the environment issuer, and not be expired. With WORKOS_REQUIRE_EMAIL=true they must also carry an email. When a token names the application it was minted for, that must be WORKOS_CLIENT_ID or one of WORKOS_ADDITIONAL_CLIENT_IDS, or the bookshelf CLI's application when WORKOS_ACCEPT_BOOKSHELF_TOKENS=true. Machine tokens are only accepted from client ids listed in WORKOS_MACHINE_CLIENTS, and their permissions come from that list, never from the token. Every machine token must carry an org_id, and a client listed in WORKOS_MACHINE_CLIENT_ORGANIZATIONS must name one of its organisations.

API keys

With WORKOS_ACCEPT_API_KEYS=true, a WorkOS API key works as a bearer token:

curl -H "Authorization: Bearer sk_..." https://my-service.example/v1/me

Keys are created and revoked in the accounts portal. Each key is validated with the WorkOS API, so this needs WORKOS_API_KEY.

• A user's key acts as that user in the organisation it was created in, as a user principal. Its email and name come from the user's WorkOS record.
• An organisation's key is a machine principal of that organisation, with the key id as its id.
• Either gets the permissions on the key and the organisation's feature flags, but no role.
• WORKOS_ALLOWED_ORGANIZATION_IDS and WORKOS_REQUIRED_FEATURE_FLAG apply as they do to tokens.
• token_id is the key id.

An accepted key is cached for a minute, so a revoked key stops working within a minute. A refused key is never cached, so every unknown sk_ bearer costs a WorkOS call. Put a rate limit in front of a service that accepts keys. When WorkOS cannot be reached the request gets a 503 rather than a 401.

Bookshelf tokens

With WORKOS_ACCEPT_BOOKSHELF_TOKENS=true, anyone signed in with the bookshelf CLI can call the service:

curl -H "Authorization: Bearer $(bookshelf auth token)" https://my-service.example/v1/me

The CLI refreshes the token when it is due. This covers bookshelf auth login only. Its bsat_ agent tokens are opaque to WorkOS, so they are still refused.

The JWKS is cached for an hour. An unknown key id triggers one early refetch, at most once a minute. When a refetch fails the stale keys keep serving, and with no keys at all the request gets a 503 rather than a 401.

Shadow mode

With AUTH_ENFORCE=false a caller whose credentials are missing, invalid or cannot be checked continues as the anonymous principal, and the failure is logged as shadow_fail. Authorisation is still enforced. A missing permission or feature flag, a disallowed organisation and a failed request check all answer 403.

Extending auth

setup takes an AuthConfig, whose fields are all optional:

• on_success and on_failure are callbacks run after each attempt, for example to write an audit log. They observe and never decide access.
• authenticator replaces the one built from the settings, so a service with its own token types, such as Bookshelf's agent tokens, can wrap build_authenticator(settings). Every build_authenticator call in a process shares one JWKS cache per JWKS URL and one API key verifier per WORKOS_API_KEY, so building an app more than once does not refetch keys or drop cached API keys.
• authenticator_dependency is a FastAPI dependency that returns or yields the authenticator for one request. It can take the request and request-scoped resources such as a database session, and depend on base_authenticator to wrap the app's authenticator.
• has_permission(principal, permission) decides what require_permission accepts, for a service where one permission implies another.
• request_checks run as check(request, principal) on every authenticated caller, before a dependency returns it. Raise AuthorizationError to refuse, for example a read-only credential on a POST. They run again whenever a dependency asks for the cached caller, so keep them free of side effects.
• resource_metadata_url adds an RFC 9728 resource_metadata hint to 401 responses. It is a URL, or a callable that builds one from the request.

A service with its own logging and middleware can skip setup and call install_auth(app, settings, config). It adds no middleware, so calling it again replaces the earlier installation.

from cr_service.auth import AuthConfig, Authenticator, base_authenticator, install_auth


async def request_authenticator(
    request: fastapi.Request,
    base: Annotated[Authenticator, Depends(base_authenticator)],
    session: Annotated[AsyncSession, Depends(get_session)],
) -> Authenticator:
    return AgentTokenAuthenticator(base, session, request)


install_auth(app, settings, AuthConfig(authenticator_dependency=request_authenticator))

With authenticator_dependency set, try_authenticate needs the request's authenticator passed as authenticator=. An authenticator that wraps LocalAuthenticator keeps its principal's credential of none, so the attempt is still recorded as skipped.

from cr_service import AuthConfig


def audit_success(request, principal):
    audit_log.info("authn_login_success", extra={"user_id": principal.id})


cr_service.setup(app, service=SERVICE, settings=settings, auth=AuthConfig(on_success=[audit_success]))

The callbacks run after the built-in ones. Those bind user_id, organization_id, auth_kind, auth_client_id and auth_credential to the log context, with auth_delegated_user_id for an agent, set the Sentry user id and organization_id tag, stamp enduser.id on the span, and log an auth_failed record with the reason for each refusal. Emails and names are never attached.

Each attempt also binds auth_outcome and sets it on request.state.auth_outcome. It is pass, fail, shadow_fail, or skipped for a principal whose credential is none. Responses carry it as the x-auth-status header, with shadow_fail reported as fail. A service with its own auth gets the header by binding auth_outcome itself. A route that never checks auth, or an optional one called without a token, gets no header.

Logging

cr_service.bind(**fields) adds fields to the current request's log context. Every record logged while the request runs carries them, and so does the http_request wide event emitted when it finishes. The middleware opens a fresh context per request, holding its request_id. Outside a request, with cr_service.log_scope(job_id=...): opens one.

The JSON line holds ts, level, logger and message, then service, version, commit, env and instance_id, then trace_id and span_id of the active span when tracing is on, then the request context, then the record's extra. Inside an endpoint the span is fastapi.endpoint, so Grafana links each line to its trace. The wide event logs at error for an unhandled exception or any 5xx, and at info otherwise. Health probes log at debug, or at warning when they fail.

The wide event

Every HTTP request produces one record with event set to http_request. It carries the request context, which holds request_id, and these fields:

request_id ties the signals together. It comes from x-request-id, then x-amzn-trace-id, or is minted when neither is set. The response carries it back in x-request-id, the request span carries it as an attribute and Sentry events carry it as a tag. So a log line leads to its trace, a trace to its log line, and both to the Sentry event.

Secrets

Any field whose name ends in a credential word is logged as [redacted], however deeply it is nested. So access_token, db_password, apiKey and private_key are masked, while max_tokens and password_policy are not. Names listed in LOG_REDACT_KEYS are masked too, once configure_logging or setup has run. The wide event's query and referer get the same treatment, and a bare code parameter counts too.

Routes that carry secrets in the URL, such as OAuth callbacks, can drop the query and referer entirely:

cr_service.setup(app, service=SERVICE, settings=settings, redact_paths=["/api/account"])

cr_service.redact.redact_fields(...) masks a mapping the same way, for anything logged another way.

configure_logging(SERVICE) at import reads LOG_LEVEL, LOG_FORMAT and ENVIRONMENT through ServiceSettings. It only replaces its own root handler, so handlers added by anything else stay. A service with its own logging stack can skip configure_logging and keep what it has. The context is still available: structlog users add cr_service.logging_config.merge_request_context to their processors, which masks secrets in the context it adds, and anything else can read cr_service.get_context().

Other helpers

• cr_service.auth.workos_api.WorkOSClient fetches users and organisations with WORKOS_API_KEY, pages through an organisation's members and feature flags, and validates API keys.
• cr_service.tracing.instrument_sqlalchemy(engine) adds a span per statement, nested under the endpoint span. It needs opentelemetry-instrumentation-sqlalchemy installed. Call it once per process, because the instrumentor ignores a second engine.
• cr_service.tracing.set_span_attributes(...) and current_trace_context() work without the tracing extra.
• Do not add opentelemetry-instrumentation-fastapi. FastAPI switches its native telemetry off when it finds that middleware, so the traces lose the dependency, endpoint and serialisation spans.

Testing a service

import pytest
from fastapi.testclient import TestClient

from cr_service.auth.testing import TokenFactory


@pytest.fixture
def tokens(settings):
    return TokenFactory.for_settings(settings)


@pytest.fixture
def client(settings, tokens):
    app = build_app()
    tokens.install(app, settings)
    return TestClient(app)


def test_create_thing(client, tokens):
    response = client.post("/v1/things", headers=tokens.headers(permissions=["things:write"]))
    assert response.status_code == 200

tokens.install swaps in a verifier that trusts the factory's key, keeping the app's AuthConfig callbacks. tokens.user_token(...) and tokens.machine_token(...) sign tokens shaped like the real ones, and tokens.sign(claims) signs anything, for malformed-token tests. Call cr_service.auth.clear_auth_caches() between tests that need a fresh JWKS cache or API key verifier.

Development

make virtual-environment
make test
make checks

Each pull request adds a changelog fragment, see changelog/README.md. Releases go through the Bump version workflow, and tags publish to PyPI.