diff --git a/daemon/handlers/auth.py b/daemon/handlers/auth.py index 5a0e27c..bd17da7 100644 --- a/daemon/handlers/auth.py +++ b/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") diff --git a/lib/auth.py b/lib/auth.py index f8f5067..9e685ab 100644 --- a/lib/auth.py +++ b/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 session ID header). + the check is skipped unless ``require_session`` is True. + 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") diff --git a/lib/auth_users.py b/lib/auth_users.py index f81b516..e846a13 100644 --- a/lib/auth_users.py +++ b/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") diff --git a/tests/test_auth.py b/tests/test_auth.py index 5472a22..3ba4b28 100644 --- a/tests/test_auth.py +++ b/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.