fix: close refresh token session binding bypass

Add require_session parameter to validate_token to enforce session_id
matching for refresh operations. Attacker with stolen refresh token can
no longer bypass session binding by omitting session_id from request.

Also adds backend guard against deleting builtin admin user (was only
blocked at Flask blueprint layer), and removes unused _ALL_RW variable.

daemon/handlers/auth.py

@@ -44,7 +44,6 @@ from lib.auth import (
     validate_token,
 )
 from lib.auth_users import (
-    ALL_SUBSYSTEMS,
     create_user,
     delete_user,
     get_user,
@@ -70,8 +69,6 @@ from lib.webauthn import (
 
 logger = logging.getLogger(__name__)
 
-_ALL_RW = {sub: "rw" for sub in ALL_SUBSYSTEMS}
-
 
 def _clear_refresh_token_after_rotation(username: str) -> None:
     """Remove the user's entry from refresh_tokens after a successful refresh rotation."""
@@ -176,7 +173,12 @@ def auth_refresh(_request: Any, body: Any) -> dict[str, Any]:
         raise ValueError("refresh_token is required")
 
     session_id = body.get("session_id")
-    payload = validate_token(refresh_token, token_type="refresh", session_id=session_id)
+    if not session_id:
+        raise ValueError("session_id is required")
+
+    payload = validate_token(
+        refresh_token, token_type="refresh", session_id=session_id, require_session=True
+    )
     if payload is None:
         raise ValueError("Invalid or expired refresh token")
 

lib/auth.py

@@ -289,6 +289,7 @@ def validate_token(
     token_string: str,
     token_type: str = "access",
     session_id: str | None = None,
+    require_session: bool = False,
 ) -> dict[str, Any] | None:
     """Validate a JWT token and check it against the blacklist.
 
@@ -298,8 +299,12 @@ def validate_token(
         session_id: Must match the ``session_id`` claim in the token payload.
             When provided, enforces session binding to prevent a stolen token
             from being usable without the originating session. When ``None``,
-            the check is skipped (used by WebSocket auth which cannot carry
+            the check is skipped unless ``require_session`` is True.
-            the session ID header).
+        require_session: If True and the token payload contains a ``session_id``,
+            the request must provide a matching ``session_id``. Used by the
+            refresh handler to prevent session binding bypass. When False
+            (default), omitting ``session_id`` is acceptable even if the token
+            contains one.
 
     Returns:
         Payload dict including permissions, or None if invalid/blacklisted.
@@ -309,7 +314,11 @@ def validate_token(
         return None
     if payload.get("type") != token_type:
         return None
-    if session_id is not None and session_id != payload.get("session_id"):
+    token_session_id = payload.get("session_id")
+    if require_session and token_session_id is not None:
+        if session_id is None or session_id != token_session_id:
+            return None
+    elif session_id is not None and session_id != token_session_id:
         return None
 
     jti = payload.get("jti")

lib/auth_users.py

@@ -284,8 +284,11 @@ def delete_user(username: str) -> bool:
         True if the user was deleted.
 
     Raises:
-        ValueError: If the user does not exist.
+        ValueError: If the user does not exist or is the builtin admin.
     """
+    if username == BUILTIN_ADMIN_USERNAME:
+        raise ValueError("Cannot delete builtin admin user")
+
     user = find_user(username)
     if user is None:
         raise ValueError(f"User {username!r} not found")

tests/test_auth.py

@@ -827,6 +827,49 @@ class TestTokenValidationEdgeCases:
         payload = validate_token(token, "refresh", session_id="wrongsession")
         assert payload is None
 
+    def test_require_session_accepts_matching_session(self) -> None:
+        """require_session=True validates when matching session_id provided."""
+        create_user("reqsess", "password123")
+        token = generate_refresh_token("reqsess", session_id="mytoken-session")
+        payload = validate_token(
+            token,
+            "refresh",
+            session_id="mytoken-session",
+            require_session=True,
+        )
+        assert payload is not None
+
+    def test_require_session_rejects_when_session_id_omitted(self) -> None:
+        """require_session=True rejects when token has session_id but request passes None."""
+        create_user("reqsess2", "password123")
+        token = generate_refresh_token("reqsess2", session_id="mytoken-session")
+        payload = validate_token(
+            token,
+            "refresh",
+            session_id=None,
+            require_session=True,
+        )
+        assert payload is None
+
+    def test_require_session_rejects_wrong_session(self) -> None:
+        """require_session=True rejects when session_id doesn't match."""
+        create_user("reqsess3", "password123")
+        token = generate_refresh_token("reqsess3", session_id="mytoken-session")
+        payload = validate_token(
+            token,
+            "refresh",
+            session_id="wrong-session",
+            require_session=True,
+        )
+        assert payload is None
+
+    def test_require_session_with_no_session_in_token(self) -> None:
+        """require_session=True with a token that has no session_id claim works normally."""
+        create_user("reqsess4", "password123")
+        token = generate_refresh_token("reqsess4")
+        payload = validate_token(token, "refresh", require_session=True)
+        assert payload is not None
+
     def test_refresh_token_generate_tokens_includes_session_id(self) -> None:
         """generate_tokens produces refresh token with session_id."""
         create_user("tokuser", "password123")
@@ -966,6 +1009,11 @@ class TestMultiUserAdmin:
         creds = list_credentials("wgscduser")
         assert len(creds) == 0
 
+    def test_delete_builtin_admin_rejected(self) -> None:
+        """Deleting the builtin admin user is rejected."""
+        with pytest.raises(ValueError, match="Cannot delete builtin admin"):
+            delete_user("admin")
+
     def test_self_deletion_prevention(self) -> None:
         """Self-deletion is prevented at the Flask blueprint layer.