feat: add middleware conformance checks

Author: alukach

src/stac_auth_proxy/app.py

@@ -21,7 +21,11 @@
     EnforceAuthMiddleware,
     OpenApiMiddleware,
 )
-from .utils.lifespan import check_server_health
+from .utils.lifespan import (
+    check_conformance,
+    check_server_health,
+    log_middleware_classes,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -40,9 +44,18 @@ async def lifespan(app: FastAPI):
 
         # Wait for upstream servers to become available
         if settings.wait_for_upstream:
+            logger.info("Running upstream server health checks...")
             for url in [settings.upstream_url, settings.oidc_discovery_internal_url]:
                 await check_server_health(url=url)
 
+        # Log all middleware connected to the app
+        await log_middleware_classes(app.user_middleware)
+        if settings.check_conformance:
+            await check_conformance(
+                app.user_middleware,
+                str(settings.upstream_url),
+            )
+
         yield
 
     app = FastAPI(

src/stac_auth_proxy/config.py

@@ -39,6 +39,7 @@ class Settings(BaseSettings):
     oidc_discovery_internal_url: HttpUrl
 
     wait_for_upstream: bool = True
+    check_conformance: bool = True
 
     # Endpoints
     healthz_prefix: str = Field(pattern=_PREFIX_PATTERN, default="/healthz")

src/stac_auth_proxy/middleware/ApplyCql2FilterMiddleware.py

@@ -13,10 +13,18 @@
 from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
 from ..utils import filters
+from ..utils.middleware import required_conformance
 
 logger = getLogger(__name__)
 
 
+@required_conformance(
+    r"http://www.opengis.net/spec/cql2/1.0/conf/basic-cql2",
+    r"http://www.opengis.net/spec/cql2/1.0/conf/cql2-text",
+    r"http://www.opengis.net/spec/cql2/1.0/conf/cql2-json",
+    r"http://www.opengis.net/spec/ogcapi-features-3/1.0/conf/features-filter",
+    r"https://api.stacspec.org/v1\.\d+\.\d+(?:-[\w\.]+)?/item-search#filter",
+)
 @dataclass(frozen=True)
 class ApplyCql2FilterMiddleware:
     """Middleware to apply the Cql2Filter to the request."""

src/stac_auth_proxy/utils/lifespan.py

@@ -2,9 +2,11 @@
 
 import asyncio
 import logging
+import re
 
 import httpx
 from pydantic import HttpUrl
+from starlette.middleware import Middleware
 
 logger = logging.getLogger(__name__)
 
@@ -40,3 +42,58 @@ async def check_server_health(
     raise RuntimeError(
         f"Upstream API {url!r} failed to respond after {max_retries} attempts"
     )
+
+
+async def log_middleware_classes(middleware_classes: list[Middleware]):
+    """Log the middleware classes connected to the application."""
+    logger.debug(
+        "Connected middleware:\n%s",
+        "\n".join(
+            [f"- {middleware.cls.__name__}" for middleware in middleware_classes]
+        ),
+    )
+
+
+async def check_conformance(
+    middleware_classes: list[Middleware],
+    api_url: str,
+    attr_name: str = "__required_conformances__",
+):
+    """Check if the upstream API supports a given conformance class."""
+    required_conformances: dict[str, list[str]] = {}
+    for middleware in middleware_classes:
+
+        for conformance in getattr(middleware.cls, attr_name, []):
+            required_conformances.setdefault(conformance, []).append(
+                middleware.cls.__name__
+            )
+
+    async with httpx.AsyncClient() as client:
+        response = await client.get(api_url)
+        response.raise_for_status()
+        api_conforms_to = response.json().get("conformsTo", [])
+    missing = [
+        req_conformance
+        for req_conformance in required_conformances.keys()
+        if not any(
+            re.match(req_conformance, conformance) for conformance in api_conforms_to
+        )
+    ]
+
+    def print_conformance(conformance):
+        return f" - {conformance} [{','.join(required_conformances[conformance])}]"
+
+    if missing:
+        missing_str = [print_conformance(c) for c in missing]
+        raise RuntimeError(
+            "\n".join(
+                [
+                    "Upstream catalog is missing the following conformance classes:",
+                    *missing_str,
+                ]
+            )
+        )
+    logger.debug(
+        "Upstream catalog conforms to the following required conformance classes: \n%s",
+        "\n".join([print_conformance(c) for c in required_conformances]),
+    )

src/stac_auth_proxy/utils/middleware.py

@@ -99,3 +99,13 @@ async def transform_response(message: Message) -> None:
             )
 
         return await self.app(scope, receive, transform_response)
+
+
+def required_conformance(*conformances: str):
+    """Register required conformance classes with a middleware class."""
+
+    def decorator(func):
+        func.__required_conformances__ = list(conformances)
+        return func
+
+    return decorator