ewilliams28/turnstone
1
0
Fork 0
mirror of https://github.com/turnstonelabs/turnstone.git synced 2026-08-27 06:14:48 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(oidc): self-heal stranded user when role mapping fails post-create (cumulative bug-1)

Browse Source 
If apply_role_mapping raised after create_oidc_user committed (transient
storage failure, race with role deletion, etc.), provision_oidc_user's
inline safety-net was skipped — and on retry the existing-identity
branch never reached the safety-net code, leaving the user permanently
stranded with zero roles.

Extracts _ensure_default_role(storage, user_id, desired_role_ids=None)
helper. Calls it on BOTH the new-user and existing-identity paths so a
user stranded by a transient failure recovers on next login.
desired_role_ids is a hint that lets the helper skip list_user_roles
when claim-driven mapping populated at least one role; the new-user
path was already paying that query, the existing-identity path now
pays it only when claim mapping returned an empty desired set.

Documents the admin-strip behavior in the helper docstring: stripping
all roles from an OIDC user no longer locks them out, since the next
login will re-grant builtin-viewer (assigned_by='oidc-default'). The
documented way to deny an OIDC user is to unlink their OIDC identity
via the admin endpoint, not to strip roles. The pre-fix behavior
(stripped user actually locked out) was the bug.

The 'oidc-default' vs 'oidc' assigned_by distinction is preserved:
apply_role_mapping's revocation lane only touches 'oidc' rows, so the
safety-net role survives every subsequent login regardless of claims.

Six new tests cover both paths, the hint short-circuit, the
list_user_roles fallback, the missing-builtin-viewer no-op, and the
self-heal regression case for already-stranded users.

(cherry picked from commit 1c41212f15)
This commit is contained in:
Patrick Buckley
2026-05-04 12:59:53 -07:00
parent 5f5eee4aab
commit 3e2fe0bc9d
2 changed files with 194 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_oidc.py
+156
View File
@@ -19,6 +19,7 @@ from turnstone.core.oidc import (
OIDCConfig,
OIDCError,
OIDCKeyNotFoundError,
_ensure_default_role,
_sanitize_log_text,
apply_role_mapping,
build_authorize_url,
@@ -1663,6 +1664,161 @@ class TestProvisionOIDCUser:
storage.assign_role.assert_not_called()
def test_existing_identity_self_heals_zero_roles(self):
"""Existing identity user with zero roles -> safety-net assigns builtin-viewer.
Models the bug-1 strand: a prior login committed user + identity but
``apply_role_mapping`` raised before reaching the safety-net. On the
next login the existing-identity branch must self-heal.
"""
config = _make_config()
existing_user = {
"user_id": "u-stranded",
"username": "alice",
"display_name": "Alice",
"password_hash": "!oidc",
}
existing_identity = {
"issuer": "https://idp.example.com",
"subject": "sub-stranded",
"user_id": "u-stranded",
"email": "alice@example.com",
"created": "2024-01-01T00:00:00",
"last_login": "2024-01-01T00:00:00",
}
storage = _mock_storage(
identity=existing_identity,
user=existing_user,
role={"role_id": "builtin-viewer", "name": "Viewer"},
)
storage.list_user_roles.return_value = []
claims = {"sub": "sub-stranded", "email": "alice@example.com", "name": "Alice"}
user = provision_oidc_user(storage, config, claims)
assert user["user_id"] == "u-stranded"
storage.list_user_roles.assert_called_once_with("u-stranded")
storage.assign_role.assert_called_once_with("u-stranded", "builtin-viewer", "oidc-default")
def test_existing_identity_does_not_re_assign_when_user_has_roles(self):
"""User already has at least one role -> safety-net no-ops."""
config = _make_config()
existing_user = {
"user_id": "u1",
"username": "alice",
"display_name": "Alice",
"password_hash": "!oidc",
}
existing_identity = {
"issuer": "https://idp.example.com",
"subject": "sub-123",
"user_id": "u1",
"email": "alice@example.com",
"created": "2024-01-01T00:00:00",
"last_login": "2024-01-01T00:00:00",
}
storage = _mock_storage(
identity=existing_identity,
user=existing_user,
role={"role_id": "builtin-viewer", "name": "Viewer"},
)
storage.list_user_roles.return_value = [{"role_id": "builtin-operator"}]
claims = {"sub": "sub-123", "email": "alice@example.com"}
provision_oidc_user(storage, config, claims)
storage.list_user_roles.assert_called_once_with("u1")
storage.assign_role.assert_not_called()
def test_existing_identity_with_claim_mapped_roles_skips_default(self):
"""Claim-driven mapping populated roles -> hint short-circuits the helper."""
config = _make_config(
role_claim="groups",
role_map={"admin": "builtin-admin"},
)
existing_user = {
"user_id": "u1",
"username": "alice",
"display_name": "Alice",
"password_hash": "!oidc",
}
existing_identity = {
"issuer": "https://idp.example.com",
"subject": "sub-123",
"user_id": "u1",
"email": "alice@example.com",
"created": "2024-01-01T00:00:00",
"last_login": "2024-01-01T00:00:00",
}
storage = _mock_storage(
identity=existing_identity,
user=existing_user,
role={"role_id": "builtin-admin", "name": "Admin"},
)
claims = {"sub": "sub-123", "groups": "admin"}
provision_oidc_user(storage, config, claims)
storage.list_user_roles.assert_not_called()
storage.assign_role.assert_not_called()
def test_new_user_safety_net_still_fires(self):
"""Fresh user with no IdP-mapped roles -> builtin-viewer fallback applied."""
config = _make_config()
storage = _mock_storage(role={"role_id": "builtin-viewer", "name": "Viewer"})
storage.list_user_roles.return_value = []
new_user = {
"user_id": "u-new",
"username": "bob",
"display_name": "Bob",
"password_hash": "!oidc",
}
storage.get_user.return_value = new_user
claims = {"sub": "sub-new", "preferred_username": "bob", "email": "bob@example.com"}
provision_oidc_user(storage, config, claims)
storage.assign_role.assert_called_once()
called_args = storage.assign_role.call_args[0]
assert called_args[1] == "builtin-viewer"
assert called_args[2] == "oidc-default"
def test_new_user_safety_net_skipped_when_apply_role_mapping_assigns_role(self):
"""Claim-driven mapping populates roles -> safety-net hint short-circuits."""
config = _make_config(
role_claim="groups",
role_map={"admin": "builtin-admin"},
)
storage = _mock_storage(role={"role_id": "builtin-admin", "name": "Admin"})
new_user = {
"user_id": "u-new",
"username": "bob",
"display_name": "Bob",
"password_hash": "!oidc",
}
storage.get_user.return_value = new_user
claims = {
"sub": "sub-new",
"preferred_username": "bob",
"email": "bob@example.com",
"groups": "admin",
}
provision_oidc_user(storage, config, claims)
storage.list_user_roles.assert_not_called()
storage.assign_role.assert_not_called()
def test_ensure_default_role_noop_when_builtin_viewer_missing(self):
"""builtin-viewer absent from role table -> helper does nothing."""
storage = _mock_storage(role=None)
_ensure_default_role(storage, "u1")
storage.get_role.assert_called_once_with("builtin-viewer")
storage.list_user_roles.assert_not_called()
storage.assign_role.assert_not_called()
# ---------------------------------------------------------------------------
# Username derivation — UUID-retry fallback tiers
turnstone/core/oidc.py
+38 -14
View File
@@ -826,6 +826,41 @@ def validate_id_token(
# ---------------------------------------------------------------------------
def _ensure_default_role(
storage: Any,
user_id: str,
desired_role_ids: set[str] | None = None,
) -> None:
"""Self-heal safety-net: assign builtin-viewer if the user has zero roles.
Runs after :func:`apply_role_mapping` on both the new-user and
existing-identity paths so a user stranded by a transient failure
during initial role mapping (e.g. a DB blip after ``create_oidc_user``
committed) recovers on next login. ``assigned_by="oidc-default"``
deliberately differs from ``"oidc"`` so claim-driven revocation in
``apply_role_mapping`` leaves it alone.
The optional ``desired_role_ids`` is a hint: when the caller already
knows claim-driven mapping populated at least one role, we skip the
``list_user_roles`` query.
No-op when builtin-viewer is unavailable (admin removed it from the
role table) or the user already has at least one role.
Note: if an admin manually strips all roles from an OIDC user, this
helper will re-grant viewer on the next login. The documented way to
deny an OIDC user access is to unlink their OIDC identity, not to
strip roles.
"""
if desired_role_ids:
return
if storage.get_role("builtin-viewer") is None:
return
if storage.list_user_roles(user_id):
return
storage.assign_role(user_id, "builtin-viewer", "oidc-default")
def provision_oidc_user(
storage: Any,
config: OIDCConfig,
@@ -852,7 +887,8 @@ def provision_oidc_user(
if identity is not None:
user_id = identity["user_id"]
storage.update_oidc_identity_login(issuer, sub)
apply_role_mapping(storage, user_id, claims, config)
desired_role_ids = apply_role_mapping(storage, user_id, claims, config)
_ensure_default_role(storage, user_id, desired_role_ids)
user: dict[str, str] | None = storage.get_user(user_id)
if user is None:
raise OIDCError(f"OIDC identity references missing user: {user_id}")
@@ -876,19 +912,7 @@ def provision_oidc_user(
raise OIDCError(f"OIDC provisioning failed: {exc}") from exc
desired_role_ids = apply_role_mapping(storage, user_id, claims, config)
# Fresh user with no IdP-mapped roles: fall back to builtin-viewer so
# they can access the app. Skipping the per-row list_user_roles
# round-trip is safe because nothing else has had a chance to assign
# a role to this just-created user_id.
#
# ``assigned_by="oidc-default"`` deliberately differs from the
# ``"oidc"`` marker used by claim-driven role mapping: ``apply_role_mapping``
# only revokes rows tagged ``"oidc"`` when the corresponding claim
# disappears, so the safety-net role survives every subsequent login
# regardless of what the IdP sends in the claim.
if not desired_role_ids and storage.get_role("builtin-viewer") is not None:
storage.assign_role(user_id, "builtin-viewer", "oidc-default")
_ensure_default_role(storage, user_id, desired_role_ids)
created_user: dict[str, str] | None = storage.get_user(user_id)
if created_user is None: