feat: configurable path param regex

docs/user-guide/configuration.md

@@ -323,10 +323,10 @@ These settings configure the CORS behavior when `PROXY_OPTIONS` is `false` (the
 
 ### `ITEMS_FILTER_PATH`
 
-: Regex pattern used to identify request paths that require the application of the items filter
+: Regex patterns used to identify request paths that require the application of the items filter. See [Filter paths and path params](#filter-paths-and-path-params).
 
-    - **Type:** Regex string
-    - **Required:** No, defaults to `^(/collections/([^/]+)/items(/[^/]+)?$|/search$)`
+    - **Type:** Regex string, or a JSON array of regex strings
+    - **Required:** No, defaults to `["^(?:/collections/(?P<collection_id>[^/]+)/items(?:/(?P<item_id>[^/]+))?|/search)$"]`
     - **Example:** `^(/collections/([^/]+)/items(/[^/]+)?$|/search$|/custom$)`
 
 ### `COLLECTIONS_FILTER_CLS`
@@ -355,8 +355,58 @@ These settings configure the CORS behavior when `PROXY_OPTIONS` is `false` (the
 
 ### `COLLECTIONS_FILTER_PATH`
 
-: Regex pattern used to identify request paths that require the application of the collections filter
+: Regex patterns used to identify request paths that require the application of the collections filter. See [Filter paths and path params](#filter-paths-and-path-params).
 
-    - **Type:** Regex string
-    - **Required:** No, defaults to `^/collections(/[^/]+)?$`
+    - **Type:** Regex string, or a JSON array of regex strings
+    - **Required:** No, defaults to `["^/collections(?:/(?P<collection_id>[^/]+))?$"]`
     - **Example:** `^.*?/collections(/[^/]+)?$`
+
+### Filter paths and path params
+
+`ITEMS_FILTER_PATH` and `COLLECTIONS_FILTER_PATH` do two jobs:
+
+1. **Scope**: They select the request paths a filter applies to.
+2. **Information**: They declare the request "path parameters" information handed to the filter.
+
+**Path params.** Named capture groups in the pattern that matched become `req.path_params`. A pattern declaring no named groups (the default) instead falls back to built-in extraction, which recognizes `/collections/{collection_id}` optionally followed by `items`, `bulk_items`, or `queryables` and an item ID. Patterns using the fallback are named in a log line at startup.
+
+**Supporting several patterns.** Covering all of your endpoints may require more than one pattern, especially if using named capture groups which do not support redefinition of the group name. To provide more than one pattern, provide the input as a JSON array.
+
+**Authentication.** A path matching either setting always requires authentication, whatever `DEFAULT_PUBLIC` is set to, and is marked accordingly in the OpenAPI spec. A separate `PRIVATE_ENDPOINTS` entry is not needed.
+
+**Example: the Aggregation extension.** The [STAC API Aggregation extension](https://github.com/stac-api-extensions/aggregation) adds four endpoints, which `stac-fastapi` registers as:
+
+| Path | Methods |
+| --- | --- |
+| `/aggregate` | GET |
+| `/aggregations` | GET |
+| `/collections/{collection_id}/aggregate` | GET |
+| `/collections/{collection_id}/aggregations` | GET |
+
+None are covered by the defaults. The two collection-scoped endpoints belong to the collections filter, and each needs its own pattern because both declare `collection_id`:
+
+```
+COLLECTIONS_FILTER_PATH='[
+  "^/collections(?:/(?P<collection_id>[^/]+))?$",
+  "^/collections/(?P<collection_id>[^/]+)/aggregate$",
+  "^/collections/(?P<collection_id>[^/]+)/aggregations$"
+]'
+```
+
+The two root-level endpoints aggregate across Items, so they belong to the items filter alongside `/search`:
+
+```
+ITEMS_FILTER_PATH='[
+  "^(?:/collections/(?P<collection_id>[^/]+)/items(?:/(?P<item_id>[^/]+))?|/search)$",
+  "^/aggregate$",
+  "^/aggregations$"
+]'
+```
+
+A pattern passes whatever it declares, so a path carrying more variables passes more of them:
+
+```
+"^/mosaic/(?P<zoom>[^/]+)/(?P<x>[^/]+)/(?P<y>[^/]+)/(?P<collection_id>[^/]+)\\.png$"
+```
+
+gives the filter `zoom`, `x`, `y`, and `collection_id`.

src/stac_auth_proxy/config.py

@@ -2,6 +2,7 @@
 
 import importlib
 import json
+import re
 from typing import Annotated, Any, Literal, Optional, Sequence, TypeAlias, Union
 
 from pydantic import BaseModel, Field, field_validator, model_validator
@@ -28,6 +29,36 @@ def str2list(x: str | Sequence[str] | None) -> Sequence[str] | None:
     return x
 
 
+# NoDecode: pydantic-settings JSON-decodes Sequence fields before validators run,
+# which would reject a plain regex string.
+FilterPaths: TypeAlias = Annotated[Sequence[str], NoDecode]
+
+
+def str2patterns(x: str | Sequence[str] | None) -> Sequence[str]:
+    """
+    Parse a filter path or paths setting into a list of regex patterns.
+
+    Used as a Pydantic validator, this function supports:
+    - Single path pattern input as a string
+    - Multiple path patterns input as a JSON encoded string
+    - Directly passing multiple path parameters
+    """
+    if x is None:
+        return []
+
+    if isinstance(x, str):
+        patterns = json.loads(x) if x.startswith("[") else [x]
+    else:
+        patterns = list(x)
+
+    for pattern in patterns:
+        try:
+            re.compile(pattern)
+        except re.error as e:
+            raise ValueError(f"{pattern!r} is not a valid regular expression: {e}")
+    return patterns
+
+
 class _ClassInput(BaseModel):
     """Input model for dynamically loading a class or function."""
 
@@ -129,9 +160,13 @@ class Settings(BaseSettings):
 
     # Filters
     items_filter: Optional[_ClassInput] = None
-    items_filter_path: str = r"^(/collections/([^/]+)/items(/[^/]+)?$|/search$)"
+    items_filter_path: FilterPaths = [
+        r"^(?:/collections/(?P<collection_id>[^/]+)/items(?:/(?P<item_id>[^/]+))?|/search)$"
+    ]
     collections_filter: Optional[_ClassInput] = None
-    collections_filter_path: str = r"^/collections(/[^/]+)?$"
+    collections_filter_path: FilterPaths = [
+        r"^/collections(?:/(?P<collection_id>[^/]+))?$"
+    ]
 
     model_config = SettingsConfigDict(
         env_nested_delimiter="_",
@@ -151,6 +186,17 @@ def parse_audience(cls, v) -> Sequence[str] | None:
         """Parse a comma separated string list of audiences into a list."""
         return str2list(v)
 
+    @field_validator("items_filter_path", "collections_filter_path", mode="before")
+    @classmethod
+    def parse_filter_paths(cls, v) -> Sequence[str]:
+        """
+        Parse the regex patterns identifying paths that a filter applies to.
+
+        Named capture groups in a pattern become the ``req.path_params`` passed to
+        the filter. A pattern declaring none falls back to the built-in extraction.
+        """
+        return str2patterns(v)
+
     @field_validator("root_path_skip_prefixes", mode="before")
     @classmethod
     def parse_root_path_skip_prefixes(cls, v) -> Sequence[str]:

src/stac_auth_proxy/middleware/AuthenticationExtensionMiddleware.py

@@ -3,7 +3,7 @@
 import logging
 import re
 from dataclasses import dataclass, field
-from typing import Any, Optional
+from typing import Any, Sequence
 from urllib.parse import urlparse
 
 from starlette.datastructures import Headers
@@ -35,8 +35,8 @@ class AuthenticationExtensionMiddleware(JsonResponseMiddleware):
         "https://stac-extensions.github.io/authentication/v1.1.0/schema.json"
     )
 
-    items_filter_path: Optional[str] = None
-    collections_filter_path: Optional[str] = None
+    items_filter_path: str | Sequence[str] | None = None
+    collections_filter_path: str | Sequence[str] | None = None
     root_path: str = ""
 
     json_content_type_expr: str = r"application/(geo\+)?json"

src/stac_auth_proxy/middleware/Cql2BuildFilterMiddleware.py

@@ -3,7 +3,7 @@
 import logging
 import re
 from dataclasses import dataclass
-from typing import Any, Awaitable, Callable, Optional
+from typing import Any, Awaitable, Callable, Optional, Sequence
 
 from cql2 import Expr, ValidationError
 from fastapi import HTTPException
@@ -32,12 +32,26 @@ class Cql2BuildFilterMiddleware:
 
     # Filters
     collections_filter: Optional[Callable] = None
-    collections_filter_path: str = r"^/collections(/[^/]+)?$"
+    collections_filter_path: str | Sequence[str] = (
+        r"^/collections(?:/(?P<collection_id>[^/]+))?$",
+    )
     items_filter: Optional[Callable] = None
-    items_filter_path: str = r"^(/collections/([^/]+)/items(/[^/]+)?$|/search$)"
+    items_filter_path: str | Sequence[str] = (
+        r"^(?:/collections/(?P<collection_id>[^/]+)/items(?:/(?P<item_id>[^/]+))?|/search)$",
+    )
 
     def __post_init__(self):
         """Set required conformances based on the filter functions."""
+        for attr in ("collections_filter_path", "items_filter_path"):
+            object.__setattr__(self, attr, requests.as_patterns(getattr(self, attr)))
+            for pattern in getattr(self, attr):
+                if not re.compile(pattern).groupindex:
+                    logger.info(
+                        "Filter path %r declares no named capture groups, "
+                        "falling back to built-in path param extraction.",
+                        pattern,
+                    )
+
         required_conformances = set()
         if self.collections_filter:
             logger.debug("Appending required conformance for collections filter")
@@ -77,7 +91,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
             logger.debug("Skipping CQL2 filter build for OPTIONS request")
             return await self.app(scope, receive, send)
 
-        filter_builder = self._get_filter(request.url.path)
+        filter_builder, path_params = self._get_filter(request.url.path)
         if not filter_builder:
             return await self.app(scope, receive, send)
 
@@ -88,7 +102,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
                         "path": request.url.path,
                         "method": request.method,
                         "query_params": dict(request.query_params),
-                        "path_params": requests.extract_variables(request.url.path),
+                        "path_params": path_params,
                         "headers": dict(request.headers),
                     },
                     **scope["state"],
@@ -112,13 +126,22 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
 
     def _get_filter(
         self, path: str
-    ) -> Optional[Callable[..., Awaitable[str | dict[str, Any]]]]:
-        """Get the CQL2 filter builder for the given path."""
+    ) -> tuple[Optional[Callable[..., Awaitable[str | dict[str, Any]]]], dict]:
+        """Get the CQL2 filter builder for the given path and its path params."""
         endpoint_filters = [
             (self.collections_filter_path, self.collections_filter),
             (self.items_filter_path, self.items_filter),
         ]
-        for expr, builder in endpoint_filters:
-            if re.match(expr, path):
-                return builder
-        return None
+        for patterns, builder in endpoint_filters:
+            for expr in patterns:
+                match = re.match(expr, path)
+                if match:
+                    return builder, self._path_params(match, path)
+        return None, {}
+
+    @staticmethod
+    def _path_params(match: re.Match, path: str) -> dict:
+        """Get the path params declared by a matched pattern's named groups."""
+        if match.re.groupindex:
+            return {k: v for k, v in match.groupdict().items() if v is not None}
+        return requests.extract_variables(path)

src/stac_auth_proxy/middleware/UpdateOpenApiMiddleware.py

@@ -2,7 +2,7 @@
 
 import re
 from dataclasses import dataclass
-from typing import Any, Optional
+from typing import Any, Optional, Sequence
 
 from starlette.datastructures import Headers
 from starlette.requests import Request
@@ -28,8 +28,8 @@ class OpenApiMiddleware(JsonResponseMiddleware):
     auth_scheme_name: str = "oidcAuth"
     auth_scheme_override: Optional[dict] = None
 
-    items_filter_path: Optional[str] = None
-    collections_filter_path: Optional[str] = None
+    items_filter_path: str | Sequence[str] | None = None
+    collections_filter_path: str | Sequence[str] | None = None
 
     json_content_type_expr: str = r"application/(vnd\.oai\.openapi\+json?|json)"
 

src/stac_auth_proxy/utils/requests.py

@@ -26,6 +26,13 @@ def extract_variables(url: str) -> dict:
     return {k: v for k, v in match.groupdict().items() if v} if match else {}
 
 
+def as_patterns(value: str | Sequence[str] | None) -> Sequence[str]:
+    """Normalize a filter path setting to a list of regex patterns."""
+    if value is None:
+        return []
+    return [value] if isinstance(value, str) else list(value)
+
+
 def dict_to_bytes(d: dict) -> bytes:
     """Convert a dictionary to a body."""
     return json.dumps(d, separators=(",", ":")).encode("utf-8")
@@ -56,8 +63,8 @@ def find_match(
     private_endpoints: EndpointMethods,
     public_endpoints: EndpointMethods,
     default_public: bool,
-    items_filter_path: Optional[str] = None,
-    collections_filter_path: Optional[str] = None,
+    items_filter_path: str | Sequence[str] | None = None,
+    collections_filter_path: str | Sequence[str] | None = None,
 ) -> "MatchResult":
     """Check if the given path and method match any of the regex patterns and methods in the endpoints."""
     primary_endpoints = private_endpoints if default_public else public_endpoints
@@ -70,7 +77,7 @@ def find_match(
 
     # If we have filter paths configured, check those as well (these are always considered to use auth if they match, regardless of default_public)
     for filter_path in [items_filter_path, collections_filter_path]:
-        if filter_path and re.match(filter_path, path):
+        if any(re.match(pattern, path) for pattern in as_patterns(filter_path)):
             return MatchResult(uses_auth=True)
 
     # If default_public and no match found in private_endpoints, it's public

tests/test_config.py

@@ -151,3 +151,50 @@ def test_cors_model_config():
     ]
     assert cors_settings.allow_methods == ["GET", "POST"]
     assert cors_settings.allow_headers == ["Authorization", "Content-Type"]
+
+
+def test_items_and_collections_path_parameters():
+    """Tests related to parsing Collections/Items regexes from inputs."""
+    common_kwargs = {
+        "upstream_url": "https://example.com",
+        "oidc_discovery_url": "https://example.com/.well-known/openid-configuration",
+    }
+
+    # Single pattern case
+    settings = Settings(
+        **common_kwargs,
+        items_filter_path=r"^/collections/([^/]+)/items$",
+    )
+    assert settings.items_filter_path == [r"^/collections/([^/]+)/items$"]
+
+    # Don't split on commas (valid regex)
+    settings = Settings(
+        **common_kwargs,
+        items_filter_path=r"^/collections/([^/]{2,64})/items$",
+    )
+    assert settings.items_filter_path == [r"^/collections/([^/]{2,64})/items$"]
+
+    # JSON array decoded into list[str]
+    settings = Settings(
+        **common_kwargs,
+        collections_filter_path='["^/a$", "^/b$"]',
+    )
+    assert settings.collections_filter_path == ["^/a$", "^/b$"]
+
+    # Directly provided list[str] unaltered
+    custom_paths = [
+        r"^/collections(?:/(?P<collection_id>[^/]+))?$",
+        r"^/collections/(?P<collection_id>[^/]+)/aggregate$",
+    ]
+    settings = Settings(
+        **common_kwargs,
+        collections_filter_path=custom_paths,
+    )
+    assert settings.collections_filter_path == custom_paths
+
+    # Reject invalid regex at load
+    with pytest.raises(ValueError, match="not a valid regular expression"):
+        settings = Settings(
+            **common_kwargs,
+            items_filter_path=r"^/collections(?:/(?P<unclosed[^/]+))?$",
+        )