Skip to main content

Authentication Service

Authorization-decision history​

Administrators can use client.auth.export_decisions() to read one typed page from /auth/audit/decisions/export. Optional filters are tenant, result, updated_after (a datetime), limit (1–1000, default 200), and cursor (a nonnegative sequence number). Results and other server vocabulary stay open to future values; unknown response fields are preserved.

from datetime import datetime, timezone

page = client.auth.export_decisions(
    tenant="__default__",
    result="deny",
    updated_after=datetime(2026, 10, 1, tzinfo=timezone.utc),
    limit=100,
)
for event in page.events:
    print(event.occurred_at, event.object_id, event.reason)

The history is a bounded in-memory buffer in each API process. A cursor belongs to the process that returned it and is not a durable or cluster-wide checkpoint. The SDK returns one page without automatically moving across replicas. These are authorization decisions, not login/logout or session events. The separate client.activity.get_recent_activity() feed contains general user activity. Agent-tool access to the decision export requires approval because the records disclose subjects, resources, and access decisions.

The opt-in live test tests/integration/test_audit_export_live.py creates a disposable non-admin user and two temporary Workrooms. With KAMIWAZA_TEST_DISPOSABLE_IDENTITIES=1, run it through uv run pytest -m integration and --emit-evidence. Activity logging must be enabled. Use a single API process or ensure all participating clients reach the same worker: the test refuses to credit missing events merely because an export request returned HTTP 200. Count Ray Serve API replicas, not Kubernetes pods: one pod may host multiple API processes. Before creating disposable resources, four read-only export samples reject sequence numbers that refer to different event times. This detects an observed process switch or history reset, but a clean sample does not prove routing affinity. The fresh-event and pagination assertions remain required. The fixture targets the default tenant; set KAMIWAZA_TEST_TENANT_ID to the exact tenant identifier when running it in another tenant.

The authentication helpers live in kamiwaza_sdk/services/auth.py. They wrap the Keycloak-backed auth gateway that fronts the Kamiwaza APIs.

Base URL rule: set base_url=https://<host> (no /auth suffix). A quick preflight is GET {base_url}/auth/ping → 200. If base_url ends with /auth, SDK calls will double-prefix and fail.

Getting Started​

from kamiwaza_sdk import KamiwazaClient
from kamiwaza_sdk.authentication import UserPasswordAuthenticator

client = KamiwazaClient("https://localhost/api")

# Option 1 – Personal Access Token (recommended for automation)
# export KAMIWAZA_API_KEY=<pat>
# client automatically loads the token from the environment.

# Option 2 – username/password (interactive or CLI bootstrap)
client.authenticator = UserPasswordAuthenticator("admin", "kamiwaza", client.auth)

# Who am I?
me = client.auth.get_current_user()
print(me.username, me.roles)

Tip: the SDK reads KAMIWAZA_API_KEY automatically. For federated users, generate a PAT in the UI (or with client.auth.create_pat) and drop it in the environment before instantiating the client.

Note: /auth/users/me depends on Keycloak token introspection. In lite environments without Keycloak configured it may respond with 401; the token is still usable with ForwardAuth and PAT endpoints in that mode.

Tokens & Logout​

# Perform a password grant and capture the refresh token
token = client.auth.login_with_password("admin", "kamiwaza")

# Explicit refresh using the refresh token value
new_token = client.auth.refresh_access_token(token.refresh_token)

# Coordinated logout clears cookies and returns front-channel redirect metadata
logout = client.auth.logout()
print(logout.front_channel_logout_url)

UserPasswordAuthenticator handles this dance automatically: it refreshes when the token is close to expiring, keeps the refresh token, and falls back to a password grant if the refresh is rejected.

If a selected workroom is deleted while a password-authenticated session is still scoped to it, client.workrooms.leave() clears the stale in-memory credentials and session access-token state, then retries once with a fresh password grant. The shared on-disk token remains available during that attempt; a successful grant replaces it, while an exhausted grant clears the stale scoped token. PAT and API-key credentials are fixed: the SDK never clears or retries them as refreshable sessions. Custom authenticators may opt into the same bounded recovery by returning True from invalidate_session(session) -> bool after invalidating their session credentials.

Personal Access Tokens​

from kamiwaza_sdk.schemas.auth import PATCreate

# Create a PAT – token only appears once in the response
pat = client.auth.create_pat(PATCreate(name="local-cli", ttl_seconds=86400))
print(pat.token)

# List and revoke
pat_list = client.auth.list_pats()
client.auth.revoke_pat(pat_list.pats[0].jti)

Use PATs for long-lived automation. When a PAT is set in KAMIWAZA_API_KEY, the client automatically authenticates without any extra code.

Ingress authorization headers​

Platform-managed ingress authorizes requests before they reach workloads such as App Garden apps. The SDK's legacy-named helper calls /auth/validate and parses the platform-attested identity header bundle returned by that endpoint.

headers = client.auth.forward_auth_headers()
print(headers.user_id, headers.user_roles)
print(headers.signature, headers.signature_ts)

A helper also surfaces the JWKS document for verifying PATs locally:

jwks = client.auth.get_jwks()
print(jwks.keys[0].kid)

Identity Providers & Local Users​

Administrative helpers for IdP lifecycle, session purges, and local-user management mirror the REST surface:

from kamiwaza_sdk.schemas.auth import (
    RegisterIdPRequest,
    LocalUserCreateRequest,
    SessionPurgeRequest,
)

client.auth.list_identity_providers()
client.auth.purge_sessions(SessionPurgeRequest(
    tenant_id="__default__",
    subject_id="5ac2cafa-1fbd-4b66-aba1-40b6c023612c",
))

client.auth.create_local_user(LocalUserCreateRequest(
    username="new-admin",
    email="new-admin@example.com",
    password="changeme!",
    roles=["admin", "user", "viewer", "guest"],
))

These calls require an authenticated admin role and are intended for operational tooling rather than end-user flows. In Auth-on deployments, create_local_user provisions the Keycloak identity (transactional, rolls back on Keycloak failure), and reset_user_password updates Keycloak only (Keycloak is authoritative for login). In Auth-off mode, these calls remain DB-only.

Registering a customer IdP (role_mappings, require_mapped_role)​

Requires Kamiwaza Core 1.3.4 or later. A new provider denies every brokered user that no role mapping matches. Send role_mappings to admit users by an upstream claim (default claim groups):

from kamiwaza_sdk.schemas.auth import IdPRoleMapping, OIDCConfig, RegisterIdPRequest

result = client.auth.register_identity_provider(RegisterIdPRequest(
    provider="oidc",
    oidc=OIDCConfig(alias="customer-kc", issuer="https://idp.example.com/realms/cust",
                    client_id="kamiwaza", client_secret="..."),
    display_name="Customer SSO",
    role_mappings=[
        IdPRoleMapping(value="/kamiwaza-users", role="user"),
        IdPRoleMapping(value="/kamiwaza-admins", role="admin"),
    ],
))
print(result.access.require_mapped_role, result.access.role_mappings)
print(result.warnings)

Field rules (omitted means unchanged on an existing provider):

  • role_mappings omitted (None): the current mappings are kept. If an earlier save stopped part way (503), this save finishes it: the mappings that save removed stay removed and their grants are revoked.
  • role_mappings as a list replaces the full set. Send every mapping you want to keep: Core revokes what omitted mappings granted, admin included.
  • role_mappings=[] removes all mappings and revokes what they granted.
  • require_mapped_role omitted: a new provider is gated, and an existing one keeps its state, except that sending role_mappings to an ungated provider gates it, which denies every user no mapping matches.
  • require_mapped_role=False admits every user the IdP authenticates.
  • result.access is Core's read-back of what the save applied. If you send any of require_mapped_role, role_claim or role_mappings and Core does not apply them (Core before 1.3.4 ignores them), the SDK raises IdPAccessNotAppliedError; the provider was saved without the gate, so upgrade Core or delete it.
  • An APIError with status 503 from a save that revokes grants asks you to re-run the same call. Its message says which case applies: either Core could not read Keycloak and changed nothing, or Core saved the new mappings and recorded the unfinished revocation on the provider (any later save of the provider finishes it). A 409 that names a user limit is refused before any change to the provider.
  • client.auth.get_identity_provider_access(alias) reads the current state, and client.auth.list_identity_provider_denied_users(alias) lists users the gate denies.
  • ensure_redirects is deprecated in 1.3.4 and scheduled for removal in 1.5.0. The SDK sends false, so Core before 1.3.4 no longer adds the localhost redirect URI; Core 1.3.4 or later manages redirect URIs from the install.