fix(auth): isolate request state and harden token parsing

Author: alexandreteles

src/fastapi_paseto/_internal/auth.py

@@ -57,12 +57,22 @@ def __init__(
     ) -> None:
         """Capture request-scoped objects used by token extraction helpers."""
 
+        self._token: str | None = None
+        self._token_parts: list[str] = []
+        self._current_user: str | int | None = None
+        self._decoded_token: Token | None = None
         self._request_json = request_json
         self._response = response
-        if request is not None:
-            self._request = request
-        if websocket is not None:
-            self._websocket = websocket
+        self._request = request
+        self._websocket = websocket
+
+    def _reset_runtime_state(self) -> None:
+        """Clear request-scoped authentication state."""
+
+        self._token = None
+        self._token_parts = []
+        self._current_user = None
+        self._decoded_token = None
 
     def _get_paseto_from_json(
         self,
@@ -106,8 +116,10 @@ def _get_paseto_from_query(
     def _get_connection(self) -> Request | WebSocket:
         """Return the current request or websocket connection."""
 
-        if hasattr(self, "_websocket"):
+        if self._websocket is not None:
             return self._websocket
+        if self._request is None:  # pragma: no cover
+            raise RuntimeError("Request or websocket connection is required")
         return self._request
 
     def _get_connection_headers(self) -> Mapping[str, str]:
@@ -123,7 +135,7 @@ def _get_connection_query_params(self) -> Mapping[str, str]:
     def _is_websocket_connection(self) -> bool:
         """Return whether the current dependency context is a websocket."""
 
-        return hasattr(self, "_websocket")
+        return self._websocket is not None
 
     def _get_paseto_identifier(self) -> str:
         """Return a new unique token identifier."""
@@ -393,6 +405,7 @@ def paseto_required(
     ) -> None:
         """Validate the current request token against the endpoint requirements."""
 
+        self._reset_runtime_state()
         try:
             validate_required_token_flags(fresh=fresh, refresh_token=refresh_token)
             if token:

src/fastapi_paseto/_internal/request.py

@@ -7,6 +7,14 @@
 from fastapi_paseto.exceptions import InvalidHeaderError
 
 
+def _build_token_error_message(token_key: str, token_type: str | None) -> str:
+    """Return the error message for a malformed token-bearing field."""
+
+    if token_type:
+        return f"Bad {token_key} header. Expected value '{token_type} <PASETO>'"
+    return f"Bad {token_key} header. Expected value '<PASETO>'"
+
+
 async def get_request_json(
     request: Request = None,
     websocket: WebSocket = None,
@@ -35,21 +43,27 @@ def extract_token_from_json(
         return None
 
     token = request_json[json_key]
+    if not isinstance(token, str):
+        raise InvalidHeaderError(
+            status_code=422,
+            message=_build_token_error_message(json_key, json_type),
+        )
+
     if not json_type:
         if not token:
             raise InvalidHeaderError(
                 status_code=422,
-                message=f"Bad {json_key} header. Excepted value 'Bearer <PASETO>'",
+                message=_build_token_error_message(json_key, json_type),
             )
         return token
 
-    token_prefix, token = token.split()
-    if not token or token_prefix != json_type:
+    parts = token.split()
+    if len(parts) != 2 or parts[0] != json_type or not parts[1]:
         raise InvalidHeaderError(
             status_code=422,
-            message=f"Bad {json_key} header. Expected value '{json_type} <PASETO>'",
+            message=_build_token_error_message(json_key, json_type),
         )
-    return token
+    return parts[1]
 
 
 def extract_token_from_header(
@@ -67,11 +81,11 @@ def extract_token_from_header(
         if len(parts) != 1:
             raise InvalidHeaderError(
                 status_code=422,
-                message=f"Bad {header_name} header. Excepted value '<PASETO>'",
+                message=f"Bad {header_name} header. Expected value '<PASETO>'",
             )
         return parts[0]
 
-    if not parts[0].__contains__(header_type) or len(parts) != 2:
+    if len(parts) != 2 or parts[0].lower() != header_type.lower() or not parts[1]:
         raise InvalidHeaderError(
             status_code=422,
             message=f"Bad {header_name} header. Expected value '{header_type} <PASETO>'",

src/fastapi_paseto/auth_config.py

@@ -1,11 +1,10 @@
-"""Shared mutable configuration state for ``AuthPASETO``."""
+"""Shared configuration state for ``AuthPASETO``."""
 
 from collections.abc import Callable, Mapping
 from datetime import timedelta
 
 from pydantic import ValidationError
 from pydantic_settings import BaseSettings
-from pyseto import Token
 
 from fastapi_paseto.config import LoadConfig
 
@@ -37,13 +36,8 @@
 
 class AuthConfig:
     """Hold class-level configuration shared by all ``AuthPASETO`` instances."""
-
-    _token: str | None = None
-    _token_parts: list[str] = []
     _token_location: list[str] | tuple[str, ...] = ("headers",)
     _websocket_token_location: list[str] | tuple[str, ...] = ("headers",)
-    _current_user: str | int | None = None
-    _decoded_token: Token | None = None
 
     _secret_key: str | None = None
     _public_key: str | None = None

tests/conftest.py

@@ -7,12 +7,8 @@
 
 
 _AUTHPASETO_DEFAULTS: dict[str, object] = {
-    "_token": None,
-    "_token_parts": [],
     "_token_location": ("headers",),
     "_websocket_token_location": ("headers",),
-    "_current_user": None,
-    "_decoded_token": None,
     "_secret_key": None,
     "_public_key": None,
     "_private_key": None,

tests/test_config.py

@@ -27,12 +27,14 @@ def protected(Authorize: AuthPASETO = Depends()):
 
 
 def test_default_config():
-    assert AuthPASETO._token is None
+    authorize = AuthPASETO()
+
+    assert authorize._token is None
     assert AuthPASETO._token_location == ("headers",)
     assert AuthPASETO._websocket_token_location == ("headers",)
-    assert AuthPASETO._current_user is None
-    assert AuthPASETO._decoded_token is None
-    assert AuthPASETO._secret_key is None
+    assert authorize._current_user is None
+    assert authorize._decoded_token is None
+    assert AuthPASETO._secret_key is None
     assert AuthPASETO._public_key is None
     assert AuthPASETO._private_key is None
     assert AuthPASETO._purpose == "local"

tests/test_headers.py

@@ -40,11 +40,11 @@ def test_header_without_paseto(client):
     }
 
 
-def test_header_without_bearer(client):
-    response = client.get("/protected", headers={"Authorization": "Test asd"})
-    assert response.status_code == 422
-    assert response.json() == {
-        "detail": "Bad Authorization header. Expected value 'Bearer <PASETO>'"
+def test_header_without_bearer(client):
+    response = client.get("/protected", headers={"Authorization": "Test asd"})
+    assert response.status_code == 422
+    assert response.json() == {
+        "detail": "Bad Authorization header. Expected value 'Bearer <PASETO>'"
     }
 
     response = client.get("/protected", headers={"Authorization": "Test "})
@@ -56,10 +56,30 @@ def test_header_without_bearer(client):
 
 def test_header_invalid_paseto(client):
     response = client.get("/protected", headers={"Authorization": "Bearer asd"})
-    assert response.status_code == 422
-    assert response.json() == {"detail": "Invalid PASETO format"}
-
-
+    assert response.status_code == 422
+    assert response.json() == {"detail": "Invalid PASETO format"}
+
+
+def test_header_requires_exact_bearer_scheme(client, Authorize):
+    token = Authorize.create_access_token(subject="test")
+
+    invalid_scheme_response = client.get(
+        "/protected",
+        headers={"Authorization": f"NotBearer {token}"},
+    )
+    assert invalid_scheme_response.status_code == 422
+    assert invalid_scheme_response.json() == {
+        "detail": "Bad Authorization header. Expected value 'Bearer <PASETO>'"
+    }
+
+    lowercase_scheme_response = client.get(
+        "/protected",
+        headers={"Authorization": f"bearer {token}"},
+    )
+    assert lowercase_scheme_response.status_code == 200
+    assert lowercase_scheme_response.json() == {"hello": "world"}
+
+
 def test_valid_header(client, Authorize):
     token = Authorize.create_access_token(subject="test")
     response = client.get("/protected", headers={"Authorization": f"Bearer {token}"})

tests/test_json.py

@@ -60,6 +60,35 @@ def get_settings_one():
     assert response.json() == {"detail": "Invalid PASETO format"}
 
 
+@pytest.mark.parametrize(
+    ("payload", "detail"),
+    [
+        (
+            {"access_token": 123},
+            "Bad access_token header. Expected value 'Bearer <PASETO>'",
+        ),
+        (
+            {"access_token": ""},
+            "Bad access_token header. Expected value 'Bearer <PASETO>'",
+        ),
+        (
+            {"access_token": "Bearer"},
+            "Bad access_token header. Expected value 'Bearer <PASETO>'",
+        ),
+        (
+            {"access_token": "Bearer token extra"},
+            "Bad access_token header. Expected value 'Bearer <PASETO>'",
+        ),
+    ],
+)
+def test_malformed_json_token_returns_auth_error(client, configure_auth, payload, detail):
+    configure_auth(authpaseto_json_type="Bearer")
+
+    response = client.request("GET", "/protected_json", json=payload)
+    assert response.status_code == 422
+    assert response.json() == {"detail": detail}
+
+
 def test_wrong_location(client, access_token):
     class SettingsOne(BaseSettings):
         authpaseto_token_location: list[str] = ["headers"]

tests/test_url_protected.py

@@ -79,17 +79,52 @@ def test_paseto_required(client, Authorize):
     assert response.json() == {"hello": "world"}
 
 
-def test_paseto_optional(client, Authorize):
-    url = "/paseto-optional"
-    # if header not define return anonym user
-    response = client.get(url)
-    assert response.status_code == 200
+def test_paseto_optional(client, Authorize):
+    url = "/paseto-optional"
+    # if header not define return anonym user
+    response = client.get(url)
+    assert response.status_code == 200
     assert response.json() == {"hello": "anonym"}
 
     token = Authorize.create_access_token(subject="test")
-    response = client.get(url, headers={"Authorization": f"Bearer {token}"})
-    assert response.status_code == 200
-    assert response.json() == {"hello": "world"}
+    response = client.get(url, headers={"Authorization": f"Bearer {token}"})
+    assert response.status_code == 200
+    assert response.json() == {"hello": "world"}
+
+
+def test_paseto_optional_does_not_leak_previous_subject(client, Authorize):
+    url = "/paseto-optional"
+    token = Authorize.create_access_token(subject="test")
+
+    authorized_response = client.get(
+        url,
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    assert authorized_response.status_code == 200
+    assert authorized_response.json() == {"hello": "world"}
+
+    anonymous_response = client.get(url)
+    assert anonymous_response.status_code == 200
+    assert anonymous_response.json() == {"hello": "anonym"}
+
+
+def test_paseto_optional_invalid_token_does_not_leak_previous_subject(client, Authorize):
+    url = "/paseto-optional"
+    token = Authorize.create_access_token(subject="test")
+
+    authorized_response = client.get(
+        url,
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    assert authorized_response.status_code == 200
+    assert authorized_response.json() == {"hello": "world"}
+
+    invalid_response = client.get(
+        url,
+        headers={"Authorization": "Bearer invalid"},
+    )
+    assert invalid_response.status_code == 200
+    assert invalid_response.json() == {"hello": "anonym"}
 
 
 def test_refresh_required(client, Authorize):

tests/test_websocket.py

@@ -203,6 +203,51 @@ def test_websocket_optional_without_token(make_client: Callable[..., TestClient]
     assert message == {"payload": None, "subject": None}
 
 
+def test_websocket_optional_does_not_leak_previous_subject(
+    make_client: Callable[..., TestClient],
+    Authorize: AuthPASETO,
+) -> None:
+    client = make_client()
+    token = Authorize.create_access_token(subject="test")
+
+    with client.websocket_connect(
+        "/required",
+        headers={"Authorization": f"Bearer {token}"},
+    ) as websocket:
+        message = websocket.receive_json()
+
+    assert message["subject"] == "test"
+
+    with client.websocket_connect("/optional") as websocket:
+        optional_message = websocket.receive_json()
+
+    assert optional_message == {"payload": None, "subject": None}
+
+
+def test_websocket_optional_invalid_token_does_not_leak_previous_subject(
+    make_client: Callable[..., TestClient],
+    Authorize: AuthPASETO,
+) -> None:
+    client = make_client()
+    token = Authorize.create_access_token(subject="test")
+
+    with client.websocket_connect(
+        "/required",
+        headers={"Authorization": f"Bearer {token}"},
+    ) as websocket:
+        message = websocket.receive_json()
+
+    assert message["subject"] == "test"
+
+    with client.websocket_connect(
+        "/optional",
+        headers={"Authorization": "Bearer invalid"},
+    ) as websocket:
+        optional_message = websocket.receive_json()
+
+    assert optional_message == {"payload": None, "subject": None}
+
+
 def test_websocket_refresh_required(
     make_client: Callable[..., TestClient],
     Authorize: AuthPASETO,