From 10728af7f75b84a979ed9a7010397bbca8810546 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Thu, 1 Oct 2026 13:11:25 +0200 Subject: [PATCH 1/2] feat(keycloak): ignore the tenant_id JWT claim unless trust_tenant_claim is set (#376) Adds SM_KEYCLOAK_TRUST_TENANT_CLAIM (default off). Tests cover the claim ignored by default, honoured when enabled, never beating the tenants resolver, and keycloak's setup-step opt-out. Documents the setting. Claude-Session: https://claude.ai/code/session_01F8RiTBUJQnZmSq56qReZeV --- docs/framework/multi-tenancy.md | 2 + docs/modules/keycloak.md | 9 +++ modules/keycloak/keycloak/provider.py | 6 +- modules/keycloak/keycloak/settings.py | 9 ++- .../keycloak/tests/test_keycloak_module.py | 12 ++++ .../keycloak/tests/test_keycloak_provider.py | 44 +++++++++++++++ .../tests/test_tenant_claim_resolver.py | 55 +++++++++++++++++++ 7 files changed, 135 insertions(+), 2 deletions(-) create mode 100644 modules/keycloak/tests/test_tenant_claim_resolver.py diff --git a/docs/framework/multi-tenancy.md b/docs/framework/multi-tenancy.md index b1110f3f..c4efc97a 100644 --- a/docs/framework/multi-tenancy.md +++ b/docs/framework/multi-tenancy.md @@ -163,6 +163,8 @@ The `users` module's `User` is platform-global and has no `tenant_id` column (dropped in #381): a user belongs to tenants only through `tenants_membership`, and a membership change applies on their next request, no re-login needed. +`keycloak` ignores its token's `tenant_id` claim unless `trust_tenant_claim` is +set (see [keycloak](/modules/keycloak#tenant-claim)). Most auth providers set no `tenant_id` claim, so `multi_tenant` with no resolver fails every tenant-scoped query closed; the boot reports that as `SM025`. diff --git a/docs/modules/keycloak.md b/docs/modules/keycloak.md index efa87c5c..90bfea39 100644 --- a/docs/modules/keycloak.md +++ b/docs/modules/keycloak.md @@ -70,6 +70,7 @@ DB-backed via `register_module_settings("keycloak", KeycloakSettings, ...)`; boo | `realm` | `SM_KEYCLOAK_REALM` | `""` | realm name | | `client_id` | `SM_KEYCLOAK_CLIENT_ID` | `""` | also the expected JWT audience | | `client_secret` | `SM_KEYCLOAK_CLIENT_SECRET` | `""` | confidential-client secret | +| `trust_tenant_claim` | `SM_KEYCLOAK_TRUST_TENANT_CLAIM` | `false` | see [Tenant claim](#tenant-claim) | | `roles_claim_path` | — | `"realm_access.roles"` | dotted path to the roles array in the token | | `admin_role` | — | `"admin"` | | | `login_redirect_url` | — | `"/dashboard/"` | post-login landing | @@ -78,6 +79,14 @@ DB-backed via `register_module_settings("keycloak", KeycloakSettings, ...)`; boo `server_url`, `realm`, `client_id`, and `client_secret` are **required in production** — `KeycloakSettings` raises at boot if any is missing outside a non-prod environment. They may be left blank in development and filled in later through the settings admin UI. +### Tenant claim + +By default the token's `tenant_id` claim is **ignored**: `UserContext.tenant_id` stays `None` and, with `multi_tenant` on and no tenant resolver, tenant-scoped queries fail closed. The claim is only as trustworthy as the realm mapper behind it: a mapper that reads a user-editable attribute lets a user pick their tenant. + +Set `trust_tenant_claim` (`SM_KEYCLOAK_TRUST_TENANT_CLAIM=true`) only when the realm alone controls the claim (a hardcoded or admin-only protocol mapper) and the [`tenants`](/modules/tenants) module is not installed. When `tenants` is installed its resolver decides the active tenant from `tenants_membership` on every request and the claim never wins, whether or not this setting is on, so realms may stop issuing it. + +Keycloak registers no setup steps: its local `users` table is legitimately empty, so a first-run wizard keyed on a superuser count would lock the install out. A Keycloak-only install creates its first organisation from `/tenants` after the first login. + ### Role mapping `role_mapping` translates Keycloak realm roles (read from `roles_claim_path`) into the framework's role names. Roles the user has that aren't in the mapping are dropped, so only explicitly mapped roles reach the `UserContext`. diff --git a/modules/keycloak/keycloak/provider.py b/modules/keycloak/keycloak/provider.py index ee3eabf9..f4d766ab 100644 --- a/modules/keycloak/keycloak/provider.py +++ b/modules/keycloak/keycloak/provider.py @@ -86,12 +86,16 @@ def _claims_to_user_context( for r in (roles_raw or []) if self._settings and r in self._settings.role_mapping ] + # The claim is the IdP's word, not ours: it is ignored unless the + # operator vouches for the realm mapper (``trust_tenant_claim``), and + # even then a registered tenant resolver overrides it every request. + trusted = bool(self._settings and self._settings.trust_tenant_claim) return UserContext( id=cache_id, email=claims.get("email", ""), name=(claims.get("preferred_username") or claims.get("name", "")), roles=mapped, - tenant_id=claims.get("tenant_id"), + tenant_id=claims.get("tenant_id") if trusted else None, ) async def _upsert_user_cache(self, request: Request, claims: dict) -> str: diff --git a/modules/keycloak/keycloak/settings.py b/modules/keycloak/keycloak/settings.py index bb8d2785..fa83e813 100644 --- a/modules/keycloak/keycloak/settings.py +++ b/modules/keycloak/keycloak/settings.py @@ -4,7 +4,7 @@ from pydantic import Field, field_validator, model_validator from pydantic_settings import SettingsConfigDict -from simple_module_core.dotenv import env_str +from simple_module_core.dotenv import env_bool, env_str from simple_module_core.environments import NON_PROD_ENVIRONMENTS from simple_module_core.redirect_safety import non_empty_redirect from simple_module_core.settings_base import DbBackedSettings @@ -25,6 +25,13 @@ class KeycloakSettings(DbBackedSettings): client_id: str = env_str("SM_KEYCLOAK_CLIENT_ID", "") client_secret: str = env_str("SM_KEYCLOAK_CLIENT_SECRET", "") + # Off by default: the JWT ``tenant_id`` claim is only as trustworthy as the + # realm mapper that produces it. A mapper reading an attribute the user can + # edit lets them name their own tenant. Turn on only for a realm whose + # protocol mapper users cannot influence; the ``tenants`` module's + # resolver ignores the claim either way. + trust_tenant_claim: bool = env_bool("SM_KEYCLOAK_TRUST_TENANT_CLAIM", False) + roles_claim_path: str = "realm_access.roles" admin_role: str = "admin" login_redirect_url: str = DEFAULT_LOGIN_REDIRECT_URL diff --git a/modules/keycloak/tests/test_keycloak_module.py b/modules/keycloak/tests/test_keycloak_module.py index 2b5164e2..6d715039 100644 --- a/modules/keycloak/tests/test_keycloak_module.py +++ b/modules/keycloak/tests/test_keycloak_module.py @@ -45,3 +45,15 @@ def test_a_real_value_is_left_alone(self): from keycloak.settings import KeycloakSettings assert KeycloakSettings(login_redirect_url="/home/").login_redirect_url == "/home/" + + +def test_keycloak_registers_no_setup_steps(): + """Opt-out: a Keycloak install's local users table is empty forever, so any + required step (e.g. "create a superuser") would lock it out of the app.""" + from keycloak.module import KeycloakModule + from simple_module_core.setup_steps import SetupRegistry + + registry = SetupRegistry() + KeycloakModule().register_setup_steps(registry) + assert not registry + assert registry.required_steps == [] diff --git a/modules/keycloak/tests/test_keycloak_provider.py b/modules/keycloak/tests/test_keycloak_provider.py index 4b2982a4..19a26cd8 100644 --- a/modules/keycloak/tests/test_keycloak_provider.py +++ b/modules/keycloak/tests/test_keycloak_provider.py @@ -92,3 +92,47 @@ def test_extract_roles_custom_claim_path(settings): } ctx = provider._claims_to_user_context(claims, cache_id="cccc") assert ctx.roles == ["admin"] + + +def _claims(tenant="forged-tenant"): + return {"sub": "kc-t", "email": "t@example.com", "tenant_id": tenant} + + +class TestTenantClaim: + """The JWT ``tenant_id`` claim is the IdP's word, trusted only by opt-in.""" + + def test_default_is_off(self): + assert KeycloakSettings().trust_tenant_claim is False + + def test_env_var_turns_it_on(self): + # The env default is read when the class is defined, so it needs a + # fresh interpreter rather than a monkeypatch. + import os + import subprocess + import sys + + code = "from keycloak.settings import KeycloakSettings as K; print(K().trust_tenant_claim)" + env = {**os.environ, "SM_KEYCLOAK_TRUST_TENANT_CLAIM": "true"} + out = subprocess.run( + [sys.executable, "-c", code], env=env, capture_output=True, text=True, check=True + ) + assert out.stdout.strip() == "True" + + def test_claim_ignored_by_default(self, provider): + assert provider._claims_to_user_context(_claims(), cache_id="a").tenant_id is None + + def test_claim_ignored_without_settings(self): + ctx = KeycloakAuthProvider()._claims_to_user_context(_claims(), cache_id="a") + assert ctx.tenant_id is None + + def test_claim_honoured_when_trusted(self, settings): + settings.trust_tenant_claim = True + ctx = KeycloakAuthProvider(settings)._claims_to_user_context(_claims("acme"), cache_id="a") + assert ctx.tenant_id == "acme" + + def test_trusted_but_no_claim_is_none(self, settings): + settings.trust_tenant_claim = True + ctx = KeycloakAuthProvider(settings)._claims_to_user_context( + {"sub": "x", "email": "e@x.io"}, cache_id="a" + ) + assert ctx.tenant_id is None diff --git a/modules/keycloak/tests/test_tenant_claim_resolver.py b/modules/keycloak/tests/test_tenant_claim_resolver.py new file mode 100644 index 00000000..7eabaaf3 --- /dev/null +++ b/modules/keycloak/tests/test_tenant_claim_resolver.py @@ -0,0 +1,55 @@ +"""A forged ``tenant_id`` JWT claim never wins over the ``tenants`` resolver (#376). + +Lives apart from ``test_keycloak_provider.py`` because that file overrides the +``settings`` fixture, which the app-level ``app`` fixture depends on. +""" + +from __future__ import annotations + +from keycloak.provider import KeycloakAuthProvider +from keycloak.settings import KeycloakSettings +from starlette.requests import Request + +_USER_ID = "11111111-1111-1111-1111-111111111111" + + +def _request(app, user) -> Request: + request = Request( + { + "type": "http", + "app": app, + "method": "GET", + "path": "/", + "query_string": b"", + "headers": [(b"host", b"testserver")], + } + ) + request.state.user = user + return request + + +def _ctx(trust: bool, tenant: str): + provider = KeycloakAuthProvider(KeycloakSettings(trust_tenant_claim=trust)) + claims = {"sub": "kc-t", "email": "t@example.com", "tenant_id": tenant} + return provider._claims_to_user_context(claims, cache_id=_USER_ID) + + +async def test_trusted_claim_reaches_the_user_context(app): + assert _ctx(True, "forged-tenant").tenant_id == "forged-tenant" + + +async def test_forged_claim_is_not_the_tenant_when_a_resolver_is_registered(app): + assert app.state.tenant_resolver is not None + # Trusting the claim is the worst case: it is on the UserContext, yet the + # resolver answers from memberships (this user has none). + ctx = _ctx(True, "forged-tenant") + assert await app.state.tenant_resolver(_request(app, ctx)) is None + + +async def test_forged_claim_never_overrides_a_real_membership(app, tenant_client): + async with tenant_client() as member: + ctx = _ctx(True, "forged-tenant") + ctx.id = member.user_id + request = _request(app, ctx) + assert await app.state.tenant_resolver(request) == member.tenant_id + assert request.state.user.tenant_id == member.tenant_id From e7433f50e9a59e0b9f609de6c04a0c60acc9f2cf Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Thu, 1 Oct 2026 14:56:23 +0200 Subject: [PATCH 2/2] fix(keycloak): re-check the tenant claim per request and refuse tenant roles (review of #376) - A session's cached tenant_id is dropped unless trust_tenant_claim is on now; the provider reads settings through app.state.keycloak, so saves and the boot-time hydration take effect instead of the register_settings snapshot. - A trusted claim must pass is_valid_tenant_id, else it is ignored. - tenant:* roles produced by role_mapping (or cached in old sessions) are stripped, with one warning per offending role. Claude-Session: https://claude.ai/code/session_01F8RiTBUJQnZmSq56qReZeV --- docs/modules/keycloak.md | 4 +- modules/keycloak/keycloak/module.py | 4 +- modules/keycloak/keycloak/provider.py | 68 +++++++++-- .../tests/test_keycloak_tenant_hardening.py | 113 ++++++++++++++++++ 4 files changed, 178 insertions(+), 11 deletions(-) create mode 100644 modules/keycloak/tests/test_keycloak_tenant_hardening.py diff --git a/docs/modules/keycloak.md b/docs/modules/keycloak.md index 90bfea39..8b093369 100644 --- a/docs/modules/keycloak.md +++ b/docs/modules/keycloak.md @@ -85,11 +85,13 @@ By default the token's `tenant_id` claim is **ignored**: `UserContext.tenant_id` Set `trust_tenant_claim` (`SM_KEYCLOAK_TRUST_TENANT_CLAIM=true`) only when the realm alone controls the claim (a hardcoded or admin-only protocol mapper) and the [`tenants`](/modules/tenants) module is not installed. When `tenants` is installed its resolver decides the active tenant from `tenants_membership` on every request and the claim never wins, whether or not this setting is on, so realms may stop issuing it. +The setting is read on every request, not only at sign-in: turning it off drops the tenant from sessions that already exist, without anyone signing out. A trusted claim must also be a well-formed tenant id (letters, digits, `_ . : -`, at most 50 characters); anything else is ignored and logged. + Keycloak registers no setup steps: its local `users` table is legitimately empty, so a first-run wizard keyed on a superuser count would lock the install out. A Keycloak-only install creates its first organisation from `/tenants` after the first login. ### Role mapping -`role_mapping` translates Keycloak realm roles (read from `roles_claim_path`) into the framework's role names. Roles the user has that aren't in the mapping are dropped, so only explicitly mapped roles reach the `UserContext`. +`role_mapping` translates Keycloak realm roles (read from `roles_claim_path`) into the framework's role names. Roles the user has that aren't in the mapping are dropped, so only explicitly mapped roles reach the `UserContext`. A mapping onto a tenant role (`tenant:owner`, `tenant:admin`, `tenant:member`) is ignored, with a warning logged once per role: those roles come only from the `tenants` module, for the active tenant. ## Models diff --git a/modules/keycloak/keycloak/module.py b/modules/keycloak/keycloak/module.py index 0b79ac38..253dd58b 100644 --- a/modules/keycloak/keycloak/module.py +++ b/modules/keycloak/keycloak/module.py @@ -43,7 +43,9 @@ def register_settings(self, app: FastAPI) -> None: lambda s: KeycloakState(settings=s), ) - app.state.auth.auth_provider = KeycloakAuthProvider(app.state.keycloak.settings) + app.state.auth.auth_provider = KeycloakAuthProvider( + app.state.keycloak.settings, state=app.state.keycloak + ) def register_menu_items(self, registry: MenuRegistry) -> None: registry.add( diff --git a/modules/keycloak/keycloak/provider.py b/modules/keycloak/keycloak/provider.py index f4d766ab..eef2ca4b 100644 --- a/modules/keycloak/keycloak/provider.py +++ b/modules/keycloak/keycloak/provider.py @@ -7,11 +7,14 @@ from typing import TYPE_CHECKING, Any from auth.contracts.schemas import UserContext +from simple_module_core.tenancy import is_tenant_role +from simple_module_db import is_valid_tenant_id from starlette.requests import Request if TYPE_CHECKING: from keycloak.jwks import JWKSCache from keycloak.settings import KeycloakSettings + from keycloak.state import KeycloakState logger = logging.getLogger(__name__) @@ -24,9 +27,21 @@ class KeycloakAuthProvider: name = "keycloak" _is_auth_provider = True - def __init__(self, settings: KeycloakSettings | None = None) -> None: - self._settings = settings + def __init__( + self, settings: KeycloakSettings | None = None, *, state: KeycloakState | None = None + ) -> None: + self._initial_settings = settings + # The module's state object: a settings save or the boot-time DB + # hydration swaps ``state.settings``, so reading through it keeps + # ``trust_tenant_claim`` / ``role_mapping`` edits live. + self._state = state self.jwks_cache: JWKSCache | None = None + self._warned_tenant_roles: set[str] = set() + + @property + def _settings(self) -> KeycloakSettings | None: + live = getattr(self._state, "settings", None) + return live if live is not None else self._initial_settings async def resolve_user(self, request: Request) -> UserContext | None: auth_header = request.headers.get("authorization", "") @@ -34,7 +49,14 @@ async def resolve_user(self, request: Request) -> UserContext | None: return await self._resolve_bearer(request, auth_header[7:]) session = request.scope.get("session", {}) - return UserContext.from_session_dict(session.get(_SESSION_USER_CTX_KEY)) + ctx = UserContext.from_session_dict(session.get(_SESSION_USER_CTX_KEY)) + if ctx is not None: + # The cookie froze what was decided at login; re-decide with the + # current settings, so turning trust_tenant_claim off takes effect + # on sessions that already exist. + ctx.tenant_id = self._trusted_tenant(ctx.tenant_id) + ctx.roles = self._without_tenant_roles(ctx.roles) + return ctx def get_login_url(self, request: Request | None, next_url: str | None = None) -> str: return "/keycloak/login" @@ -86,18 +108,46 @@ def _claims_to_user_context( for r in (roles_raw or []) if self._settings and r in self._settings.role_mapping ] - # The claim is the IdP's word, not ours: it is ignored unless the - # operator vouches for the realm mapper (``trust_tenant_claim``), and - # even then a registered tenant resolver overrides it every request. - trusted = bool(self._settings and self._settings.trust_tenant_claim) return UserContext( id=cache_id, email=claims.get("email", ""), name=(claims.get("preferred_username") or claims.get("name", "")), - roles=mapped, - tenant_id=claims.get("tenant_id") if trusted else None, + roles=self._without_tenant_roles(mapped), + tenant_id=self._trusted_tenant(claims.get("tenant_id")), ) + def _trusted_tenant(self, value: Any) -> str | None: + """The ``tenant_id`` claim, if the operator trusts it and it is well-formed. + + The claim is the IdP's word, not ours: it is ignored unless the operator + vouches for the realm mapper (``trust_tenant_claim``), and even then a + registered tenant resolver overrides it every request. It must also pass + the same id check as any other tenant id taken from outside. + """ + if value is None or not (self._settings and self._settings.trust_tenant_claim): + return None + if not is_valid_tenant_id(value): + logger.warning("Ignoring malformed tenant_id claim %r", value) + return None + return value + + def _without_tenant_roles(self, roles: list[str]) -> list[str]: + """Drop ``tenant:*`` roles: only the tenants module may grant them, for + the active tenant. A ``role_mapping`` entry producing one would + otherwise hand every user that tenant role in every tenant.""" + kept = [] + for role in roles: + if not is_tenant_role(role): + kept.append(role) + elif role not in self._warned_tenant_roles: + self._warned_tenant_roles.add(role) + logger.warning( + "Keycloak role_mapping produced tenant role %r; ignored " + "(tenant roles come from memberships only)", + role, + ) + return kept + async def _upsert_user_cache(self, request: Request, claims: dict) -> str: try: from sqlalchemy import select diff --git a/modules/keycloak/tests/test_keycloak_tenant_hardening.py b/modules/keycloak/tests/test_keycloak_tenant_hardening.py new file mode 100644 index 00000000..0e578093 --- /dev/null +++ b/modules/keycloak/tests/test_keycloak_tenant_hardening.py @@ -0,0 +1,113 @@ +"""Tenant claims and tenant roles from Keycloak (review of #376). + +* a session's cached ``tenant_id`` is re-decided against the *current* + ``trust_tenant_claim``, so switching it off reaches existing sessions; +* a trusted claim must still be a well-formed tenant id; +* ``role_mapping`` cannot mint ``tenant:*`` roles — only memberships grant them. +""" + +from __future__ import annotations + +import logging + +import pytest +from auth.contracts.schemas import UserContext +from keycloak.provider import KeycloakAuthProvider +from keycloak.settings import KeycloakSettings +from keycloak.state import KeycloakState +from starlette.requests import Request + + +def _request(session: dict) -> Request: + return Request( + {"type": "http", "method": "GET", "path": "/", "headers": [], "session": session} + ) + + +def _session(tenant_id: str | None, roles: list[str] | None = None) -> dict: + ctx = UserContext(id="u1", email="u@x.io", name="u", roles=roles or [], tenant_id=tenant_id) + return {"user_ctx": ctx.to_session_dict()} + + +@pytest.fixture +def state() -> KeycloakState: + return KeycloakState(settings=KeycloakSettings(trust_tenant_claim=True)) + + +class TestSessionTenantFollowsTheCurrentSetting: + async def test_kept_while_trusted(self, state): + provider = KeycloakAuthProvider(state.settings, state=state) + ctx = await provider.resolve_user(_request(_session("acme"))) + assert ctx is not None + assert ctx.tenant_id == "acme" + + async def test_dropped_once_trust_is_turned_off(self, state): + provider = KeycloakAuthProvider(state.settings, state=state) + # A settings save swaps the state's settings object in place. + state.settings = KeycloakSettings(trust_tenant_claim=False) + ctx = await provider.resolve_user(_request(_session("acme"))) + assert ctx is not None + assert ctx.tenant_id is None + + async def test_malformed_cached_tenant_is_dropped(self, state): + provider = KeycloakAuthProvider(state.settings, state=state) + ctx = await provider.resolve_user(_request(_session("has space"))) + assert ctx is not None + assert ctx.tenant_id is None + + async def test_tenant_roles_in_an_old_session_are_stripped(self, state): + provider = KeycloakAuthProvider(state.settings, state=state) + ctx = await provider.resolve_user(_request(_session(None, ["user", "tenant:owner"]))) + assert ctx is not None + assert ctx.roles == ["user"] + + +@pytest.mark.parametrize("claim", ["has space", "x" * 51, "-leading", "", 42, ["acme"]]) +def test_malformed_trusted_claim_is_ignored(claim): + provider = KeycloakAuthProvider(KeycloakSettings(trust_tenant_claim=True)) + ctx = provider._claims_to_user_context({"sub": "s", "tenant_id": claim}, cache_id="a") + assert ctx.tenant_id is None + + +def test_well_formed_trusted_claim_is_kept(): + provider = KeycloakAuthProvider(KeycloakSettings(trust_tenant_claim=True)) + ctx = provider._claims_to_user_context({"sub": "s", "tenant_id": "acme-1"}, cache_id="a") + assert ctx.tenant_id == "acme-1" + + +def test_role_mapping_cannot_grant_tenant_roles(caplog): + settings = KeycloakSettings( + role_mapping={"kc-owner": "tenant:owner", "kc-admin": "tenant:admin", "kc-user": "user"} + ) + provider = KeycloakAuthProvider(settings) + claims = {"sub": "s", "realm_access": {"roles": ["kc-owner", "kc-admin", "kc-user"]}} + + with caplog.at_level(logging.WARNING, logger="keycloak.provider"): + first = provider._claims_to_user_context(claims, cache_id="a") + second = provider._claims_to_user_context(claims, cache_id="a") + + assert first.roles == ["user"] + assert second.roles == ["user"] + warnings = [r for r in caplog.records if "tenant role" in r.getMessage()] + assert len(warnings) == 2 # once per offending role, not once per login + + +def test_the_host_wires_the_provider_to_the_live_settings(): + """A save or the boot hydration replaces ``app.state.keycloak.settings``; + the provider must read through to it, not keep the boot-time object.""" + from simple_module_hosting.app_builder import create_app + from simple_module_hosting.settings import Settings + + app = create_app( + Settings( + database_url="sqlite+aiosqlite:///:memory:", + environment="testing", + secret_key="test-secret-key", + multi_tenant=False, + auth_provider="keycloak", + ) + ) + provider = app.state.auth.auth_provider + replacement = KeycloakSettings(trust_tenant_claim=True) + app.state.keycloak.settings = replacement + assert provider._settings is replacement