fix(auth): normalize invalid bearer token handling

Author: 818sushirolls

server/auth.py

@@ -1,67 +1,71 @@
-"""Lightweight auth helpers for future RBAC/JWT integration."""
-
-import os
-from datetime import datetime, timedelta, timezone
-from typing import Any
-
-import jwt
-
-ROLE_ADMIN = "admin"
-ROLE_USER = "user"
-ALLOWED_ROLES = {ROLE_ADMIN, ROLE_USER}
-
-
-def _jwt_secret() -> str:
-    return os.getenv("JWT_SECRET", "dev-jwt-secret")
-
-
-def _jwt_algorithm() -> str:
-    return os.getenv("JWT_ALGORITHM", "HS256")
-
-
-def create_access_token(
-    user_id: str,
-    role: str,
-    expires_hours: int = 24,
-) -> str:
-    """Create a JWT access token with basic role claims."""
-    if role not in ALLOWED_ROLES:
-        raise ValueError(f"Unsupported role: {role}")
-
-    now = datetime.now(timezone.utc)
-    payload = {
-        "sub": user_id,
-        "role": role,
-        "iat": now,
-        "exp": now + timedelta(hours=max(expires_hours, 1)),
-    }
-    return jwt.encode(payload, _jwt_secret(), algorithm=_jwt_algorithm())
-
-
-def decode_access_token(token: str) -> dict[str, Any]:
-    """Decode and validate a JWT access token."""
-    return jwt.decode(token, _jwt_secret(), algorithms=[_jwt_algorithm()])
-
-
-def extract_bearer_token(auth_header: str | None) -> str | None:
-    """Extract bearer token from an Authorization header string."""
-    if not auth_header or not auth_header.startswith("Bearer "):
-        return None
-    token = auth_header.removeprefix("Bearer ").strip()
-    return token or None
-
-
-def authenticate_request(
-    auth_header: str | None,
-    required_role: str | None = None,
-) -> dict[str, Any]:
-    """Validate a bearer token header and optionally enforce a role."""
-    token = extract_bearer_token(auth_header)
-    if token is None:
-        raise PermissionError("Missing or invalid bearer token")
-
-    payload = decode_access_token(token)
-    if required_role and payload.get("role") != required_role:
-        raise PermissionError(f"Role '{required_role}' is required")
-
-    return payload
+"""Lightweight auth helpers for future RBAC/JWT integration."""
+
+import os
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import jwt
+from jwt import InvalidTokenError
+
+ROLE_ADMIN = "admin"
+ROLE_USER = "user"
+ALLOWED_ROLES = {ROLE_ADMIN, ROLE_USER}
+
+
+def _jwt_secret() -> str:
+    return os.getenv("JWT_SECRET", "dev-jwt-secret")
+
+
+def _jwt_algorithm() -> str:
+    return os.getenv("JWT_ALGORITHM", "HS256")
+
+
+def create_access_token(
+    user_id: str,
+    role: str,
+    expires_hours: int = 24,
+) -> str:
+    """Create a JWT access token with basic role claims."""
+    if role not in ALLOWED_ROLES:
+        raise ValueError(f"Unsupported role: {role}")
+
+    now = datetime.now(timezone.utc)
+    payload = {
+        "sub": user_id,
+        "role": role,
+        "iat": now,
+        "exp": now + timedelta(hours=max(expires_hours, 1)),
+    }
+    return jwt.encode(payload, _jwt_secret(), algorithm=_jwt_algorithm())
+
+
+def decode_access_token(token: str) -> dict[str, Any]:
+    """Decode and validate a JWT access token."""
+    return jwt.decode(token, _jwt_secret(), algorithms=[_jwt_algorithm()])
+
+
+def extract_bearer_token(auth_header: str | None) -> str | None:
+    """Extract bearer token from an Authorization header string."""
+    if not auth_header or not auth_header.startswith("Bearer "):
+        return None
+    token = auth_header.removeprefix("Bearer ").strip()
+    return token or None
+
+
+def authenticate_request(
+    auth_header: str | None,
+    required_role: str | None = None,
+) -> dict[str, Any]:
+    """Validate a bearer token header and optionally enforce a role."""
+    token = extract_bearer_token(auth_header)
+    if token is None:
+        raise PermissionError("Missing or invalid bearer token")
+
+    try:
+        payload = decode_access_token(token)
+    except InvalidTokenError as exc:
+        raise PermissionError("Invalid or expired bearer token") from exc
+    if required_role and payload.get("role") != required_role:
+        raise PermissionError(f"Role '{required_role}' is required")
+
+    return payload

server/tests/test_auth.py

@@ -1,80 +1,88 @@
-from jwt import InvalidTokenError
-
-from server.auth import (
-    ROLE_ADMIN,
-    ROLE_USER,
-    authenticate_request,
-    create_access_token,
-    decode_access_token,
-    extract_bearer_token,
-)
-
-
-def test_create_and_decode_access_token_round_trip():
-    token = create_access_token("alice", ROLE_ADMIN, expires_hours=1)
-
-    payload = decode_access_token(token)
-
-    assert payload["sub"] == "alice"
-    assert payload["role"] == ROLE_ADMIN
-
-
-def test_create_access_token_rejects_unknown_role():
-    try:
-        create_access_token("alice", "superadmin")
-        assert False, "Expected ValueError for unsupported role"
-    except ValueError as exc:
-        assert "Unsupported role" in str(exc)
-
-
-def test_decode_access_token_rejects_invalid_token():
-    try:
-        decode_access_token("not-a-token")
-        assert False, "Expected InvalidTokenError"
-    except InvalidTokenError:
-        assert True
-
-
-def test_extract_bearer_token_valid_header():
-    token = extract_bearer_token("Bearer abc.def.ghi")
-    assert token == "abc.def.ghi"
-
-
-def test_extract_bearer_token_invalid_header():
-    assert extract_bearer_token(None) is None
-    assert extract_bearer_token("") is None
-    assert extract_bearer_token("Basic xyz") is None
-    assert extract_bearer_token("Bearer   ") is None
-
-
-def test_user_role_token_supported():
-    token = create_access_token("bob", ROLE_USER, expires_hours=1)
-    payload = decode_access_token(token)
-    assert payload["role"] == ROLE_USER
-
-
-def test_authenticate_request_accepts_valid_bearer_token():
-    token = create_access_token("alice", ROLE_ADMIN, expires_hours=1)
-
-    payload = authenticate_request(f"Bearer {token}")
-
-    assert payload["sub"] == "alice"
-    assert payload["role"] == ROLE_ADMIN
-
-
-def test_authenticate_request_rejects_missing_bearer_token():
-    try:
-        authenticate_request(None)
-        assert False, "Expected PermissionError"
-    except PermissionError as exc:
-        assert "Missing or invalid bearer token" in str(exc)
-
-
-def test_authenticate_request_enforces_required_role():
-    token = create_access_token("bob", ROLE_USER, expires_hours=1)
-
-    try:
-        authenticate_request(f"Bearer {token}", required_role=ROLE_ADMIN)
-        assert False, "Expected PermissionError"
-    except PermissionError as exc:
-        assert "Role 'admin' is required" in str(exc)
+from jwt import InvalidTokenError
+
+from server.auth import (
+    ROLE_ADMIN,
+    ROLE_USER,
+    authenticate_request,
+    create_access_token,
+    decode_access_token,
+    extract_bearer_token,
+)
+
+
+def test_create_and_decode_access_token_round_trip():
+    token = create_access_token("alice", ROLE_ADMIN, expires_hours=1)
+
+    payload = decode_access_token(token)
+
+    assert payload["sub"] == "alice"
+    assert payload["role"] == ROLE_ADMIN
+
+
+def test_create_access_token_rejects_unknown_role():
+    try:
+        create_access_token("alice", "superadmin")
+        assert False, "Expected ValueError for unsupported role"
+    except ValueError as exc:
+        assert "Unsupported role" in str(exc)
+
+
+def test_decode_access_token_rejects_invalid_token():
+    try:
+        decode_access_token("not-a-token")
+        assert False, "Expected InvalidTokenError"
+    except InvalidTokenError:
+        assert True
+
+
+def test_extract_bearer_token_valid_header():
+    token = extract_bearer_token("Bearer abc.def.ghi")
+    assert token == "abc.def.ghi"
+
+
+def test_extract_bearer_token_invalid_header():
+    assert extract_bearer_token(None) is None
+    assert extract_bearer_token("") is None
+    assert extract_bearer_token("Basic xyz") is None
+    assert extract_bearer_token("Bearer   ") is None
+
+
+def test_user_role_token_supported():
+    token = create_access_token("bob", ROLE_USER, expires_hours=1)
+    payload = decode_access_token(token)
+    assert payload["role"] == ROLE_USER
+
+
+def test_authenticate_request_accepts_valid_bearer_token():
+    token = create_access_token("alice", ROLE_ADMIN, expires_hours=1)
+
+    payload = authenticate_request(f"Bearer {token}")
+
+    assert payload["sub"] == "alice"
+    assert payload["role"] == ROLE_ADMIN
+
+
+def test_authenticate_request_rejects_missing_bearer_token():
+    try:
+        authenticate_request(None)
+        assert False, "Expected PermissionError"
+    except PermissionError as exc:
+        assert "Missing or invalid bearer token" in str(exc)
+
+
+def test_authenticate_request_enforces_required_role():
+    token = create_access_token("bob", ROLE_USER, expires_hours=1)
+
+    try:
+        authenticate_request(f"Bearer {token}", required_role=ROLE_ADMIN)
+        assert False, "Expected PermissionError"
+    except PermissionError as exc:
+        assert "Role 'admin' is required" in str(exc)
+
+
+def test_authenticate_request_rejects_invalid_bearer_token():
+    try:
+        authenticate_request("Bearer not-a-token")
+        assert False, "Expected PermissionError"
+    except PermissionError as exc:
+        assert "Invalid or expired bearer token" in str(exc)