feat(keycloak): opt-in trust_tenant_claim, resolver stays authoritative (#376) [stack 6/11]

docs/framework/multi-tenancy.md

@@ -182,6 +182,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`.

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,9 +79,19 @@ 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.
+
+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
 

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(

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,17 +27,36 @@ 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", "")
         if auth_header.startswith("Bearer "):
             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"
@@ -90,10 +112,42 @@ def _claims_to_user_context(
             id=cache_id,
             email=claims.get("email", ""),
             name=(claims.get("preferred_username") or claims.get("name", "")),
-            roles=mapped,
-            tenant_id=claims.get("tenant_id"),
+            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

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

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 == []

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

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