Switch access control from groups to realm roles

backend/app/security/idp_jwt.py

@@ -31,6 +31,7 @@ class ProviderTokenVerifier:
         admin_realm: str | None,
         admin_client_id: str | None,
         admin_client_secret: str | None,
+        member_required_realm_roles: list[str],
     ) -> None:
         self.issuer = issuer.strip() if issuer else None
         self.jwks_url = jwks_url.strip() if jwks_url else self._infer_jwks_url(self.issuer)
@@ -63,6 +64,7 @@ class ProviderTokenVerifier:
         self._admin_token_cached: str | None = None
         self._admin_token_expires_at: float = 0
         self._principal_cache: dict[str, tuple[float, ProviderPrincipal]] = {}
+        self.member_required_realm_roles = {r.strip().lower() for r in member_required_realm_roles if r and r.strip()}
 
     @staticmethod
     def _infer_introspection_endpoint(issuer: str | None) -> str | None:
@@ -151,6 +153,7 @@ class ProviderTokenVerifier:
             name=name,
             preferred_username=preferred_username,
             groups=groups,
+            realm_roles=principal.realm_roles,
         )
         return self._enrich_groups_from_admin(enriched)
 
@@ -233,8 +236,16 @@ class ProviderTokenVerifier:
             name=principal.name,
             preferred_username=principal.preferred_username,
             groups=groups,
+            realm_roles=principal.realm_roles,
         )
 
+    def _require_member_role(self, principal: ProviderPrincipal) -> None:
+        if not self.member_required_realm_roles:
+            return
+        user_roles = {r.strip().lower() for r in principal.realm_roles if isinstance(r, str) and r.strip()}
+        if not user_roles.intersection(self.member_required_realm_roles):
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="member_forbidden")
+
     def verify_access_token(self, token: str) -> ProviderPrincipal:
         now = time.time()
         cached = self._principal_cache.get(token)
@@ -289,8 +300,18 @@ class ProviderTokenVerifier:
             name=claims.get("name"),
             preferred_username=claims.get("preferred_username"),
             groups=[str(g) for g in claims.get("groups", []) if str(g)] if isinstance(claims.get("groups"), list) else [],
+            realm_roles=[
+                str(r)
+                for r in (
+                    claims.get("realm_access", {}).get("roles", [])
+                    if isinstance(claims.get("realm_access"), dict)
+                    else []
+                )
+                if str(r)
+            ],
         )
         enriched = self._enrich_from_userinfo(principal, token)
+        self._require_member_role(enriched)
 
         exp = claims.get("exp")
         if isinstance(exp, int):
@@ -322,6 +343,7 @@ def _get_verifier() -> ProviderTokenVerifier:
         admin_realm=settings.keycloak_admin_realm,
         admin_client_id=settings.keycloak_admin_client_id,
         admin_client_secret=settings.keycloak_admin_client_secret,
+        member_required_realm_roles=settings.member_required_realm_roles,
     )