starred/shelfmark
1
0
Fork 0
mirror of https://github.com/calibrain/shelfmark.git synced 2026-09-24 13:40:21 +01:00
Code Issues Packages Projects Releases 10 Wiki Activity

fix: sync renamed CWA usernames safely (#1203)

Browse Source 
## Summary

- sync an existing CWA-backed user's username when CWA renames it
- keep username collisions safe by assigning a stable `__cwa` alias
instead of overwriting a local account
- allow username updates through `UserDB` and cover
rename/collision/repeat-sync behavior

Fixes #1197.

## Testing

- `uv run ruff check shelfmark tests`
- `uv run ruff format --check shelfmark tests`
- `uv run vulture shelfmark`
- `uv run pytest tests/core/test_cwa_user_sync.py
tests/core/test_user_db.py tests/core/test_admin_users_api.py
tests/core/test_auth_api.py -k "cwa or update_user"` (36 passed)
- `uv run pytest tests/ -x --tb=short -m "not integration and not e2e"
--ignore=tests/config/test_entrypoint_permissions.py -q` (2445 passed, 5
skipped)

The entrypoint permission tests were excluded locally because macOS
ships Bash 3.2, which does not support the `${1,,}` expansion used by
`entrypoint.sh`; the same failure reproduces on an unchanged checkout.
`make python-typecheck` also currently reports the existing
`settings.py:147` callback return-type mismatch on the unchanged base.

Co-authored-by: CaliBrain <calibrain@l4n.xyz>
This commit is contained in:
Floze
2026-08-15 11:40:32 -04:00
committed by GitHub
co-authored by CaliBrain
parent 52c1702419
commit 58a5b5ed27
5 changed files with 124 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shelfmark/core/cwa_user_sync.py
+1
View File
@@ -39,6 +39,7 @@ def upsert_cwa_user(
email=normalized_email,
role=role,
allow_email_link=True,
sync_username=True,
collision_strategy=collision_strategy,
alias_suffix=_CWA_ALIAS_SUFFIX,
context=context,
shelfmark/core/external_user_linking.py
+61 -3
View File
@@ -108,6 +108,7 @@ def _build_updates(
auth_source: str,
role: str,
sync_role: bool,
username: str | object,
email: str | None | object,
display_name: str | None | object,
subject_field: str | None,
@@ -116,6 +117,8 @@ def _build_updates(
updates: dict[str, Any] = {"auth_source": auth_source}
if sync_role:
updates["role"] = _normalize_role(role)
if username is not UNSET:
updates["username"] = _normalize_username(username)
if email is not UNSET:
updates["email"] = _normalize_email(email)
if display_name is not UNSET:
@@ -125,10 +128,17 @@ def _build_updates(
return updates
def _next_suffix_username(user_db: UserDB, base_username: str) -> str:
def _next_suffix_username(
user_db: UserDB,
base_username: str,
*,
exclude_user_id: int | None = None,
) -> str:
candidate = base_username
suffix = 1
while user_db.get_user(username=candidate):
while existing := user_db.get_user(username=candidate):
if exclude_user_id is not None and int(existing.get("id") or 0) == exclude_user_id:
return candidate
candidate = f"{base_username}_{suffix}"
suffix += 1
return candidate
@@ -185,6 +195,38 @@ def _resolve_create_username(
return _next_suffix_username(user_db, alias_base), None, "username_collision_alias"
def _resolve_update_username(
user_db: UserDB,
*,
current_user: dict[str, Any],
requested_username: str,
strategy: CollisionStrategy,
alias_suffix: str,
) -> str:
current_user_id = int(current_user["id"])
existing = user_db.get_user(username=requested_username)
if existing is None or int(existing.get("id") or 0) == current_user_id:
return requested_username
if strategy == "suffix":
return _next_suffix_username(
user_db,
requested_username,
exclude_user_id=current_user_id,
)
if strategy == "alias":
return _next_suffix_username(
user_db,
f"{requested_username}{alias_suffix}",
exclude_user_id=current_user_id,
)
# `takeover` can select an existing row during creation, but once an
# identity is already matched it must never replace a different username
# owner. Preserve the matched row's current collision-free name instead.
return str(current_user["username"])
def upsert_external_user(
user_db: UserDB,
*,
@@ -197,6 +239,7 @@ def upsert_external_user(
subject: str | None = None,
allow_email_link: bool = False,
sync_role: bool = True,
sync_username: bool = False,
allow_create: bool = True,
collision_strategy: CollisionStrategy = "takeover",
alias_suffix: str | None = None,
@@ -229,10 +272,26 @@ def upsert_external_user(
subject=subject,
allow_email_link=allow_email_link,
)
resolved_alias_suffix = alias_suffix or f"__{auth_source}"
update_username: str | object = UNSET
if (
matched is not None
and sync_username
and normalize_auth_source(matched.get("auth_source"), matched.get("oidc_subject"))
== auth_source
):
update_username = _resolve_update_username(
user_db,
current_user=matched,
requested_username=normalized_username,
strategy=collision_strategy,
alias_suffix=resolved_alias_suffix,
)
updates = _build_updates(
auth_source=auth_source,
role=normalized_role,
sync_role=sync_role,
username=update_username,
email=normalized_email if email is not UNSET else UNSET,
display_name=normalized_display_name if display_name is not UNSET else UNSET,
subject_field=subject_field,
@@ -261,7 +320,6 @@ def upsert_external_user(
)
return None, "not_found"
resolved_alias_suffix = alias_suffix or f"__{auth_source}"
create_username, takeover_target, create_reason = _resolve_create_username(
user_db,
auth_source=auth_source,
shelfmark/core/user_db.py
+2
View File
@@ -344,6 +344,7 @@ class UserDB:
_ALLOWED_UPDATE_COLUMNS: ClassVar[frozenset[str]] = frozenset(
{
"username",
"email",
"display_name",
"password_hash",
@@ -353,6 +354,7 @@ class UserDB:
}
)
_USER_UPDATE_STATEMENTS: ClassVar[dict[str, str]] = {
"username": "UPDATE users SET username = ? WHERE id = ?",
"email": "UPDATE users SET email = ? WHERE id = ?",
"display_name": "UPDATE users SET display_name = ? WHERE id = ?",
"password_hash": "UPDATE users SET password_hash = ? WHERE id = ?",
tests/core/test_cwa_user_sync.py
+58
View File
@@ -95,6 +95,64 @@ def test_upsert_updates_existing_cwa_user_by_username_before_email(user_db):
assert user["role"] == "admin"
def test_upsert_renames_existing_cwa_user_matched_by_email(user_db):
cwa_user = user_db.create_user(
username="old_reader",
email="reader@example.com",
role="user",
auth_source="cwa",
)
user, action = upsert_cwa_user(
user_db,
cwa_username="renamed_reader",
cwa_email="reader@example.com",
role="user",
)
assert action == "updated"
assert user["id"] == cwa_user["id"]
assert user["username"] == "renamed_reader"
assert user_db.get_user(username="old_reader") is None
def test_upsert_uses_stable_alias_when_renamed_cwa_username_is_taken(user_db):
cwa_user = user_db.create_user(
username="old_reader",
email="reader@example.com",
role="user",
auth_source="cwa",
)
local_user = user_db.create_user(
username="renamed_reader",
email="local@example.com",
role="user",
auth_source="builtin",
)
first, first_action = upsert_cwa_user(
user_db,
cwa_username="renamed_reader",
cwa_email="reader@example.com",
role="admin",
)
second, second_action = upsert_cwa_user(
user_db,
cwa_username="renamed_reader",
cwa_email="reader@example.com",
role="admin",
)
assert first_action == second_action == "updated"
assert first["id"] == second["id"] == cwa_user["id"]
assert first["username"] == second["username"] == "renamed_reader__cwa"
assert first["role"] == second["role"] == "admin"
local_after = user_db.get_user(user_id=local_user["id"])
assert local_after is not None
assert local_after["username"] == "renamed_reader"
assert local_after["email"] == "local@example.com"
def test_sync_prunes_cwa_users_missing_from_source(user_db):
active_cwa = user_db.create_user(
username="active_cwa",
tests/core/test_user_db.py
+2
View File
@@ -568,11 +568,13 @@ class TestUserCRUD:
user = user_db.create_user(username="john", role="user")
user_db.update_user(
user["id"],
username="jane",
role="admin",
email="new@example.com",
auth_source="proxy",
)
updated = user_db.get_user(user_id=user["id"])
assert updated["username"] == "jane"
assert updated["role"] == "admin"
assert updated["email"] == "new@example.com"
assert updated["auth_source"] == "proxy"