Authenticate option requests.

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Change Log
 
+## [2.0.1] - 2024-07-15
+- Security Improvement: Added authentication and authorization checks for HTTP
+  OPTIONS requests in OpaMiddleware. This ensures that OPTIONS requests are
+  subjected to the same security policies as other HTTP methods, preventing
+  potential information leaks.
+  [See advisory for more details](https://github.com/advisories/GHSA-5f5c-8rvc-j8wf)
+- Update dependencies due to multiple vulnerabilities.
+
 ## [2.0.0] - 2024-02-07
 - Drop Python 3.7 support due to FastAPI update
 - Update dependencies due to vulnerabilities:

fastapi_opa/opa/opa_middleware.py

@@ -18,12 +18,7 @@
 from fastapi_opa.auth.exceptions import AuthenticationException
 from fastapi_opa.opa.opa_config import OPAConfig
 
-try:
-    Pattern = re.Pattern
-except AttributeError:
-    # Python3.6 does not contain re.Pattern
-    Pattern = None
-
+Pattern = re.Pattern
 logger = logging.getLogger(__name__)
 
 
@@ -76,15 +71,13 @@ async def __call__(
         own_receive = OwnReceive(receive)
         request = Request(scope, own_receive, send)
 
-        if request.method == "OPTIONS":
-            return await self.app(scope, receive, send)
-
         # allow openapi endpoints without authentication
         if should_skip_endpoint(request.url.path, self.skip_endpoints):
             return await self.app(scope, receive, send)
 
         # authenticate user or get redirect to identity provider
         successful = False
+        user_info_or_auth_redirect = None
         for auth in self.config.authentication:
             try:
                 user_info_or_auth_redirect = auth.authenticate(

tests/conftest.py

@@ -3,6 +3,8 @@
 import nest_asyncio
 import pytest
 from fastapi import FastAPI
+from fastapi import HTTPException
+from fastapi import Response
 from fastapi.testclient import TestClient
 
 from fastapi_opa import OPAConfig
@@ -15,6 +17,12 @@
 
 nest_asyncio.apply()
 
+# Sample data for the test
+WRITABLE_ITEMS = {
+    1: True,
+    2: False,
+}
+
 
 @pytest.fixture
 def client():
@@ -29,6 +37,21 @@ def client():
     async def root() -> Dict:
         return {"msg": "success"}
 
+    @app.get("/items/{item_id}")
+    async def read_item(item_id: int):
+        if item_id not in WRITABLE_ITEMS:
+            raise HTTPException(status_code=404)
+        return {"item_id": item_id}
+
+    @app.options("/items/{item_id}")
+    async def read_item_options(response: Response, item_id: int) -> Dict:
+        if item_id not in WRITABLE_ITEMS:
+            raise HTTPException(status_code=404)
+        response.headers["Allow"] = "OPTIONS, GET" + (
+            ", POST" if WRITABLE_ITEMS[item_id] else ""
+        )
+        return {}
+
     yield TestClient(app)
 
 
@@ -76,6 +99,21 @@ def client_multiple_authentications(api_key_auth):
     async def root() -> Dict:
         return {"msg": "success"}
 
+    @app.get("/items/{item_id}")
+    async def read_item(item_id: int):
+        if item_id not in WRITABLE_ITEMS:
+            raise HTTPException(status_code=404)
+        return {"item_id": item_id}
+
+    @app.options("/items/{item_id}")
+    async def read_item_options(response: Response, item_id: int) -> Dict:
+        if item_id not in WRITABLE_ITEMS:
+            raise HTTPException(status_code=404)
+        response.headers["Allow"] = "OPTIONS, GET" + (
+            ", POST" if WRITABLE_ITEMS[item_id] else ""
+        )
+        return {}
+
     yield TestClient(app)
 
 

tests/test_option_requests.py

@@ -0,0 +1,51 @@
+from unittest.mock import patch
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture
+def mock_opa_response():
+    with patch("fastapi_opa.opa.opa_middleware.requests.post") as mock_post:
+        mock_post.return_value.status_code = 200
+        mock_post.return_value.json.return_value = {"result": {"allow": True}}
+        yield mock_post
+
+
+def test_options_request_with_auth(
+    client_multiple_authentications, api_key_auth, mock_opa_response
+):
+    client: TestClient = client_multiple_authentications
+
+    # Test OPTIONS request for an existing item with authentication
+    response = client.options(
+        "/items/1",
+        headers={api_key_auth["header_key"]: api_key_auth["api_key"]},
+    )
+    assert response.status_code == 200
+    assert response.headers["Allow"] == "OPTIONS, GET, POST"
+    assert response.json() == {}
+
+    # Test OPTIONS request for a non-existing item with authentication
+    response = client.options(
+        "/items/3",
+        headers={api_key_auth["header_key"]: api_key_auth["api_key"]},
+    )
+    assert response.status_code == 404
+    assert response.json() == {"detail": "Not Found"}
+
+
+def test_options_request_without_auth(
+    client_multiple_authentications, mock_opa_response
+):
+    client: TestClient = client_multiple_authentications
+
+    # Test OPTIONS request for an existing item without authentication
+    response = client.options("/items/1")
+    assert response.status_code == 401
+    assert response.json() == {"message": "Unauthorized"}
+
+    # Test OPTIONS request for a non-existing item without authentication
+    response = client.options("/items/3")
+    assert response.status_code == 401
+    assert response.json() == {"message": "Unauthorized"}