mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-10-16 15:45:04 +08:00
Derive OIDC username from email when IdP sends no username claim
Since the move to python-social-auth in 26.0, social_core's get_username falls back to a random uuid4 hex string when the IdP does not send a username claim (e.g. preferred_username, often absent with CILogon). Add a pipeline step before get_username that derives the username from the email's local part for new users, matching the pre-26.0 custos behavior. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
e69583662d
commit
df34e9b9db
@@ -119,6 +119,9 @@ AUTH_PIPELINE = (
|
||||
"galaxy.authnz.psa_authnz.verify",
|
||||
# Checks if the current social-account is already associated in the site.
|
||||
"social_core.pipeline.social_auth.social_user",
|
||||
# Custom Galaxy step: Fall back to the email's local part when the IdP sends
|
||||
# no username claim; otherwise get_username substitutes a random string.
|
||||
"galaxy.authnz.psa_authnz.username_from_email",
|
||||
# Make up a username for this person, appends a random string at the end if
|
||||
# there's any collision.
|
||||
"social_core.pipeline.user.get_username",
|
||||
@@ -707,6 +710,18 @@ def contains_required_data(response=None, is_new=False, backend=None, **kwargs):
|
||||
raise MalformedContents(err_msg=f"Missing refresh token. {hint_msg}")
|
||||
|
||||
|
||||
def username_from_email(details=None, user=None, **kwargs):
|
||||
"""
|
||||
Pipeline step: derive a username from the email's local part for new users
|
||||
whose IdP sends no username claim (e.g. no ``preferred_username``).
|
||||
"""
|
||||
if user or not details or details.get("username") or not details.get("email"):
|
||||
return
|
||||
username = ready_name_for_url(details["email"].split("@")[0]).lower()
|
||||
if username:
|
||||
return {"details": {**details, "username": username}}
|
||||
|
||||
|
||||
def verify(strategy=None, response=None, details=None, **kwargs):
|
||||
provider = strategy.config.get("SOCIAL_AUTH_SECONDARY_AUTH_PROVIDER")
|
||||
endpoint = strategy.config.get("SOCIAL_AUTH_SECONDARY_AUTH_ENDPOINT")
|
||||
|
||||
@@ -26,6 +26,7 @@ from jwt import (
|
||||
InvalidSignatureError,
|
||||
)
|
||||
from social_core.backends.open_id_connect import OpenIdConnectAuth
|
||||
from social_core.pipeline.user import get_username
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
@@ -36,6 +37,7 @@ from galaxy.authnz.psa_authnz import (
|
||||
AUTH_PIPELINE,
|
||||
decode_access_token,
|
||||
PSAAuthnz,
|
||||
username_from_email,
|
||||
)
|
||||
|
||||
|
||||
@@ -367,3 +369,37 @@ def test_oidc_config_custom_auth_pipeline_and_extra(mock_oidc_config_file, mock_
|
||||
app_config=mock_app.config,
|
||||
)
|
||||
assert psa_authnz.config["SOCIAL_AUTH_PIPELINE"] == custom_auth_pipeline + tuple(custom_auth_pipeline_extra)
|
||||
|
||||
|
||||
def _run_get_username(details):
|
||||
backend = MagicMock()
|
||||
backend.setting.side_effect = lambda name, default=None: default
|
||||
strategy = MagicMock()
|
||||
strategy.storage.user.username_max_length.return_value = 255
|
||||
strategy.storage.user.clean_username.side_effect = lambda value: value
|
||||
strategy.storage.user.user_exists.return_value = False
|
||||
return get_username(strategy, details, backend)["username"]
|
||||
|
||||
|
||||
def test_username_from_email_without_username_claim():
|
||||
details = {"email": "Jane.Doe+lab@example.org", "username": None}
|
||||
result = username_from_email(details=details)
|
||||
assert result["details"]["username"] == "janedoelab"
|
||||
assert _run_get_username(result["details"]) == "janedoelab"
|
||||
|
||||
|
||||
def test_username_from_email_keeps_username_claim():
|
||||
details = {"email": "jane@example.org", "username": "jdoe"}
|
||||
assert username_from_email(details=details) is None
|
||||
assert _run_get_username(details) == "jdoe"
|
||||
|
||||
|
||||
def test_username_from_email_skips_existing_user():
|
||||
details = {"email": "jane@example.org", "username": None}
|
||||
assert username_from_email(details=details, user=MagicMock()) is None
|
||||
|
||||
|
||||
def test_username_from_email_runs_before_get_username():
|
||||
assert AUTH_PIPELINE.index("galaxy.authnz.psa_authnz.username_from_email") < AUTH_PIPELINE.index(
|
||||
"social_core.pipeline.user.get_username"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user