Tech-debt batch 14: auth — OAuth email proof, verified invites, effective-permission admin guards (#622 #623 #624)

CHANGELOG.md

@@ -27,6 +27,28 @@ All notable changes to this project are documented in this file. The format is b
   session (#615).
 
 ### Security
+- Users: account administration never exceeds the actor's own authority. Outside a full
+  admin session, a role can be granted and an account managed (edit, disable, reset link,
+  invite) only when its effective permissions (roles plus direct grants) are within the
+  actor's: a delegate holding `users.manage`, or an admin's scoped token, cannot grant
+  `admin`, cannot reach admin/superuser accounts, and cannot grant a role mapped to
+  permissions it lacks. Nobody can disable or demote themselves or the last active
+  administrator (#624).
+- Users: OAuth sign-in links to an existing account by email, or marks a new account
+  verified, only with the provider's proof that the email is verified: OIDC
+  `email_verified: true`, Google's People API `metadata.verified` on the primary email,
+  a verified GitHub address from `/user/emails`; Microsoft (Entra) email claims are never
+  proof. An unverified account (SSO-provisioned ones included, pending invites excepted)
+  never absorbs a later OAuth login, closing an account pre-hijacking path (#622).
+  **Behaviour change:** a provider without proof cannot sign in to an existing account by
+  matching its email alone (`400 OAUTH_USER_ALREADY_EXISTS`); existing subject links still
+  sign in.
+- Tenants: accepting an invitation requires a verified email (#623). Verification is read
+  fresh from the user directory (new `DirectoryUser.email_verified`; `None` when the
+  provider doesn't track it, in which case Keycloak's `email_verified` token claim decides),
+  so an account verified after signing in can accept without signing in again.
+  **Behaviour change:** with `users.require_verification` off, self-registered accounts must
+  verify their email (or be marked verified by an admin) before accepting an invitation.
 - DB: sessions bound to a SQLAlchemy `Connection` now inherit its engine's tenant
   policy and audit configuration, preserving strict tenant isolation and audit callbacks (#630).
 - Sign-in now rotates the session: users' `on_after_login` (password, OAuth, accept-invite)

framework/core/simple_module_core/user_directory.py

@@ -45,6 +45,9 @@ class DirectoryUser:
     #: Role names, or ``None`` when the provider does not know them (Keycloak
     #: keeps roles in the IdP). Only filled when ``with_roles=True`` was asked.
     roles: tuple[str, ...] | None = None
+    #: Whether the account's email is proven, or ``None`` when the provider
+    #: does not track it (Keycloak: read the token's ``email_verified``).
+    email_verified: bool | None = None
 
     @property
     def label(self) -> str:

framework/db/simple_module_db/user_directory_sql.py

@@ -26,6 +26,7 @@ def to_directory_user(
     *,
     roles: tuple[str, ...] | None = None,
     is_active: bool = True,
+    email_verified: bool | None = None,
 ) -> DirectoryUser:
     """Convert identity columns, preserving unknown versus empty roles.
 
@@ -38,4 +39,5 @@ def to_directory_user(
         full_name=row.full_name,
         is_active=is_active,
         roles=roles,
+        email_verified=email_verified,
     )

modules/auth/auth/contracts/schemas.py

@@ -58,6 +58,9 @@ class UserContext:
     # routes). ``None`` (a session, a service key) requires nothing.
     required_permission: str | None = None
 
+    # Providers must explicitly attest ownership of the principal's email.
+    email_verified: bool = False
+
     @property
     def is_scoped(self) -> bool:
         """Whether a scope cap applies (any token-authenticated principal)."""
@@ -83,6 +86,7 @@ def from_user(
         return cls(
             id=str(user.id),
             email=user.email,
+            email_verified=getattr(user, "is_verified", False) is True,
             name=user.full_name or user.email,
             roles=[r.name for r in user.roles],
             auth_kind=auth_kind,
@@ -98,6 +102,7 @@ def to_session_dict(self) -> dict[str, Any]:
         return {
             "id": self.id,
             "email": self.email,
+            "email_verified": self.email_verified,
             "name": self.name,
             "roles": list(self.roles),
             "tenant_id": self.tenant_id,
@@ -117,6 +122,7 @@ def from_session_dict(cls, payload: Any) -> UserContext | None:
             return cls(
                 id=str(payload["id"]),
                 email=str(payload["email"]),
+                email_verified=payload.get("email_verified") is True,
                 name=str(payload["name"]),
                 roles=list(payload.get("roles") or []),
                 tenant_id=payload.get("tenant_id"),

modules/keycloak/keycloak/provider.py

@@ -108,6 +108,7 @@ def _claims_to_user_context(
         return UserContext(
             id=cache_id,
             email=claims.get("email", ""),
+            email_verified=claims.get("email_verified") is True,
             name=(claims.get("preferred_username") or claims.get("name", "")),
             roles=self._without_tenant_roles(mapped),
             tenant_id=self._trusted_tenant(claims.get("tenant_id")),

modules/keycloak/tests/test_keycloak_email_verification.py

@@ -0,0 +1,21 @@
+"""Only an explicit verified-email claim can authorize invitation acceptance."""
+
+import pytest
+from auth.contracts.schemas import UserContext
+from keycloak.provider import KeycloakAuthProvider
+
+
+@pytest.mark.parametrize("claim", [True, False, None, "true", "false", 1])
+def test_email_verification_survives_keycloak_session(claim):
+    claims = {"email": "bob@corp.com"}
+    if claim is not None:
+        claims["email_verified"] = claim
+    ctx = KeycloakAuthProvider()._claims_to_user_context(claims, cache_id="bob")
+    assert ctx.email_verified is (claim is True)
+    restored = UserContext.from_session_dict(ctx.to_session_dict())
+    assert restored.email_verified is (claim is True)
+
+
+def test_legacy_session_does_not_assert_verified_email():
+    ctx = UserContext.from_session_dict({"id": "bob", "email": "bob@corp.com", "name": "Bob"})
+    assert ctx.email_verified is False

modules/tenants/tenants/endpoints/api.py

@@ -9,8 +9,11 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 from fastapi import APIRouter, Depends, Request, Response
 from simple_module_core.permissions import grants
+from simple_module_core.user_directory import get_user_directory
 from simple_module_hosting.permissions import RequiresPermission, resolved_permissions_for
 from simple_module_hosting.token_guard import Authenticated
 
@@ -51,6 +54,17 @@ def _email_of(request: Request) -> str | None:
     return getattr(getattr(request.state, "user", None), "email", None)
 
 
+async def _email_verified(request: Request, invitations: Any, user_id: str) -> bool:
+    """Read email proof fresh from the directory; the session's principal is a
+    sign-in-time snapshot. Providers that don't track it (Keycloak) fall back to
+    the principal's token claim (#623)."""
+    found = await get_user_directory(request.app).get_many(invitations.db, [user_id])
+    known = found[user_id].email_verified if user_id in found else None
+    if known is not None:
+        return known
+    return getattr(getattr(request.state, "user", None), "email_verified", False) is True
+
+
 @router.get("/", response_model=list[MyTenantView], dependencies=[Depends(Authenticated())])
 async def list_mine(user_id: UserIdDep, service: TenantServiceDep) -> list[MyTenantView]:
     rows = await service.list_for_user(user_id)
@@ -176,7 +190,12 @@ async def accept_invitation(
     email = _email_of(request)
     if not email:
         raise TenantError("invitation_email_mismatch", status_code=403)
-    tenant = await invitations.accept(payload.get("token", ""), user_id=user_id, user_email=email)
+    tenant = await invitations.accept(
+        payload.get("token", ""),
+        user_id=user_id,
+        user_email=email,
+        email_verified=await _email_verified(request, invitations, user_id),
+    )
     switch_active(request, tenant.id)
     membership = await invitations.tenants.get_membership(tenant.id, user_id)
     role = membership.role if membership else MembershipRole.MEMBER

modules/tenants/tenants/hooks/useTenantErrors.ts

@@ -27,6 +27,7 @@ export function useTenantErrors() {
     invitation_expired: t(keys.tenants.errors.invitation_expired),
     invitation_used: t(keys.tenants.errors.invitation_used),
     invitation_email_mismatch: t(keys.tenants.errors.invitation_email_mismatch),
+    invitation_email_unverified: t(keys.tenants.errors.invitation_email_unverified),
   };
 
   /** Translated copy for a failed call: its `{detail}` code (and `limit` for

modules/tenants/tenants/invitations.py

@@ -142,7 +142,9 @@ async def lookup(
     def is_expired(invitation: Invitation) -> bool:
         return _aware(invitation.expires_at) <= _now()
 
-    async def accept(self, token: str, *, user_id: str, user_email: str) -> Tenant:
+    async def accept(
+        self, token: str, *, user_id: str, user_email: str, email_verified: bool = False
+    ) -> Tenant:
         found = await self.lookup(token, for_update=True)
         if found is None:
             raise TenantError("invitation_not_found", status_code=404)
@@ -157,6 +159,9 @@ async def accept(self, token: str, *, user_id: str, user_email: str) -> Tenant:
         # a forwarded or leaked link cannot enrol a different account.
         if user_email.strip().lower() != invitation.email:
             raise TenantError("invitation_email_mismatch", status_code=403)
+        # ...and a self-asserted, unverified address proves nothing (#623).
+        if email_verified is not True:
+            raise TenantError("invitation_email_unverified", status_code=403)
         invitation.accepted_at = _now()
         await self.tenants.add_member(
             tenant.id,

modules/tenants/tenants/locales/de.json

@@ -137,6 +137,7 @@
     "invitation_expired": "Diese Einladung ist abgelaufen.",
     "invitation_used": "Diese Einladung wurde bereits verwendet.",
     "invitation_email_mismatch": "Diese Einladung muss mit der eingeladenen E-Mail-Adresse angenommen werden.",
+    "invitation_email_unverified": "Bestätigen Sie Ihre E-Mail-Adresse, bevor Sie diese Einladung annehmen.",
     "generic": "Etwas ist schiefgelaufen. Bitte versuchen Sie es erneut.",
     "already_member": "Diese Person ist bereits Mitglied dieser Organisation.",
     "tenant_manager_required": "Nur ein Inhaber oder Administrator dieser Organisation kann dies tun.",

modules/tenants/tenants/locales/en.json

@@ -137,6 +137,7 @@
     "invitation_expired": "That invitation has expired.",
     "invitation_used": "That invitation has already been used.",
     "invitation_email_mismatch": "This invitation must be accepted by the invited email address.",
+    "invitation_email_unverified": "Verify your email address before accepting this invitation.",
     "generic": "Something went wrong. Please try again.",
     "already_member": "That person is already a member of this organisation.",
     "tenant_manager_required": "Only an owner or admin of this organisation can do that.",

modules/tenants/tests/test_tenants_invitation_verification.py

@@ -0,0 +1,46 @@
+"""A leaked invitation must not prove ownership of an unverified address (#623)."""
+
+import uuid
+
+from users.models import User
+
+
+async def test_unverified_account_cannot_accept_invitation(app, user_client):
+    app.state.users.settings.require_verification = False
+    async with (
+        user_client("owner@corp.com") as (owner, _),
+        user_client("bob@corp.com") as (
+            bob,
+            bob_id,
+        ),
+    ):
+        await owner.post("/api/tenants/", json={"name": "Corp"})
+        issued = await owner.post(
+            "/api/tenants/current/invitations",
+            json={"email": "bob@corp.com", "role": "admin"},
+        )
+        assert issued.status_code == 201
+        token = issued.json()["token"]
+        async with app.state.sm.db.session_factory() as db:
+            user = await db.get(User, uuid.UUID(bob_id))
+            user.is_verified = False
+            await db.commit()
+
+        rejected = await bob.post("/api/tenants/invitations/accept", json={"token": token})
+        assert rejected.status_code == 403
+        assert rejected.json()["detail"] == "invitation_email_unverified"
+        members = (await owner.get("/api/tenants/current/members")).json()
+        assert bob_id not in {member["user_id"] for member in members}
+        pending = (await owner.get("/api/tenants/current/invitations")).json()
+        assert issued.json()["id"] in {invitation["id"] for invitation in pending}
+
+        # Rejection neither consumes the link nor grants the invited role.
+        # Verification is read fresh, so the SAME session can accept once the
+        # account is verified — no re-sign-in needed.
+        async with app.state.sm.db.session_factory() as db:
+            user = await db.get(User, uuid.UUID(bob_id))
+            user.is_verified = True
+            await db.commit()
+        accepted = await bob.post("/api/tenants/invitations/accept", json={"token": token})
+        assert accepted.status_code == 200
+        assert accepted.json()["role"] == "admin"

modules/users/tests/_users_oauth_support.py

@@ -25,12 +25,15 @@ async def get_access_token(self, code: str, redirect_uri: str) -> dict:
     async def get_id_email(self, access_token: str) -> tuple[str, str]:
         return self._account_id, self._account_email
 
+    async def get_profile(self, access_token: str) -> dict:
+        return {"sub": self._account_id, "email": self._account_email, "email_verified": True}
+
 
 def install_fake_provider(app, account_id: str, account_email: str) -> None:
     from users.oauth import OAuthProvider
 
-    app.state.users.oauth_clients["microsoft"] = OAuthProvider(
-        "microsoft", "Microsoft", _FakeOAuthClient(account_id, account_email)
+    app.state.users.oauth_clients["oidc"] = OAuthProvider(
+        "oidc", "OIDC", _FakeOAuthClient(account_id, account_email)
     )
 
 
@@ -70,11 +73,11 @@ async def seed_password_user(app, email: str, **fields) -> uuid.UUID:
 
 async def oauth_callback(client: httpx.AsyncClient) -> httpx.Response:
     """Drive /login then /callback; return the callback response."""
-    login = await client.get("/api/users/auth/microsoft/login", follow_redirects=False)
+    login = await client.get("/api/users/auth/oidc/login", follow_redirects=False)
     assert login.status_code == 302
     state = parse_qs(urlparse(login.headers["location"]).query)["state"][0]
     return await client.get(
-        f"/api/users/auth/microsoft/callback?code=abc&state={state}",
+        f"/api/users/auth/oidc/callback?code=abc&state={state}",
         follow_redirects=False,
     )
 

modules/users/tests/demo/test_users_demo_credentials.py

@@ -7,6 +7,7 @@
 from simple_module_test import forge_session_cookie
 from users.auth_local.token_api import _create_token_pair
 from users.demo import ensure_demo_users
+from users.models import User
 
 
 @pytest.fixture
@@ -37,11 +38,18 @@ async def test_demo_token_writes_follow_read_only_setting(
     demo_app.state.users.settings.demo_read_only = read_only
     headers = {"Authorization": f"Bearer {demo_credentials[credential]}"}
     # A successful read proves this credential actually resolves the demo identity.
-    target = demo_app.state.users.demo_user_ids[1]
     assert (await demo_client.get("/api/users/admin", headers=headers)).status_code == 200
+    # A role-less account: the PAT's scopes (users.manage, api_tokens.use) cover it, so
+    # only the demo read-only gate decides. A target holding permissions beyond the
+    # token's scopes is refused regardless (#624).
+    async with demo_app.state.sm.db.session_factory() as db:
+        plain = User(email="plain@example.com", hashed_password=None, is_external=True)
+        db.add(plain)
+        await db.commit()
+        target = str(plain.id)
     response = await demo_client.patch(
         f"/api/users/admin/{target}",
-        json={"email": demo_app.state.users.settings.demo_user_email, "full_name": "Changed"},
+        json={"email": "plain@example.com", "full_name": "Changed"},
         headers=headers,
     )
     assert response.status_code == (403 if read_only else 200), response.text

modules/users/tests/test_users_admin_effective_permissions.py

@@ -0,0 +1,70 @@
+"""Delegated management compares effective permissions (#624): direct grants and token scopes."""
+
+import httpx
+import pytest
+from api_tokens.constants import TokenKind
+from api_tokens.service import ApiTokenService
+from simple_module_test import forge_session_cookie
+from sqlalchemy import select
+from test_api_admin import _make_user
+from users.models import Role, User
+
+pytestmark = pytest.mark.anyio
+
+
+async def _client(app, headers=None, cookies=None):
+    return httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app),
+        base_url="http://testserver",
+        headers=headers,
+        cookies=cookies,
+    )
+
+
+async def test_delegate_cannot_manage_user_with_unheld_direct_grant(users_app, users_db):
+    users_db.add(Role(name="manager"))
+    await users_db.commit()
+    users_app.state.sm.permissions.map_role("manager", ["users.manage"])
+    actor = await _make_user(users_db, "manager@example.com", role_names=["manager", "user"])
+    target = await _make_user(users_db, "granted@example.com", role_names=["user"])
+    ordinary = await _make_user(users_db, "plain@example.com", role_names=["user"])
+
+    async def direct_grants(request, user):
+        return {"files.delete"} if user.id == str(target.id) else set()
+
+    users_app.state.sm.permissions.add_grant_source(direct_grants)
+    cookie = forge_session_cookie(
+        str(users_app.state.sm.settings.secret_key), {"user_id": str(actor.id)}
+    )
+    async with await _client(users_app, cookies={"session": cookie}) as client:
+        denied = await client.patch(f"/api/users/admin/{target.id}/disable")
+        allowed = await client.patch(f"/api/users/admin/{ordinary.id}/disable")
+    assert denied.status_code == 403
+    assert allowed.status_code == 200
+    await users_db.refresh(target)
+    assert target.is_active is True
+
+
+async def test_admin_pat_cannot_grant_role_beyond_its_scopes(users_app, users_db):
+    users_db.add(Role(name="deleter"))
+    await users_db.commit()
+    users_app.state.sm.permissions.map_role("deleter", ["files.delete"])
+    admin = (
+        await users_db.execute(select(User).where(User.email == "admin@example.com"))
+    ).scalar_one()
+    target = await _make_user(users_db, "target@example.com", role_names=[])
+    _, pat = await ApiTokenService(users_db).create(
+        kind=TokenKind.PERSONAL,
+        name="scoped admin",
+        scopes=["users.manage", "api_tokens.use"],
+        expires_in_days=1,
+        max_lifetime_days=30,
+        owner_user_id=admin.id,
+    )
+    await users_db.commit()
+    async with await _client(users_app, headers={"Authorization": f"Bearer {pat}"}) as client:
+        url = f"/api/users/admin/{target.id}/roles"
+        beyond = await client.put(url, json={"role_names": ["deleter"]})
+        assert beyond.status_code == 403
+        within = await client.put(url, json={"role_names": []})
+        assert within.status_code == 200