feat: SSO reads roles from the provider's groups, and needs a verified address
Everybody arriving through SSO landed as a learner and was promoted by
hand. That is fine for three people and wrong for three hundred: the
list of who may edit the bank then lives in two places and only one of
them is maintained. OIDC_ROLE_CLAIM names the claim that carries somebody's
groups ("groups" for most providers, "roles" for Entra app roles), and
OIDC_ADMIN_GROUPS / OIDC_MODERATOR_GROUPS say which values mean what.
All three shapes a provider might send are read — a list, a
space-separated string, a comma-separated one — and matched case-blind.
Applied on every sign-in, not only at creation, because a group somebody
can be added to and never removed from is not a list anybody can rely
on. The one demotion it refuses is the last administrator: a mistyped
group name should not lock everybody out of the settings page,
including the person who could fix the group name. It is logged when it
happens.
Blank OIDC_ROLE_CLAIM is the old behaviour exactly: role "user", set
once, never touched.
And an address the provider will not vouch for is not an identity. This
matches on email, so an explicit email_verified:false would hand an
existing account to whoever typed its address at a careless provider.
Refused now — only on an explicit false, since a provider that omits the
claim is not asserting anything either way.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
This commit is contained in:
parent
cc4463d42c
commit
7f36e07af0
4 changed files with 208 additions and 0 deletions
|
|
@ -86,6 +86,15 @@ class Settings(BaseSettings):
|
|||
OIDC_CLIENT_SECRET: str = ""
|
||||
OIDC_SCOPES: str = "openid email profile" # space-separated
|
||||
OIDC_PROVIDER_NAME: str = "SSO" # Display name on login button
|
||||
# Which claim carries the groups or roles a person belongs to at the
|
||||
# provider — "groups" for most, "roles" for Entra app roles. Blank means
|
||||
# no mapping: everybody who signs in through SSO is a learner, which is
|
||||
# the safe default and was the only behaviour there was.
|
||||
OIDC_ROLE_CLAIM: str = ""
|
||||
# Values in that claim which mean admin, and which mean moderator.
|
||||
# Comma-separated, matched case-insensitively. Admin wins over moderator.
|
||||
OIDC_ADMIN_GROUPS: str = ""
|
||||
OIDC_MODERATOR_GROUPS: str = ""
|
||||
|
||||
LOG_LEVEL: str = "INFO" # DEBUG, INFO, WARNING, ERROR
|
||||
|
||||
|
|
|
|||
|
|
@ -540,6 +540,7 @@ async def sso_callback(request: Request, db: Session = Depends(get_db)):
|
|||
from authlib.integrations.starlette_client import OAuth
|
||||
from starlette.responses import RedirectResponse
|
||||
from app.config import settings as cfg
|
||||
from app.services import sso_roles
|
||||
|
||||
if not cfg.OIDC_PROVIDER_URL or not cfg.OIDC_CLIENT_ID:
|
||||
raise HTTPException(status_code=400, detail="SSO is not configured")
|
||||
|
|
@ -565,6 +566,15 @@ async def sso_callback(request: Request, db: Session = Depends(get_db)):
|
|||
if not email:
|
||||
return RedirectResponse(url=f"{cfg.APP_URL}/login?error=no_email")
|
||||
|
||||
# An address the provider will not vouch for is not an identity. Matching
|
||||
# on email means whoever proves an address owns the account that already
|
||||
# uses it, so an unverified claim would hand over an existing account to
|
||||
# anybody who typed the address into a provider that does not check. Only
|
||||
# refused when the provider says so explicitly: a provider that omits the
|
||||
# claim is not asserting the address is unverified.
|
||||
if userinfo.get("email_verified") is False:
|
||||
return RedirectResponse(url=f"{cfg.APP_URL}/login?error=email_unverified")
|
||||
|
||||
# Find or create user
|
||||
user = db.query(User).filter(User.email == email).first()
|
||||
if not user:
|
||||
|
|
@ -589,5 +599,14 @@ async def sso_callback(request: Request, db: Session = Depends(get_db)):
|
|||
db.commit()
|
||||
db.refresh(user)
|
||||
|
||||
# The provider's directory decides what they are here, when it has been
|
||||
# told to. Applied on every sign-in rather than only at creation: a list
|
||||
# that can add somebody to the educators group and never take them out is
|
||||
# not a list anybody can rely on. Off unless configured — see
|
||||
# services/sso_roles.
|
||||
if sso_roles.apply(db, user, userinfo, cfg):
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
|
||||
access_token = create_access_token(data={"sub": user.email})
|
||||
return RedirectResponse(url=f"{cfg.APP_URL}/sso-callback?token={access_token}")
|
||||
|
|
|
|||
84
backend/app/services/sso_roles.py
Normal file
84
backend/app/services/sso_roles.py
Normal file
|
|
@ -0,0 +1,84 @@
|
|||
"""What a person's groups at the identity provider make them here.
|
||||
|
||||
Without this, everybody who arrives through SSO is a learner and somebody
|
||||
promotes them by hand — which is fine for three people and wrong for three
|
||||
hundred, because the list of who may edit the bank then lives in two places
|
||||
and only one of them is maintained.
|
||||
|
||||
With it, the provider's directory is the list. A person removed from the
|
||||
educators group at the provider stops being an educator here at their next
|
||||
sign-in, without anybody remembering to do it.
|
||||
|
||||
Nothing happens unless it is configured. A blank OIDC_ROLE_CLAIM means the old
|
||||
behaviour exactly: role "user", set once, never touched again.
|
||||
"""
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
ROLES = ("admin", "moderator", "user")
|
||||
|
||||
|
||||
def _values(raw) -> set[str]:
|
||||
return {part.strip().lower() for part in str(raw or "").split(",") if part.strip()}
|
||||
|
||||
|
||||
def _claimed(userinfo: dict, claim: str) -> set[str]:
|
||||
"""The groups the provider says this person is in.
|
||||
|
||||
Providers disagree about the shape: a list of strings is the common one,
|
||||
Entra sends app roles the same way, and some send a single space- or
|
||||
comma-separated string. All three are read.
|
||||
"""
|
||||
raw = (userinfo or {}).get(claim)
|
||||
if raw is None:
|
||||
return set()
|
||||
if isinstance(raw, str):
|
||||
return {part.strip().lower() for part in raw.replace(",", " ").split() if part.strip()}
|
||||
if isinstance(raw, (list, tuple, set)):
|
||||
return {str(item).strip().lower() for item in raw if str(item).strip()}
|
||||
return set()
|
||||
|
||||
|
||||
def is_configured(settings) -> bool:
|
||||
return bool(str(getattr(settings, "OIDC_ROLE_CLAIM", "") or "").strip()
|
||||
and (_values(getattr(settings, "OIDC_ADMIN_GROUPS", ""))
|
||||
or _values(getattr(settings, "OIDC_MODERATOR_GROUPS", ""))))
|
||||
|
||||
|
||||
def role_for(userinfo: dict, settings) -> str | None:
|
||||
"""The role this person's groups earn, or None when mapping is off.
|
||||
|
||||
Admin beats moderator: somebody in both groups is the more privileged of
|
||||
the two, never the lesser.
|
||||
"""
|
||||
if not is_configured(settings):
|
||||
return None
|
||||
groups = _claimed(userinfo, str(settings.OIDC_ROLE_CLAIM).strip())
|
||||
if groups & _values(settings.OIDC_ADMIN_GROUPS):
|
||||
return "admin"
|
||||
if groups & _values(settings.OIDC_MODERATOR_GROUPS):
|
||||
return "moderator"
|
||||
return "user"
|
||||
|
||||
|
||||
def apply(db, user, userinfo: dict, settings) -> str | None:
|
||||
"""Bring a user's role into line with the provider. Returns the new role, or None.
|
||||
|
||||
Demotion is the point — a directory that can only add is not a directory —
|
||||
but not of the last administrator. Losing every admin to a mistyped group
|
||||
name locks everybody out of the settings page, including the person who
|
||||
could fix the group name, so that one demotion is refused and logged.
|
||||
"""
|
||||
wanted = role_for(userinfo, settings)
|
||||
if wanted is None or wanted == user.role:
|
||||
return None
|
||||
if user.role == "admin" and wanted != "admin":
|
||||
from app.models.user import User
|
||||
others = db.query(User.id).filter(User.role == "admin", User.id != user.id).count()
|
||||
if not others:
|
||||
logger.warning("SSO would demote the last admin (%s); keeping the role", user.email)
|
||||
return None
|
||||
logger.info("SSO role for %s: %s -> %s", user.email, user.role, wanted)
|
||||
user.role = wanted
|
||||
return wanted
|
||||
96
backend/tests/test_sso_roles.py
Normal file
96
backend/tests/test_sso_roles.py
Normal file
|
|
@ -0,0 +1,96 @@
|
|||
"""What the identity provider's groups make somebody here.
|
||||
|
||||
Run: DATABASE_URL=sqlite:///:memory: PYTHONPATH=backend python -m unittest discover -s backend/tests
|
||||
"""
|
||||
import os
|
||||
os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
|
||||
|
||||
import unittest
|
||||
from types import SimpleNamespace
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from app.database import Base
|
||||
from app.models.user import User
|
||||
from app.services import sso_roles
|
||||
|
||||
|
||||
def config(claim="groups", admin="", moderator=""):
|
||||
return SimpleNamespace(OIDC_ROLE_CLAIM=claim, OIDC_ADMIN_GROUPS=admin,
|
||||
OIDC_MODERATOR_GROUPS=moderator)
|
||||
|
||||
|
||||
class RoleMappingTests(unittest.TestCase):
|
||||
def test_off_unless_configured(self):
|
||||
# The old behaviour, exactly: no claim named, no opinion about anybody.
|
||||
self.assertIsNone(sso_roles.role_for({"groups": ["peds-admins"]}, config()))
|
||||
self.assertIsNone(sso_roles.role_for({"groups": ["peds-admins"]},
|
||||
config(claim="", admin="peds-admins")))
|
||||
|
||||
def test_reads_the_three_shapes_a_provider_sends(self):
|
||||
cfg = config(admin="Peds-Admins")
|
||||
for claim in (["peds-admins"], "peds-admins", "peds-admins other",
|
||||
"peds-admins,other", ("PEDS-ADMINS",)):
|
||||
self.assertEqual(sso_roles.role_for({"groups": claim}, cfg), "admin", claim)
|
||||
|
||||
def test_admin_beats_moderator(self):
|
||||
cfg = config(admin="bosses", moderator="educators")
|
||||
self.assertEqual(
|
||||
sso_roles.role_for({"groups": ["educators", "bosses"]}, cfg), "admin")
|
||||
|
||||
def test_belonging_to_neither_is_a_learner(self):
|
||||
cfg = config(admin="bosses", moderator="educators")
|
||||
self.assertEqual(sso_roles.role_for({"groups": ["everyone"]}, cfg), "user")
|
||||
self.assertEqual(sso_roles.role_for({}, cfg), "user")
|
||||
|
||||
|
||||
class RoleSyncTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool)
|
||||
Base.metadata.create_all(self.engine)
|
||||
self.db = Session(self.engine)
|
||||
|
||||
def tearDown(self):
|
||||
self.db.close()
|
||||
self.engine.dispose()
|
||||
|
||||
def add(self, email, role):
|
||||
user = User(name=email, email=email, hashed_password="unused", role=role)
|
||||
self.db.add(user)
|
||||
self.db.commit()
|
||||
return user
|
||||
|
||||
def test_the_directory_promotes_and_demotes(self):
|
||||
# Demotion is the point. A group somebody can be added to and never
|
||||
# removed from is not a list anybody can rely on.
|
||||
cfg = config(admin="bosses", moderator="educators")
|
||||
learner = self.add("new@example.test", "user")
|
||||
self.assertEqual(sso_roles.apply(self.db, learner, {"groups": ["educators"]}, cfg),
|
||||
"moderator")
|
||||
self.assertEqual(learner.role, "moderator")
|
||||
self.assertEqual(sso_roles.apply(self.db, learner, {"groups": []}, cfg), "user")
|
||||
self.assertEqual(learner.role, "user")
|
||||
|
||||
def test_it_will_not_demote_the_last_admin(self):
|
||||
# A mistyped group name should not lock everybody out of the settings
|
||||
# page, including the person who could fix the group name.
|
||||
cfg = config(admin="bosses")
|
||||
boss = self.add("boss@example.test", "admin")
|
||||
self.assertIsNone(sso_roles.apply(self.db, boss, {"groups": ["typo"]}, cfg))
|
||||
self.assertEqual(boss.role, "admin")
|
||||
# With a second admin standing, the demotion goes through.
|
||||
self.add("other@example.test", "admin")
|
||||
self.assertEqual(sso_roles.apply(self.db, boss, {"groups": ["typo"]}, cfg), "user")
|
||||
self.assertEqual(boss.role, "user")
|
||||
|
||||
def test_an_unconfigured_site_never_touches_a_role(self):
|
||||
boss = self.add("boss@example.test", "admin")
|
||||
self.assertIsNone(sso_roles.apply(self.db, boss, {"groups": []}, config()))
|
||||
self.assertEqual(boss.role, "admin")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Reference in a new issue