Count a wrong current password as a failed login, so changing a password cannot be used to guess it

app/api/auth.py

@@ -32,15 +32,6 @@ class TokenResponse(BaseModel):
 
 # -- Local auth ------------------------------------------------------------------
 
-def _locked(state: dict) -> HTTPException:
-    # 423 rather than 401, so a client can tell "locked" from "wrong password".
-    if state["permanent"]:
-        detail = "This account is locked after repeated failed logins. Contact an administrator to unlock it."
-    else:
-        detail = f"This account is locked after repeated failed logins. Try again after {state['until']} UTC."
-    return HTTPException(status_code=status.HTTP_423_LOCKED, detail=detail)
-
-
 @router.post("/login", response_model=TokenResponse)
 async def login(body: LoginRequest, request: Request, response: Response, db: aiosqlite.Connection = Depends(get_db)):
     async with db.execute(
@@ -56,12 +47,12 @@ async def login(body: LoginRequest, request: Request, response: Response, db: ai
     # guessed at and a correct password does not get past the lock.
     state = lockout.describe(user)
     if state["locked"]:
-        raise _locked(state)
+        raise lockout.locked_exception(state)
 
     if not user["hashed_password"] or not verify_password(body.password, user["hashed_password"]):
         state = await lockout.record_failure(db, user["id"])
         if state["locked"]:
-            raise _locked(state)
+            raise lockout.locked_exception(state)
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
 
     await lockout.record_success(db, user["id"])

app/api/users.py

@@ -182,10 +182,24 @@ async def delete_user(user_id: int, user: AdminUser, db: aiosqlite.Connection =
 async def change_my_password(body: ChangePasswordRequest, user: CurrentUser, db: aiosqlite.Connection = Depends(get_db)):
     if user.get("_via_suite"):
         raise HTTPException(status_code=400, detail="Password managed by pktHub for suite-authenticated sessions")
-    async with db.execute("SELECT hashed_password FROM users WHERE id = ?", (user["id"],)) as cur:
+    async with db.execute(
+        f"SELECT hashed_password, {lockout.LOCK_COLUMNS} FROM users WHERE id = ?", (user["id"],)
+    ) as cur:
         row = await cur.fetchone()
+    # A wrong current password counts as a failed login, exactly as at the login
+    # form: a signed-in session must not be a free way to guess it. A locked
+    # account is refused before the password is looked at.
+    if row:
+        state = lockout.describe(row)
+        if state["locked"]:
+            raise lockout.locked_exception(state)
     if not row or not row["hashed_password"] or not verify_password(body.current_password, row["hashed_password"]):
+        if row:
+            state = await lockout.record_failure(db, user["id"])
+            if state["locked"]:
+                raise lockout.locked_exception(state)
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Current password is incorrect")
+    await lockout.record_success(db, user["id"])
     await db.execute(
         "UPDATE users SET hashed_password = ? WHERE id = ?",
         (hash_password(body.new_password), user["id"]),

app/auth/lockout.py

@@ -10,13 +10,18 @@
 
 While an account is locked the password is not checked at all, so a locked
 account cannot be guessed at.
+
+The count covers every place a password is checked against an account: the
+login, and the current password asked for when changing one, so a signed-in
+session cannot be used to guess it.
 """
 from __future__ import annotations
 
 import json
 import logging
 
 import aiosqlite
+from fastapi import HTTPException, status
 
 log = logging.getLogger("pktwifi.auth")
 
@@ -42,6 +47,15 @@ def describe(row) -> dict:
     }
 
 
+def locked_exception(state: dict) -> HTTPException:
+    # 423 rather than 401, so a client can tell "locked" from "wrong password".
+    if state["permanent"]:
+        detail = "This account is locked after repeated failed logins. Contact an administrator to unlock it."
+    else:
+        detail = f"This account is locked after repeated failed logins. Try again after {state['until']} UTC."
+    return HTTPException(status_code=status.HTTP_423_LOCKED, detail=detail)
+
+
 async def _int_setting(db: aiosqlite.Connection, key: str, default: int, ceiling: int) -> int:
     async with db.execute("SELECT value FROM settings WHERE key = ?", (key,)) as cur:
         row = await cur.fetchone()

docs/ADMIN_GUIDE.md

@@ -27,7 +27,7 @@ Prompts for install directory and port, then handles the venv, `config.yaml` + s
 
 All roles can view every page; analysts and admins can acknowledge/resolve alerts; only admins reach Settings. Manage accounts at Settings → Security → Users — create/edit/deactivate/delete, reset password, and mark one active admin as the **default admin** (star icon): if every auth method is ever disabled, the app auto-signs everyone in as that account instead of dead-ending.
 
-**Failed-login lockout.** A local account is locked for 30 minutes after a set number of failed logins in a row (Settings → Security → Auth → *Failed logins before lockout*, default 3). Failures never expire; only a successful login resets the count. If it then fails that many times again it stays locked until an admin clicks the unlock icon beside it on the Users tab. While locked, even the right password is refused. A successful login clears the failure count and any earlier lockout. If the only admin is locked, unlock it from the server, in the install directory with the app's own Python:
+**Failed-login lockout.** A local account is locked for 30 minutes after a set number of failed logins in a row (Settings → Security → Auth → *Failed logins before lockout*, default 3). Failures never expire; only a successful login resets the count. If it then fails that many times again it stays locked until an admin clicks the unlock icon beside it on the Users tab. While locked, even the right password is refused. A wrong current password when changing a password counts as a failed login too, so a signed-in session cannot be used to guess it. A successful login clears the failure count and any earlier lockout. If the only admin is locked, unlock it from the server, in the install directory with the app's own Python:
 
 ```bash
 python3 scripts/unlock_user.py <username>

frontend/src/api/client.ts

@@ -47,6 +47,11 @@ async function request<T>(path: string, options: RequestInit = {}): Promise<T> {
   const res = await fetch(`/api${path}`, { ...options, headers })
 
   if (res.status === 401) {
+    // A wrong current password on change-password is not an expired session. It
+    // must not be retried either: every attempt counts toward the lockout, and a
+    // retry would count each one twice.
+    const detail = (await res.clone().json().catch(() => null))?.detail
+    if (detail === 'Current password is incorrect') throw new Error(detail)
     const refreshed = await tryRefresh()
     if (refreshed) {
       headers['Authorization'] = `Bearer ${_accessToken}`

tests/test_login_lockout.py

@@ -15,6 +15,9 @@
     never turns the next one permanent,
   * an admin can unlock a user, and so can the host-side script,
   * the limit is a setting, with 3 as the default when it is missing or junk,
+  * a wrong current password when changing a password counts as a failed
+    login and locks the account the same way, so a signed-in session cannot be
+    used to guess it,
   * an unknown username is refused without anything being counted,
   * old client events are purged on their own retention window.
 """
@@ -163,6 +166,37 @@ def fail(user: str, n: int) -> list[int]:
         sql("DELETE FROM settings WHERE key = 'login_max_failed_attempts'")
         check("so does no value at all", fail("gina", 3)[-1] == 423)
 
+        print("\n── changing a password counts too ──")
+        def headers_for(name: str, pw: str = GOOD) -> dict:
+            return {"Authorization": f"Bearer {login(name, pw).json()['access_token']}"}
+
+        def change(h: dict, current: str, new: str = "a-new-password"):
+            return client.post("/api/users/me/change-password",
+                               json={"current_password": current, "new_password": new}, headers=h)
+
+        make_user("nina")
+        nh = headers_for("nina")
+        codes = [change(nh, BAD).status_code for _ in range(3)]
+        check("wrong current passwords are refused, and the third locks", codes == [401, 401, 423], str(codes))
+        check("the right current password is then refused too", change(nh, GOOD).status_code == 423)
+        check("the same lock applies to signing in", login("nina", GOOD).status_code == 423)
+        check("a session already open keeps working", client.get("/api/users/me", headers=nh).status_code == 200)
+        lapse("nina")
+        codes = [change(nh, BAD).status_code for _ in range(3)]
+        check("a second round locks it permanently", codes[-1] == 423 and row("nina")["is_locked"] == 1, str(codes))
+        check("an admin unlock lifts it", client.post(f"/api/users/{row('nina')['id']}/unlock", headers=h).status_code == 204)
+        check("and the password can then be changed", change(nh, GOOD, "brand-new-password").status_code == 200)
+        check("the new password signs in", login("nina", "brand-new-password").status_code == 200)
+        check("counters are clear", row("nina")["failed_login_count"] == 0 and row("nina")["lockout_count"] == 0)
+
+        make_user("omar")
+        oh = headers_for("omar")
+        change(oh, BAD)
+        check("a failure is counted", row("omar")["failed_login_count"] == 1)
+        check("a successful change clears it", change(oh, GOOD).status_code == 200 and row("omar")["failed_login_count"] == 0)
+        check("sign-in is required", client.post("/api/users/me/change-password",
+                                                 json={"current_password": GOOD, "new_password": "x" * 8}).status_code == 401)
+
         print("\n── accounts that do not exist ──")
         before = sql("SELECT COUNT(*) AS n FROM users")[0]["n"]
         check("an unknown username is refused plainly", fail("nobody", 5) == [401] * 5)