Metadata-Version: 2.4
Name: audit-framework-keycloak
Version: 0.1.1
Summary: Keycloak IdentityResolver for audit-framework — resolve groups/roles/users via the Admin API, with a TTL cache.
Project-URL: Homepage, https://github.com/vanmarkic/audit-logger
Project-URL: Repository, https://github.com/vanmarkic/audit-logger
License-Expression: MIT
Keywords: audit,audit-log,identity,keycloak,oidc,plugin,rbac
Requires-Python: >=3.11
Requires-Dist: audit-framework<0.2,>=0.1
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == 'dev'
Provides-Extra: httpx
Requires-Dist: httpx>=0.27; extra == 'httpx'
Description-Content-Type: text/markdown

# audit-framework-keycloak

The **Keycloak `IdentityResolver`** for
[`audit-framework`](../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

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

## Use

```python
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")   # -> "alice@acme.io"
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:

```python
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:

```yaml
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

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

```bash
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`](./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:

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