fix(users): apply user updates only after the payload validates (#1360)

`PUT /api/users/me` and `PUT /api/admin/users/<id>` write the new
password hash, and then the profile fields, before the rest of the
payload is checked. When the request is rejected further down as an
invalid role, an admin-only setting, an invalid settings value, the
route answers 400 with those writes already committed, so the caller
sees an error while the password has in fact changed.

Both routes now validate the whole payload before touching the database,
and the password hash is folded into the same `update_user` call as the
other fields so the field write is a single transaction. Error messages,
status codes and the order they are reported in are unchanged.

## Verification

- New tests in `tests/core/test_self_user_routes.py` and
`tests/core/test_admin_users_api.py` assert that a rejected update
leaves the password, profile fields and role as they were, plus a
positive case that a valid payload still applies all three. They fail on
current main and pass here.
- Full suite (3150), ruff, ruff format, basedpyright, vulture green.
This commit is contained in:
splitsec2
2026-09-20 12:50:14 -04:00
committed by GitHub
parent acd59f7cbb
commit 127dd82615
4 changed files with 121 additions and 9 deletions
+12 -5
View File
@@ -280,6 +280,7 @@ def register_admin_routes(app: Flask, user_db: UserDB) -> None:
# Handle optional password update
password = data.get("password", "")
password_hash: str | None = None
if password:
if not capabilities["canSetPassword"]:
return jsonify(
@@ -292,7 +293,7 @@ def register_admin_routes(app: Flask, user_db: UserDB) -> None:
return jsonify(
{"error": f"Password must be at least {MIN_PASSWORD_LENGTH} characters"}
), 400
user_db.update_user(user_id, password_hash=generate_password_hash(password))
password_hash = generate_password_hash(password)
# Update user fields
user_fields = {}
@@ -358,10 +359,8 @@ def register_admin_routes(app: Flask, user_db: UserDB) -> None:
if field in user_fields and user_fields[field] == user.get(field):
user_fields.pop(field)
if user_fields:
user_db.update_user(user_id, **user_fields)
# Update per-user settings
# Validate per-user settings
validated_settings: dict[str, Any] | None = None
if "settings" in data:
if not isinstance(data["settings"], dict):
return jsonify({"error": "Settings must be an object"}), 400
@@ -375,6 +374,14 @@ def register_admin_routes(app: Flask, user_db: UserDB) -> None:
}
), 400
# Apply the writes only once the whole payload has been accepted.
if password_hash is not None:
user_fields["password_hash"] = password_hash
if user_fields:
user_db.update_user(user_id, **user_fields)
if validated_settings is not None:
user_db.set_user_settings(user_id, validated_settings)
# Ensure runtime reads see updated per-user overrides immediately.
try:
+11 -4
View File
@@ -302,6 +302,7 @@ def register_self_user_routes(app: Flask, user_db: UserDB) -> None:
auth_source = capabilities["authSource"]
password = data.get("password", "")
password_hash: str | None = None
if password:
if not capabilities["canSetPassword"]:
return jsonify(
@@ -314,7 +315,7 @@ def register_self_user_routes(app: Flask, user_db: UserDB) -> None:
return jsonify(
{"error": f"Password must be at least {MIN_PASSWORD_LENGTH} characters"}
), 400
user_db.update_user(user_id, password_hash=generate_password_hash(password))
password_hash = generate_password_hash(password)
user_fields: dict[str, Any] = {}
if "email" in data:
@@ -363,9 +364,7 @@ def register_self_user_routes(app: Flask, user_db: UserDB) -> None:
if field in user_fields and user_fields[field] == user.get(field):
user_fields.pop(field)
if user_fields:
user_db.update_user(user_id, **user_fields)
validated_settings: dict[str, Any] | None = None
if "settings" in data:
settings_payload = data["settings"]
if not isinstance(settings_payload, dict):
@@ -397,6 +396,14 @@ def register_self_user_routes(app: Flask, user_db: UserDB) -> None:
}
), 400
# Apply the writes only once the whole payload has been accepted.
if password_hash is not None:
user_fields["password_hash"] = password_hash
if user_fields:
user_db.update_user(user_id, **user_fields)
if validated_settings is not None:
user_db.set_user_settings(user_id, validated_settings)
try:
app_config.refresh(force=True)
+34
View File
@@ -980,6 +980,40 @@ class TestAdminUserPasswordUpdate:
assert resp.status_code == 400
assert "Cannot set password for PROXY users" in resp.json["error"]
def test_update_password_not_applied_when_role_is_rejected(self, admin_client, user_db):
"""A rejected role leaves the stored password untouched."""
user = user_db.create_user(username="alice", role="user", password_hash="old_hash")
resp = admin_client.put(
f"/api/admin/users/{user['id']}",
json={"password": "newpass99", "role": "superuser"},
)
assert resp.status_code == 400
assert resp.json["error"] == "Role must be 'admin' or 'user'"
updated = user_db.get_user(user_id=user["id"])
assert updated["password_hash"] == "old_hash"
assert updated["role"] == "user"
def test_update_password_not_applied_when_settings_are_rejected(self, admin_client, user_db):
"""A rejected settings payload leaves the password and fields untouched."""
user = user_db.create_user(username="alice", role="user", password_hash="old_hash")
resp = admin_client.put(
f"/api/admin/users/{user['id']}",
json={
"password": "newpass99",
"role": "admin",
"settings": {"BOOK_LANGUAGE": ["klingon"]},
},
)
assert resp.status_code == 400
assert resp.json["error"] == "Invalid settings payload"
updated = user_db.get_user(user_id=user["id"])
assert updated["password_hash"] == "old_hash"
assert updated["role"] == "user"
# ---------------------------------------------------------------------------
# POST /api/admin/users/sync-cwa
+64
View File
@@ -242,3 +242,67 @@ def test_users_me_update_rejects_oidc_email_change(app, user_db):
assert resp.status_code == 400
assert resp.json["error"] == "Cannot change email for OIDC users"
assert user_db.get_user(user_id=user["id"])["email"] is None
def test_users_me_update_keeps_password_and_profile_when_settings_rejected(app, user_db):
user = user_db.create_user(
username="alice",
display_name="Alice Old",
password_hash="old_hash",
)
client = _authed_client_for_user(app, user)
with patch("shelfmark.core.self_user_routes.load_active_auth_mode", return_value="builtin"):
with patch(
"shelfmark.core.self_user_routes.app_config.get",
side_effect=_visible_sections_config_get(["delivery"]),
):
resp = client.put(
"/api/users/me",
json={
"password": "newpass99",
"display_name": "Alice New",
"settings": {
"USER_NOTIFICATION_ROUTES": [
{"event": "all", "url": "ntfys://ntfy.sh/alice"}
],
},
},
)
assert resp.status_code == 400
assert resp.json["error"] == "Some settings are admin-only"
stored = user_db.get_user(user_id=user["id"])
assert stored["password_hash"] == "old_hash"
assert stored["display_name"] == "Alice Old"
def test_users_me_update_applies_password_profile_and_settings_together(app, user_db):
user = user_db.create_user(
username="alice",
display_name="Alice Old",
password_hash="old_hash",
)
client = _authed_client_for_user(app, user)
with patch("shelfmark.core.self_user_routes.load_active_auth_mode", return_value="builtin"):
with patch(
"shelfmark.core.self_user_routes.app_config.get",
side_effect=_visible_sections_config_get(["delivery"]),
):
resp = client.put(
"/api/users/me",
json={
"password": "newpass99",
"display_name": "Alice New",
"settings": {"DESTINATION": "/books/alice"},
},
)
assert resp.status_code == 200
stored = user_db.get_user(user_id=user["id"])
assert stored["password_hash"] != "old_hash"
assert stored["display_name"] == "Alice New"
assert user_db.get_user_settings(user["id"])["DESTINATION"] == "/books/alice"