adding a password reset system for admins

server/auth/auth.py

@@ -127,6 +127,8 @@ def reset_password(self, username: str, new_password: str) -> bool:
 
         password_hash = self.hash_password(new_password)
         self._db.update_user_password(username, password_hash)
+        self.invalidate_user_sessions(username)
+        self._db.revoke_user_refresh_tokens(username, int(time.time()))
         return True
 
     def get_user(self, username: str) -> "UserRecord | None":

server/core/administration.py

@@ -87,6 +87,10 @@ def _show_admin_menu(self, user: NetworkUser) -> None:
                 text=Localization.get(user.locale, "account-approval"),
                 id="account_approval",
             ),
+            MenuItem(
+                text=Localization.get(user.locale, "reset-user-password"),
+                id="reset_user_password",
+            ),
             MenuItem(
                 text=Localization.get(user.locale, "ban-user"),
                 id="ban_user",
@@ -221,6 +225,48 @@ def _show_demote_admin_menu(self, user: NetworkUser) -> None:
         )
         self._user_states[user.username] = {"menu": "demote_admin_menu"}
 
+    def _show_reset_password_user_menu(self, user: NetworkUser) -> None:
+        """Show reset password menu with users admins may reset."""
+        resettable_users = self._db.get_non_admin_users(exclude_banned=True)
+
+        if not resettable_users:
+            user.speak_l("no-users-to-reset-password", buffer="misc")
+            self._show_admin_menu(user)
+            return
+
+        items = []
+        for resettable_user in resettable_users:
+            items.append(
+                MenuItem(
+                    text=resettable_user.username,
+                    id=f"reset_password_{resettable_user.username}",
+                )
+            )
+        items.append(MenuItem(text=Localization.get(user.locale, "back"), id="back"))
+
+        user.show_menu(
+            "reset_password_user_menu",
+            items,
+            multiletter=True,
+            escape_behavior=EscapeBehavior.SELECT_LAST,
+        )
+        self._user_states[user.username] = {"menu": "reset_password_user_menu"}
+
+    def _show_reset_password_editbox(self, user: NetworkUser, target_username: str) -> None:
+        """Show editbox for entering a replacement password."""
+        prompt = Localization.get(user.locale, "reset-user-password-prompt", player=target_username)
+        user.show_editbox(
+            "reset_user_password",
+            prompt,
+            default_value="",
+            multiline=False,
+            read_only=False,
+        )
+        self._user_states[user.username] = {
+            "menu": "reset_password_editbox",
+            "target_username": target_username,
+        }
+
     def _show_promote_confirm_menu(self, user: NetworkUser, target_username: str) -> None:
         """Show confirmation menu for promoting a user to admin."""
         question = Localization.get(user.locale, "confirm-promote", player=target_username)
@@ -468,6 +514,8 @@ async def _handle_admin_menu_selection(self, user: NetworkUser, selection_id: st
         """Handle admin menu selection."""
         if selection_id == "account_approval":
             self._show_account_approval_menu(user)
+        elif selection_id == "reset_user_password":
+            self._show_reset_password_user_menu(user)
         elif selection_id == "promote_admin":
             self._show_promote_admin_menu(user)
         elif selection_id == "demote_admin":
@@ -552,6 +600,40 @@ async def _handle_demote_admin_selection(self, user: NetworkUser, selection_id:
             target_username = selection_id[7:]  # Remove "demote_" prefix
             self._show_demote_confirm_menu(user, target_username)
 
+    async def _handle_reset_password_user_selection(
+        self, user: NetworkUser, selection_id: str
+    ) -> None:
+        """Handle reset password user menu selection."""
+        if selection_id == "back":
+            self._show_admin_menu(user)
+        elif selection_id.startswith("reset_password_"):
+            target_username = selection_id[15:]
+            self._show_reset_password_editbox(user, target_username)
+
+    async def _handle_reset_password_editbox(
+        self, admin: NetworkUser, text: str, state: dict
+    ) -> None:
+        """Handle replacement password submission."""
+        target_username = state.get("target_username")
+        if not target_username:
+            self._show_reset_password_user_menu(admin)
+            return
+
+        password = text or ""
+        min_length = self._password_min_length
+        max_length = self._password_max_length
+        if not (min_length <= len(password) <= max_length):
+            admin.speak_l(
+                "credential-password-length",
+                min=min_length,
+                max=max_length,
+                buffer="activity",
+            )
+            self._show_reset_password_editbox(admin, target_username)
+            return
+
+        await self._reset_user_password(admin, target_username, password)
+
     async def _handle_promote_confirm_selection(
         self, user: NetworkUser, selection_id: str, state: dict
     ) -> None:
@@ -777,6 +859,40 @@ async def _approve_user(self, admin: NetworkUser, username: str) -> None:
 
         self._show_account_approval_menu(admin)
 
+    @require_admin
+    async def _reset_user_password(
+        self, admin: NetworkUser, username: str, new_password: str
+    ) -> None:
+        """Reset a user's password to an admin-provided temporary value."""
+        target_record = self._db.get_user(username)
+        if not target_record or target_record.trust_level.value >= TrustLevel.ADMIN.value:
+            _speak_activity(admin, "reset-user-password-unavailable", player=username)
+            self._show_reset_password_user_menu(admin)
+            return
+
+        if self._auth.reset_password(username, new_password):
+            _speak_activity(admin, "reset-user-password-done", player=username)
+            target_user = self._users.get(username)
+            if target_user:
+                target_user.speak_l("your-password-was-reset", buffer="activity")
+                for msg in target_user.get_queued_messages():
+                    await target_user.connection.send(msg)
+                await target_user.connection.send(
+                    {
+                        "type": "disconnect",
+                        "reconnect": False,
+                        "show_message": True,
+                        "return_to_login": True,
+                        "message": Localization.get(
+                            target_user.locale, "your-password-was-reset"
+                        ),
+                    }
+                )
+        else:
+            _speak_activity(admin, "reset-user-password-unavailable", player=username)
+
+        self._show_reset_password_user_menu(admin)
+
     @require_admin
     async def _decline_user(self, admin: NetworkUser, username: str, reason: str = "") -> None:
         """Decline and delete a pending user account."""

server/core/server.py

@@ -2281,6 +2281,10 @@ async def _dispatch_menu_selection(
             ),
             "promote_admin_menu": (self._handle_promote_admin_selection, (user, selection_id)),
             "demote_admin_menu": (self._handle_demote_admin_selection, (user, selection_id)),
+            "reset_password_user_menu": (
+                self._handle_reset_password_user_selection,
+                (user, selection_id),
+            ),
             "promote_confirm_menu": (
                 self._handle_promote_confirm_selection,
                 (user, selection_id, state),
@@ -4185,6 +4189,11 @@ async def _handle_editbox(self, client: ClientConnection, packet: dict) -> None:
             await self._handle_unban_reason_editbox(user, text, state)
             return
 
+        if current_menu == "reset_password_editbox":
+            text = packet.get("text", "")
+            await self._handle_reset_password_editbox(user, text, state)
+            return
+
         # Forward to game if user is in a table
         table = self._tables.find_user_table(username)
         if table and table.game:

server/locales/en/main.ftl

@@ -348,6 +348,14 @@ account-action-empty-reason = No reason given.
 account-request = account request
 account-action = account action taken
 
+# Password recovery
+reset-user-password = Reset User Password
+no-users-to-reset-password = No users available for password reset.
+reset-user-password-prompt = Enter a new temporary password for { $player } (or press Escape to cancel):
+reset-user-password-done = { $player }'s password has been reset.
+reset-user-password-unavailable = { $player } is not available for password reset.
+your-password-was-reset = Your password was reset by an administrator. Please log in with the new password.
+
 # Admin promotion/demotion
 promote-admin = Promote Admin
 demote-admin = Demote Admin

server/persistence/database.py

@@ -446,6 +446,16 @@ def revoke_refresh_token(
         )
         self._conn.commit()
 
+    def revoke_user_refresh_tokens(self, username: str, revoked_at: int) -> None:
+        """Revoke all active refresh tokens for a user."""
+        cursor = self._conn.cursor()
+        cursor.execute(
+            "UPDATE refresh_tokens SET revoked_at = ? "
+            "WHERE lower(username) = lower(?) AND revoked_at IS NULL",
+            (revoked_at, username),
+        )
+        self._conn.commit()
+
     def get_user_count(self) -> int:
         """Get the total number of users in the database."""
         cursor = self._conn.cursor()

server/tests/test_administration.py

@@ -47,7 +47,7 @@ def __init__(self):
     def get_pending_users(self):
         return [SimpleNamespace(username=name) for name in self.pending_users]
 
-    def get_non_admin_users(self):
+    def get_non_admin_users(self, exclude_banned=True):
         return [SimpleNamespace(username=name) for name in self.non_admin_users]
 
     def get_admin_users(self, include_server_owner=True):
@@ -56,6 +56,13 @@ def get_admin_users(self, include_server_owner=True):
             return users
         return users
 
+    def get_user(self, username):
+        if username in self.non_admin_users:
+            return SimpleNamespace(username=username, trust_level=TrustLevel.USER)
+        if username in self.admin_users:
+            return SimpleNamespace(username=username, trust_level=TrustLevel.ADMIN)
+        return None
+
 
 class AdminHost(AdministrationMixin):
     def __init__(self, db=None):
@@ -136,13 +143,20 @@ def test_show_admin_menu_includes_owner_actions():
 
     host._show_admin_menu(admin_user)
     admin_ids = _get_menu_ids(admin_user)
-    assert admin_ids == ["account_approval", "ban_user", "unban_user", "back"]
+    assert admin_ids == [
+        "account_approval",
+        "reset_user_password",
+        "ban_user",
+        "unban_user",
+        "back",
+    ]
     assert host._user_states["admin"]["menu"] == "admin_menu"
 
     host._show_admin_menu(owner_user)
     owner_ids = _get_menu_ids(owner_user)
     assert owner_ids == [
         "account_approval",
+        "reset_user_password",
         "ban_user",
         "unban_user",
         "promote_admin",
@@ -218,6 +232,22 @@ def test_show_demote_admin_menu_filters_self_and_empty():
     assert _get_menu_ids(owner_user) == ["demote_eve", "back"]
 
 
+def test_show_reset_password_user_menu_handles_empty_and_entries():
+    db = DummyDB()
+    host = AdminHost(db=db)
+    admin_user = DummyUser("admin", TrustLevel.ADMIN)
+
+    host._show_reset_password_user_menu(admin_user)
+    assert admin_user.spoken[-1][0] == "no-users-to-reset-password"
+    assert host._user_states["admin"]["menu"] == "admin_menu"
+
+    db.non_admin_users = ["alice"]
+    admin_user.spoken.clear()
+    host._show_reset_password_user_menu(admin_user)
+    assert admin_user.menus[-1]["menu_id"] == "reset_password_user_menu"
+    assert _get_menu_ids(admin_user) == ["reset_password_alice", "back"]
+
+
 @pytest.mark.asyncio
 async def test_handle_account_approval_selection_routes(monkeypatch):
     host = AdminHost()
@@ -237,6 +267,54 @@ async def test_handle_account_approval_selection_routes(monkeypatch):
     assert calls == [("admin", "admin"), ("pending", "alice")]
 
 
+@pytest.mark.asyncio
+async def test_handle_reset_password_user_selection_routes():
+    host = AdminHost()
+    admin_user = DummyUser("admin", TrustLevel.ADMIN)
+    calls = []
+
+    host._show_admin_menu = types.MethodType(
+        lambda self, user: calls.append(("admin", user.username)), host
+    )
+    host._show_reset_password_editbox = types.MethodType(
+        lambda self, user, target: calls.append(("editbox", target)), host
+    )
+
+    await host._handle_reset_password_user_selection(admin_user, "back")
+    await host._handle_reset_password_user_selection(admin_user, "reset_password_alice")
+
+    assert calls == [("admin", "admin"), ("editbox", "alice")]
+
+
+@pytest.mark.asyncio
+async def test_reset_user_password_updates_auth_and_disconnects_online_user():
+    db = DummyDB()
+    db.non_admin_users = ["alice"]
+    host = AdminHost(db=db)
+    admin_user = DummyUser("admin", TrustLevel.ADMIN)
+    target_user = DummyUser("alice", TrustLevel.USER)
+    sent = []
+
+    async def send(payload):
+        sent.append(payload)
+
+    target_user.connection = SimpleNamespace(send=send)
+    target_user.get_queued_messages = lambda: []
+    host._users = {"alice": target_user}
+    calls = []
+    host._auth = SimpleNamespace(
+        reset_password=lambda username, password: calls.append((username, password)) or True
+    )
+
+    await host._reset_user_password(admin_user, "alice", "new-secret")
+
+    assert calls == [("alice", "new-secret")]
+    assert admin_user.spoken[-1][0] == "reset-user-password-done"
+    assert target_user.spoken[-1][0] == "your-password-was-reset"
+    assert sent[-1]["type"] == "disconnect"
+    assert sent[-1]["return_to_login"] is True
+
+
 @pytest.mark.asyncio
 async def test_handle_pending_user_actions_selection_paths(monkeypatch):
     host = AdminHost()

server/tests/test_database.py

@@ -140,6 +140,18 @@ def test_update_user_preferences_and_locale(db):
     assert row[1] == "pl"
 
 
+def test_revoke_user_refresh_tokens_revokes_only_target(db):
+    db.store_refresh_token("alice", "tok-a1", 100, 1)
+    db.store_refresh_token("Alice", "tok-a2", 100, 1)
+    db.store_refresh_token("bob", "tok-b1", 100, 1)
+
+    db.revoke_user_refresh_tokens("alice", 50)
+
+    assert db.get_refresh_token("tok-a1")["revoked_at"] == 50
+    assert db.get_refresh_token("tok-a2")["revoked_at"] == 50
+    assert db.get_refresh_token("tok-b1")["revoked_at"] is None
+
+
 def test_fluent_languages_default_empty(db):
     user = db.create_user("alice", "hash", approved=True)
     assert user.fluent_languages == []

server/tests/test_integration.py

@@ -166,6 +166,19 @@ def test_session_management(self):
         self.auth.invalidate_session(token)
         assert self.auth.validate_session(token) is None
 
+    def test_reset_password_invalidates_existing_sessions_and_refresh_tokens(self):
+        """Resetting a password revokes old credentials and accepts the new password."""
+        self.auth.register("resetuser", "oldpass", approved=True)
+        session_token, _ = self.auth.create_session("resetuser", 60)
+        refresh_token, _ = self.auth.create_refresh_token("resetuser", 60)
+
+        assert self.auth.reset_password("resetuser", "newpass") is True
+
+        assert self.auth.validate_session(session_token) is None
+        assert self.auth.refresh_session(refresh_token, 60, 60) is None
+        assert self.auth.authenticate("resetuser", "oldpass") == AuthResult.WRONG_PASSWORD
+        assert self.auth.authenticate("resetuser", "newpass") == AuthResult.SUCCESS
+
 
 class TestTableManagerIntegration:
     """Test table manager operations."""