fix(oidc): #1569 populate User.email from standard 'email' claim when email_claim is preferred_username

When an operator configures `Email Claim = preferred_username` (e.g. Authentik) the
  primary `_resolve_provider_email` correctly rejects the identity value as non-email
  shaped and returns None, leaving auto-provisioned users with `email=None` even though
  the same token carries a valid standard `email` claim.

  Add a narrow fallback in the auto-create-users branch only: when
  `provider.email_claim != "email"` and the primary returned None, resolve the standard
  `email` claim with the same Fall A/B shape + email_verified enforcement and use it for
  `User.email` and `UserOIDCLink.provider_email`.

  The auto-link-existing-accounts gate is left on the primary `provider_email`, so the
  GHSA Fall-B / Fall-C guards remain intact - the fallback never feeds account matching.
This commit is contained in:
maziggy
2026-06-02 10:23:21 +02:00
parent b7d7c82501
commit be15a375a6
3 changed files with 277 additions and 2 deletions
+1
View File
File diff suppressed because one or more lines are too long
+65 -2
View File
@@ -562,6 +562,57 @@ def _resolve_provider_email(provider: OIDCProvider, claims: dict, provider_sub:
return raw_email
def _resolve_standard_email_for_user_record(provider: OIDCProvider, claims: dict, provider_sub: str) -> str | None:
"""Resolve the standard 'email' claim for populating a newly-created User.email.
Issue #1569: when an operator sets email_claim to a non-email identity claim
(e.g. preferred_username on Authentik), the primary _resolve_provider_email
returns None because the identity value isn't email-shaped. This helper lets
the auto-create-users path still capture the user's real email from the
standard 'email' claim that the IdP usually sends alongside.
This is NOT a substitute for _resolve_provider_email and does NOT feed
auto_link_existing_accounts — that gate stays on the primary resolver, so
the GHSA Fall-B/C security guards remain intact.
Applies the same Fall A/B shape + email_verified logic as the primary
resolver does for the standard 'email' claim.
"""
raw_claim_value = claims.get("email")
if raw_claim_value is not None and not isinstance(raw_claim_value, str):
logger.warning(
"OIDC provider %d: standard 'email' claim has unexpected type %s for sub=%r, ignoring",
provider.id,
type(raw_claim_value).__name__,
provider_sub,
)
return None
raw_email = raw_claim_value.lower().strip() if raw_claim_value else None
if not raw_email:
return None
if not _is_valid_email_shaped(raw_email):
logger.warning(
"OIDC provider %d: standard 'email' claim failed shape check for sub=%r, ignoring",
provider.id,
provider_sub,
)
return None
email_verified = claims.get("email_verified")
if provider.require_email_verified:
if email_verified is True:
return raw_email
logger.info(
"OIDC provider %d: ignoring fallback email for sub=%r because email_verified=%r",
provider.id,
provider_sub,
email_verified,
)
return None
if email_verified is False:
return None
return raw_email
# ---------------------------------------------------------------------------
# Settings helpers (email 2FA flag)
# ---------------------------------------------------------------------------
@@ -1905,6 +1956,18 @@ async def oidc_callback(
raw = provider_sub[:30]
candidate = re.sub(r"[^a-zA-Z0-9._-]", "", raw)[:30] or "oidcuser"
# Issue #1569: when email_claim is configured to a non-email
# identity claim (e.g. preferred_username on Authentik), the
# primary resolver returns None for the email field because the
# identity value isn't email-shaped. Fall back to the standard
# 'email' claim for User.email so the operator can split
# username-from-preferred_username and email-from-email.
# The auto-link gate above stays on provider_email, so the
# GHSA Fall-B/C guards remain intact.
user_email_for_storage = provider_email
if user_email_for_storage is None and provider.email_claim != "email":
user_email_for_storage = _resolve_standard_email_for_user_record(provider, claims, provider_sub)
username = candidate
counter = 1
while True:
@@ -1932,7 +1995,7 @@ async def oidc_callback(
new_user = User(
username=username,
email=provider_email,
email=user_email_for_storage,
# M-1: auth_source="oidc" prevents local password-reset flow
# for users who should only authenticate via OIDC.
auth_source="oidc",
@@ -1949,7 +2012,7 @@ async def oidc_callback(
user_id=new_user.id,
provider_id=provider_id,
provider_user_id=provider_sub,
provider_email=provider_email,
provider_email=user_email_for_storage,
)
)
await db.commit()
+211
View File
@@ -4241,6 +4241,217 @@ class TestOIDCEmailResolutionExtra:
assert upd2.json()["email_claim"] == "preferred_username"
# ===========================================================================
# Issue #1569: standard 'email' claim fallback for User.email during
# auto-provisioning when email_claim is a non-email identity claim.
# ===========================================================================
class TestOIDCStandardEmailFallback:
"""Issue #1569: when email_claim is set to a non-email identity claim
(e.g. preferred_username on Authentik) and the IdP token also carries a
standard 'email' claim, auto-created users get User.email from the
standard claim — without affecting the auto-link gate."""
async def _get_user_and_link(self, db_session: AsyncSession, provider_id: int, sub: str):
from sqlalchemy import select
from backend.app.models.oidc_provider import UserOIDCLink
from backend.app.models.user import User as UserModel
link_result = await db_session.execute(
select(UserOIDCLink)
.where(UserOIDCLink.provider_id == provider_id)
.where(UserOIDCLink.provider_user_id == sub)
)
link = link_result.scalar_one_or_none()
if link is None:
return None, None
user_result = await db_session.execute(select(UserModel).where(UserModel.id == link.user_id))
return user_result.scalar_one_or_none(), link
@pytest.mark.asyncio
@pytest.mark.integration
async def test_email_claim_preferred_username_falls_back_to_standard_email(
self,
async_client: AsyncClient,
db_session: AsyncSession,
):
"""email_claim=preferred_username + both claims → username from preferred_username,
email populated from the standard 'email' claim."""
private_pem, jwks_data = _make_test_rsa_key()
issuer = "https://issue1569-both.example"
client_id = "issue1569-both-client"
admin_token = await _setup_and_login(async_client, "i1569b_adm", "I1569b123!")
provider_id = await _create_provider_via_api(
async_client,
admin_token,
issuer,
client_id,
email_claim="preferred_username",
require_email_verified=False,
suffix="i1569both",
)
sub = f"sub-i1569-both-{secrets.token_hex(6)}"
await _run_oidc_callback(
async_client,
db_session,
provider_id=provider_id,
claims={
"sub": sub,
"preferred_username": "jdoe",
"email": "jdoe@example.com",
"email_verified": True,
},
private_pem=private_pem,
jwks_data=jwks_data,
issuer=issuer,
client_id=client_id,
)
db_session.expire_all()
user, link = await self._get_user_and_link(db_session, provider_id, sub)
assert user is not None and link is not None
assert user.username == "jdoe"
assert user.email == "jdoe@example.com"
assert link.provider_email == "jdoe@example.com"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_email_claim_preferred_username_no_standard_email_keeps_email_none(
self,
async_client: AsyncClient,
db_session: AsyncSession,
):
"""email_claim=preferred_username + no standard 'email' claim → behaviour unchanged:
username derived from preferred_username, User.email stays None."""
private_pem, jwks_data = _make_test_rsa_key()
issuer = "https://issue1569-noemail.example"
client_id = "issue1569-noemail-client"
admin_token = await _setup_and_login(async_client, "i1569n_adm", "I1569n123!")
provider_id = await _create_provider_via_api(
async_client,
admin_token,
issuer,
client_id,
email_claim="preferred_username",
require_email_verified=False,
suffix="i1569none",
)
sub = f"sub-i1569-none-{secrets.token_hex(6)}"
await _run_oidc_callback(
async_client,
db_session,
provider_id=provider_id,
claims={"sub": sub, "preferred_username": "jdoe2"},
private_pem=private_pem,
jwks_data=jwks_data,
issuer=issuer,
client_id=client_id,
)
db_session.expire_all()
user, link = await self._get_user_and_link(db_session, provider_id, sub)
assert user is not None and link is not None
assert user.username == "jdoe2"
assert user.email is None
assert link.provider_email is None
@pytest.mark.asyncio
@pytest.mark.integration
async def test_fallback_respects_email_verified_false(
self,
async_client: AsyncClient,
db_session: AsyncSession,
):
"""email_claim=preferred_username + standard 'email' claim with email_verified=False
→ fallback drops the email (User.email None) — matches Fall B semantics."""
private_pem, jwks_data = _make_test_rsa_key()
issuer = "https://issue1569-evfalse.example"
client_id = "issue1569-evfalse-client"
admin_token = await _setup_and_login(async_client, "i1569f_adm", "I1569f123!")
provider_id = await _create_provider_via_api(
async_client,
admin_token,
issuer,
client_id,
email_claim="preferred_username",
require_email_verified=False,
suffix="i1569evfalse",
)
sub = f"sub-i1569-evfalse-{secrets.token_hex(6)}"
await _run_oidc_callback(
async_client,
db_session,
provider_id=provider_id,
claims={
"sub": sub,
"preferred_username": "jdoe3",
"email": "jdoe3@example.com",
"email_verified": False,
},
private_pem=private_pem,
jwks_data=jwks_data,
issuer=issuer,
client_id=client_id,
)
db_session.expire_all()
user, link = await self._get_user_and_link(db_session, provider_id, sub)
assert user is not None and link is not None
assert user.username == "jdoe3"
assert user.email is None
assert link.provider_email is None
@pytest.mark.asyncio
@pytest.mark.integration
async def test_fallback_skipped_when_email_claim_is_email(
self,
async_client: AsyncClient,
db_session: AsyncSession,
):
"""email_claim='email' (default) — fallback path must NOT fire. Primary
resolver result is authoritative; this guards against the fallback
accidentally widening Fall-A/B semantics."""
private_pem, jwks_data = _make_test_rsa_key()
issuer = "https://issue1569-default.example"
client_id = "issue1569-default-client"
admin_token = await _setup_and_login(async_client, "i1569d_adm", "I1569d123!")
provider_id = await _create_provider_via_api(
async_client,
admin_token,
issuer,
client_id,
email_claim="email",
require_email_verified=True,
suffix="i1569default",
)
sub = f"sub-i1569-default-{secrets.token_hex(6)}"
# email_verified absent → Fall A drops it. Fallback must NOT
# re-instate the email since email_claim='email' is already the
# primary path.
await _run_oidc_callback(
async_client,
db_session,
provider_id=provider_id,
claims={"sub": sub, "email": "jdoe4@example.com"},
private_pem=private_pem,
jwks_data=jwks_data,
issuer=issuer,
client_id=client_id,
)
db_session.expire_all()
user, link = await self._get_user_and_link(db_session, provider_id, sub)
assert user is not None and link is not None
assert user.email is None
assert link.provider_email is None
# ===========================================================================
# E2E: Fall C (custom email claim) auto-link actually links existing user
# ===========================================================================