audit-framework-keycloak 0.1.1

Keycloak IdentityResolver for audit-framework — resolve groups/roles/users via the Admin API, with a TTL cache.

audit-framework-keycloak

The Keycloak IdentityResolver for audit-framework. Broadcast policies target roles, groups and users; this plugin turns those into the concrete user ids — and per-user email/phone contacts — the dispatcher delivers to, by querying the Keycloak Admin REST API as a service account (AD-7).

Install

pip install audit-framework-keycloak          # bring your own transport
pip install audit-framework-keycloak[httpx]   # + httpx for the default transport

Use

from audit_framework_keycloak import (
    KeycloakIdentityResolver, CachedIdentityResolver, httpx_transport,
)

resolver = KeycloakIdentityResolver(
    base_url="https://kc.internal:8443",
    realm="acme",
    client_id="audit-svc",
    client_secret=os.environ["KEYCLOAK_CLIENT_SECRET"],
    transport=httpx_transport(shared_httpx_client),   # optional; pooled in prod
)

# Memoise the slow-changing group/role lookups (default TTL 300s).
identity = CachedIdentityResolver(resolver, ttl_seconds=300)

await identity.resolve_role("auditor")        # -> ["<user-uuid>", ...]
await identity.resolve_group("legal-team")    # by name (or "/parent/child" path)
await identity.resolve_user("<user-uuid>")    # passthrough: ["<user-uuid>"] or []
await identity.get_user_contact("<user-uuid>", "email")   # -> "[e-mail]"
await identity.get_user_contact("<user-uuid>", "sms")     # -> phone attribute

It advertises itself as the keycloak provider for the identity_resolver port via the audit_framework.plugins entry point, so it's discoverable through the registry.

End-to-end example (wiring into the pipeline)

The resolver feeds two stages: the broadcast layer (role/group/user → recipient ids) and the dispatcher (recipient id → email/SMS contact). Wire one cached instance into both:

import os
import httpx
from audit_framework.core.dispatcher import Dispatcher
from audit_framework.core.middlewares import BroadcastPolicyMiddleware, DispatchMiddleware
from audit_framework.core.pipeline import Pipeline
from audit_framework_keycloak import (
    KeycloakIdentityResolver, CachedIdentityResolver, httpx_transport,
)

http_client = httpx.AsyncClient(timeout=10.0)   # shared, pooled connections

identity = CachedIdentityResolver(
    KeycloakIdentityResolver(
        base_url="https://kc.internal:8443",
        realm="acme",
        client_id="audit-svc",
        client_secret=os.environ["KEYCLOAK_CLIENT_SECRET"],
        transport=httpx_transport(http_client),
    ),
    ttl_seconds=300,
)

# Same `identity` resolves recipients (broadcast) AND their contacts (dispatch).
dispatcher = Dispatcher(channels, renderer, notification_store, identity=identity)
pipeline = (
    Pipeline()
    # ... AuditPolicy / Redact / Store / SinkFanOut middlewares first ...
    .use(BroadcastPolicyMiddleware(policy_store, identity, throttle_store=throttle))
    .use(DispatchMiddleware(dispatcher))
)

A broadcast policy that targets a Keycloak realm role, delivered over two channels — the resolver turns auditor into the user ids, then each id into an email address:

broadcast_policies:
  - name: notify-auditors-on-delete
    match: { action: [DELETE] }
    targets:
      - { type: role,  value: auditor,    channels: [email, in_app] }
      - { type: group, value: legal-team, channels: [email] }

Service-account permissions

The client uses the OAuth2 client_credentials grant, so enable Service Accounts on it and grant it the realm-management client roles needed to read the directory: view-users, query-users, query-groups. Without them the Admin API returns 403 and resolution fails loudly.

If the service-account client lives in a different realm than the one it administers (e.g. a client in master), pass token_realm=....

No hard HTTP dependency

All Admin API access goes through an injected transport — async (method, url, *, params, headers, data, json) -> HttpResult. So the resolver is fully unit-testable without a network or httpx (the test suite is stdlib-only with a fake transport), and you control connection pooling. The bundled httpx_transport() (the httpx extra) is the production default.

Caching (AD-7)

CachedIdentityResolver wraps any resolver and caches group and role membership for a TTL (default 300s) — these are queried on every matching event but change slowly. resolve_user and get_user_contact are not cached: they're cheap passthroughs, and contact changes should take effect immediately.

Token & pagination handling

• Token: the service-account token is fetched lazily, cached, and refreshed automatically on a 401 (then the call is retried once). Concurrent resolves share a single refresh via an asyncio.Lock.
• Pagination: members/role-user listings are fetched page-by-page (first/max, Keycloak's default cap is 100) until a short page is seen, so large groups resolve completely.

Development

pip install -e packages/audit-framework
pip install -e "packages/audit-framework-keycloak[dev]"
pytest -q packages/audit-framework-keycloak    # stdlib-only; fake transport, no Keycloak

Live integration test

The unit suite drives a fake transport. For end-to-end coverage against a real server, tests/integration_test.py self-provisions a throwaway realm (client, service account, users, group, role), exercises the resolver, and tears it down. It is gated: skipped unless KEYCLOAK_URL is set and httpx is installed, so the default run and CI stay green.

docker compose -f docker-compose.integration.yml up -d        # boot Keycloak (dev mode)
pip install -e "packages/audit-framework-keycloak[httpx,dev]"
KEYCLOAK_URL=http://localhost:8080 \
KEYCLOAK_ADMIN=admin KEYCLOAK_ADMIN_PASSWORD=admin \
    pytest -q -m integration packages/audit-framework-keycloak
docker compose -f docker-compose.integration.yml down

License

MIT

For AI agents & coding assistants

This package ships its agent guide — AGENTS.md — inside the wheel (installed at <site-packages>/audit_framework_keycloak/AGENTS.md). Read it offline, with no docs site and no network, even from an airgapped Nexus PyPI mirror:

python -m audit_framework_keycloak
python -c "import audit_framework_keycloak; print(audit_framework_keycloak.overview())"

AGENTS.md is the single source of truth: mental model, one runnable quickstart, and the exact public API. audit_framework_keycloak.overview() returns it, and tests/guide_test.py compiles its examples so the guide can't drift from the code.