Merge pull request #804 from seapagan/security/critical-token-and-cors

fix: enforce token type separation and harden CORS defaults

Author: seapagan

.env.example

@@ -45,7 +45,8 @@ ACCESS_TOKEN_EXPIRE_MINUTES=120
 
 # List of origins that can access this API, separated by a comma, eg:
 # CORS_ORIGINS=http://localhost,https://www.gnramsay.com
-# If you want all origins to access (the default), use * or comment out:
+# For public APIs using Bearer tokens, * is acceptable but will log a warning.
+# Use explicit origins if serving browser clients.
 CORS_ORIGINS=*
 
 # Email Settings - OPTIONAL for development, REQUIRED for production

SECURITY-REVIEW.md

@@ -4,6 +4,10 @@
 
 ### 1. Refresh Token Authentication Bypass
 
+> [!NOTE]
+> ✅ **Done**: Access tokens now include `typ="access"` and `get_jwt_user`
+> enforces it; refresh or typ-less tokens are rejected.
+
 **Location**: `app/managers/auth.py:478-544` (`get_jwt_user`)
 
 - **Issue**: Refresh tokens can be used as access tokens because access tokens
@@ -19,6 +23,10 @@
 
 ### 2. CORS Wildcard with Credentials - SEVERE
 
+> [!NOTE]
+> ✅ **Done**: CORS credentials are disabled for the API and startup now warns
+> when `CORS_ORIGINS=*` is used.
+
 **Location**: `app/main.py:169`, `app/config/settings.py:56`
 
 - **Issue**: Default CORS configuration allows ALL origins (`cors_origins="*"`)

app/main.py

@@ -35,6 +35,13 @@
 
 BLIND_USER_ERROR = 66
 
+# set up CORS
+cors_list = [
+    origin.strip()
+    for origin in get_settings().cors_origins.split(",")
+    if origin.strip()
+]
+
 # gatekeeper to ensure the user has read the docs and noted the major changes
 # since the last version.
 if not get_settings().i_read_the_damn_docs:
@@ -62,6 +69,15 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[Any, None]:
     # Initialize loguru logging within the server process.
     get_log_config()
 
+    if "*" in cors_list:
+        warning_msg = (
+            "CORS_ORIGINS is set to '*', allowing any origin to access the "
+            "API. This is fine for public APIs with bearer tokens, but you "
+            "should set explicit origins if serving browser clients."
+        )
+        logger.warning(warning_msg)  # Console via uvicorn
+        loguru_logger.warning(warning_msg)  # File via loguru
+
     redis_client = None
 
     # Test database connection
@@ -160,13 +176,10 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[Any, None]:
     name="static",
 )
 
-# set up CORS
-cors_list = (get_settings().cors_origins).split(",")
-
 app.add_middleware(
     CORSMiddleware,
     allow_origins=cors_list,
-    allow_credentials=True,
+    allow_credentials=False,
     allow_methods=["*"],
     allow_headers=["*"],
 )

app/managers/auth.py

@@ -52,6 +52,7 @@ def encode_token(user: User) -> str:
         try:
             payload = {
                 "sub": user.id,
+                "typ": "access",
                 "exp": datetime.datetime.now(tz=datetime.timezone.utc)
                 + datetime.timedelta(
                     minutes=get_settings().access_token_expire_minutes
@@ -492,6 +493,16 @@ async def get_jwt_user(
             algorithms=["HS256"],
             options={"verify_sub": False},
         )
+        if payload.get("typ") != "access":
+            increment_auth_failure("invalid_token", "jwt")
+            category_logger.warning(
+                "Authentication attempted with non-access token",
+                LogCategory.AUTH,
+            )
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail=ResponseMessages.INVALID_TOKEN,
+            )
         user_data = await get_user_by_id_(payload["sub"], db)
 
         # Check user validity - user must exist, be verified, and not banned

docs/deployment/deployment.md

@@ -16,8 +16,8 @@ Before deploying to production, ensure you've configured all critical settings p
   - Used for JWT token signing and session security
 
 - **CORS_ORIGINS**
-  - **Don't** leave as `*` in production
-  - Set to your actual frontend domain(s)
+  - For browser clients, set to your actual frontend domain(s)
+  - For public APIs using Bearer tokens, `*` is acceptable but will log a warning
   - Example: `CORS_ORIGINS=https://app.example.com,https://www.example.com`
   - Multiple origins separated by commas
 

docs/important.md

@@ -6,7 +6,12 @@ API.
 
 ## Breaking Changes in `HEAD`
 
-None.
+### CORS credentials disabled by default
+
+Credentialed CORS requests are now disabled for the API. If you relied on
+cookie-based auth from browser clients, you must switch to Bearer tokens or
+explicitly re-enable credentials and restrict `CORS_ORIGINS` to your frontend
+domains.
 
 ## Breaking Changes in `0.8.0`
 

docs/usage/configuration/environment.md

@@ -178,9 +178,11 @@ is an HTTP-header based mechanism that allows a server to indicate any origins
 (domain, scheme, or port) other than its own from which a browser should permit
 loading resources.
 
-For a **PUBLIC API** (unless its going through an API gateway!), set
-`CORS_ORIGINS=*`, otherwise list the domains (**and ports**) required. If you
-use an API gateway of some nature, that will probably need to be listed.
+For a **PUBLIC API** using Bearer tokens, `CORS_ORIGINS=*` is acceptable.
+If you serve browser clients, list the required domains (**and ports**) instead
+to restrict access. The app will log a warning when `CORS_ORIGINS=*` is set so
+you can confirm the intent. If you use an API gateway, that will probably need
+to be listed.
 
 ```ini
 CORS_ORIGINS=*

tests/integration/test_protected_user_routes.py

@@ -1,9 +1,15 @@
 """Integration tests for user routes."""
 
+import datetime
+
+import jwt
 import pytest
 from fastapi import status
 
+from app.config.settings import get_settings
 from app.database.helpers import hash_password
+from app.managers.auth import AuthManager
+from app.models.user import User
 
 
 @pytest.mark.integration
@@ -64,3 +70,57 @@ async def test_routes_bad_auth(self, client, route) -> None:
 
         assert response.status_code == status.HTTP_401_UNAUTHORIZED
         assert response.json() == {"detail": "That token is Invalid"}
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "route",
+        test_routes,
+    )
+    async def test_routes_refresh_token_rejected(
+        self, client, test_db, route
+    ) -> None:
+        """Test that refresh tokens are rejected on protected routes."""
+        test_user = User(**self.test_user)
+        test_db.add(test_user)
+        await test_db.commit()
+        refresh_token = AuthManager.encode_refresh_token(test_user)
+
+        route_name, method = route
+        fn = getattr(client, method)
+        response = await fn(
+            route_name, headers={"Authorization": f"Bearer {refresh_token}"}
+        )
+
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED
+        assert response.json() == {"detail": "That token is Invalid"}
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "route",
+        test_routes,
+    )
+    async def test_routes_missing_typ_rejected(
+        self, client, test_db, route
+    ) -> None:
+        """Test that tokens without typ are rejected on protected routes."""
+        test_user = User(**self.test_user)
+        test_db.add(test_user)
+        await test_db.commit()
+        token = jwt.encode(
+            {
+                "sub": test_user.id,
+                "exp": datetime.datetime.now(tz=datetime.timezone.utc)
+                + datetime.timedelta(minutes=10),
+            },
+            get_settings().secret_key,
+            algorithm="HS256",
+        )
+
+        route_name, method = route
+        fn = getattr(client, method)
+        response = await fn(
+            route_name, headers={"Authorization": f"Bearer {token}"}
+        )
+
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED
+        assert response.json() == {"detail": "That token is Invalid"}

tests/unit/test_auth_manager.py

@@ -42,6 +42,7 @@ def test_encode_token(self) -> None:
             options={"verify_sub": False},
         )
         assert payload["sub"] == 1
+        assert payload["typ"] == "access"
         assert isinstance(payload["exp"], int)
         # TODO(seapagan): better comparison to ensure the exp is in the future
         # but close to the expected expiry time taking into account the setting
@@ -68,6 +69,7 @@ def test_encode_refresh_token(self) -> None:
         )
 
         assert payload["sub"] == 1
+        assert payload["typ"] == "refresh"
         assert isinstance(payload["exp"], int)
         # TODO(seapagan): better comparison to ensure the exp is in the future
         # but close to the expected expiry time taking into account the expiry

tests/unit/test_cors_config.py

@@ -0,0 +1,24 @@
+"""Tests for CORS middleware configuration."""
+
+from typing import Any, cast
+
+import pytest
+from fastapi.middleware.cors import CORSMiddleware
+
+from app.main import app
+
+
+@pytest.mark.unit
+def test_cors_middleware_disables_credentials() -> None:
+    """Ensure the API does not allow credentialed CORS by default."""
+    cors_middleware = next(
+        middleware
+        for middleware in app.user_middleware
+        if cast("Any", middleware.cls) is CORSMiddleware
+    )
+
+    kwargs = cast("dict[str, Any]", cors_middleware.kwargs)
+    allow_origins = cast("list[str]", kwargs["allow_origins"])
+
+    assert kwargs["allow_credentials"] is False
+    assert "*" in allow_origins