feat: validate auth tokens

Author: alukach

pyproject.toml

@@ -8,6 +8,7 @@ classifiers = [
 dependencies = [
   "authlib>=1.3.2",
   "brotli>=1.1.0",
+  "eoapi-auth-utils>=0.4.0",
   "fastapi>=0.115.5",
   "httpx>=0.28.0",
   "pydantic-settings>=2.6.1",
@@ -29,7 +30,7 @@ known_first_party = ["stac_auth_proxy"]
 profile = "black"
 
 [tool.ruff]
-ignore = ["E501", "D205", "D212"]
+ignore = ["E501", "D203", "D205", "D212"]
 select = ["D", "E", "F"]
 
 [build-system]
@@ -38,6 +39,7 @@ requires = ["hatchling>=1.12.0"]
 
 [dependency-groups]
 dev = [
+  "jwcrypto>=1.5.6",
   "pre-commit>=3.5.0",
   "pytest>=8.3.3",
 ]

src/stac_auth_proxy/app.py

@@ -7,8 +7,8 @@
 
 from typing import Optional
 
+from eoapi.auth_utils import OpenIdConnectAuth
 from fastapi import Depends, FastAPI
-from fastapi.security import OpenIdConnect
 
 from .config import Settings
 from .handlers import OpenApiSpecHandler
@@ -21,13 +21,12 @@ def create_app(settings: Optional[Settings] = None) -> FastAPI:
     settings = settings or Settings()
 
     app = FastAPI(openapi_url=None)
+
     app.add_middleware(AddProcessTimeHeaderMiddleware)
 
-    auth_scheme = OpenIdConnect(
-        openIdConnectUrl=str(settings.oidc_discovery_url),
-        scheme_name="OpenID Connect",
-        description="OpenID Connect authentication for STAC API access",
-    )
+    auth_scheme = OpenIdConnectAuth(
+        openid_configuration_url=str(settings.oidc_discovery_url)
+    ).valid_token_dependency
 
     proxy = ReverseProxy(upstream=str(settings.upstream_url))
 

tests/conftest.py

@@ -1,10 +1,65 @@
 """Pytest fixtures."""
 
+import json
 import threading
+from typing import Any
+from unittest.mock import MagicMock, patch
 
 import pytest
 import uvicorn
 from fastapi import FastAPI
+from jwcrypto import jwk, jwt
+
+
+@pytest.fixture
+def test_key() -> jwk.JWK:
+    """Generate a test RSA key."""
+    return jwk.JWK.generate(
+        kty="RSA", size=2048, kid="test", use="sig", e="AQAB", alg="RS256"
+    )
+
+
+@pytest.fixture
+def public_key(test_key: jwk.JWK) -> dict[str, Any]:
+    """Export public key."""
+    return test_key.export_public(as_dict=True)
+
+
+@pytest.fixture(autouse=True)
+def mock_jwks(public_key: dict[str, Any]):
+    """Mock JWKS endpoint."""
+    mock_oidc_config = {"jwks_uri": "https://example.com/jwks"}
+
+    mock_jwks = {"keys": [public_key]}
+
+    with (
+        patch("urllib.request.urlopen") as mock_urlopen,
+        patch("jwt.PyJWKClient.fetch_data") as mock_fetch_data,
+    ):
+        mock_oidc_config_response = MagicMock()
+        mock_oidc_config_response.read.return_value = json.dumps(
+            mock_oidc_config
+        ).encode()
+        mock_oidc_config_response.status = 200
+
+        mock_urlopen.return_value.__enter__.return_value = mock_oidc_config_response
+        mock_fetch_data.return_value = mock_jwks
+        yield mock_urlopen
+
+
+@pytest.fixture
+def token_builder(test_key: jwk.JWK):
+    """Generate a valid JWT token builder."""
+
+    def build_token(payload: dict[str, Any], key=None) -> str:
+        jwt_token = jwt.JWT(
+            header={k: test_key.get(k) for k in ["alg", "kid"]},
+            claims=payload,
+        )
+        jwt_token.make_signed_token(key or test_key)
+        return jwt_token.serialize()
+
+    return build_token
 
 
 @pytest.fixture(scope="session")

tests/test_authn.py

@@ -0,0 +1,54 @@
+"""Test authentication cases for the proxy app."""
+
+import pytest
+from fastapi.testclient import TestClient
+
+from utils import AppFactory
+
+
+app_factory = AppFactory(
+    oidc_discovery_url="https://samples.auth0.com/.well-known/openid-configuration",
+    default_public=False,
+    public_endpoints={},
+    private_endpoints={},
+)
+
+
+@pytest.mark.parametrize(
+    "path,method",
+    [
+        ("/", "GET"),
+        ("/conformance", "GET"),
+        ("/queryables", "GET"),
+        ("/search", "GET"),
+        ("/search", "POST"),
+        ("/collections", "GET"),
+        ("/collections", "POST"),
+        ("/collections/example-collection", "GET"),
+        ("/collections/example-collection", "PUT"),
+        ("/collections/example-collection", "DELETE"),
+        ("/collections/example-collection/items", "GET"),
+        ("/collections/example-collection/items", "POST"),
+        ("/collections/example-collection/items/example-item", "GET"),
+        ("/collections/example-collection/items/example-item", "PUT"),
+        ("/collections/example-collection/items/example-item", "DELETE"),
+        ("/collections/example-collection/bulk_items", "POST"),
+        ("/api.html", "GET"),
+        ("/api", "GET"),
+    ],
+)
+def test_default_public_false(source_api_server, path, method, token_builder):
+    """
+    Private endpoints permit access with a valid token.
+    """
+    test_app = app_factory(upstream_url=source_api_server)
+    valid_auth_token = token_builder({})
+
+    client = TestClient(test_app)
+    response = client.request(method=method, url=path, headers={})
+    assert response.status_code == 403
+
+    response = client.request(
+        method=method, url=path, headers={"Authorization": f"Bearer {valid_auth_token}"}
+    )
+    assert response.status_code == 200