stac_auth_proxy.middleware

Custom middleware.

OpenApiMiddleware dataclass

Bases: JsonResponseMiddleware

Middleware to add the OpenAPI spec to the response.

Parameters:

Name	Type	Description	Default
app	Callable[list, Awaitable[None]]		required
openapi_spec_path	str		required
oidc_discovery_url	str		required
private_endpoints	dict[str, Sequence[Literal[GET, POST, PUT, DELETE, PATCH]]]		required
public_endpoints	dict[str, Sequence[Literal[GET, POST, PUT, DELETE, PATCH]]]		required
default_public	bool		required
root_path	str		''
auth_scheme_name	str		'oidcAuth'
auth_scheme_override	dict | None		None
items_filter_path	str | None		None
collections_filter_path	str | None		None
json_content_type_expr	str		'application/(vnd\\.oai\\.openapi\\+json?|json)'

Source code in src/stac_auth_proxy/middleware/UpdateOpenApiMiddleware.py

@dataclass(frozen=True)
class OpenApiMiddleware(JsonResponseMiddleware):
    """Middleware to add the OpenAPI spec to the response."""

    app: ASGIApp
    openapi_spec_path: str
    oidc_discovery_url: str
    private_endpoints: EndpointMethods
    public_endpoints: EndpointMethods
    default_public: bool
    root_path: str = ""
    auth_scheme_name: str = "oidcAuth"
    auth_scheme_override: Optional[dict] = None

    items_filter_path: Optional[str] = None
    collections_filter_path: Optional[str] = None

    json_content_type_expr: str = r"application/(vnd\.oai\.openapi\+json?|json)"

    def should_transform_response(self, request: Request, scope: Scope) -> bool:
        """Only transform responses for the OpenAPI spec path."""
        return bool(
            match_path(self.openapi_spec_path, request.url.path)
            and re.match(
                self.json_content_type_expr,
                Headers(scope=scope).get("content-type", ""),
            )
            and 200 <= scope["status"] < 300
        )

    def transform_json(self, data: dict[str, Any], request: Request) -> dict[str, Any]:
        """Augment the OpenAPI spec with auth information."""
        # Remove any existing servers field from upstream API
        # This ensures we don't have conflicting server declarations
        if "servers" in data:
            del data["servers"]

        # Add servers field with root path if root_path is set
        if self.root_path:
            data["servers"] = [{"url": self.root_path}]

        # Add security scheme
        components = ensure_type(data, "components", dict)
        securitySchemes = ensure_type(components, "securitySchemes", dict)
        securitySchemes[self.auth_scheme_name] = self.auth_scheme_override or {
            "type": "openIdConnect",
            "openIdConnectUrl": self.oidc_discovery_url,
        }

        # Add security to private endpoints and filtered endpoints
        for path, method_config in data["paths"].items():
            for method, config in method_config.items():
                if method == "options":
                    # OPTIONS requests are not authenticated, https://fetch.spec.whatwg.org/#cors-protocol-and-credentials
                    continue
                match = find_match(
                    path,
                    method,
                    self.private_endpoints,
                    self.public_endpoints,
                    self.default_public,
                    items_filter_path=self.items_filter_path,
                    collections_filter_path=self.collections_filter_path,
                )
                if match.uses_auth:
                    security = ensure_type(config, "security", list)
                    security.append({self.auth_scheme_name: match.required_scopes})
        return data

should_transform_response(request: Request, scope: Scope) -> bool

Only transform responses for the OpenAPI spec path.

Source code in src/stac_auth_proxy/middleware/UpdateOpenApiMiddleware.py

def should_transform_response(self, request: Request, scope: Scope) -> bool:
    """Only transform responses for the OpenAPI spec path."""
    return bool(
        match_path(self.openapi_spec_path, request.url.path)
        and re.match(
            self.json_content_type_expr,
            Headers(scope=scope).get("content-type", ""),
        )
        and 200 <= scope["status"] < 300
    )

transform_json(data: dict[str, Any], request: Request) -> dict[str, Any]

Augment the OpenAPI spec with auth information.

Source code in src/stac_auth_proxy/middleware/UpdateOpenApiMiddleware.py

def transform_json(self, data: dict[str, Any], request: Request) -> dict[str, Any]:
    """Augment the OpenAPI spec with auth information."""
    # Remove any existing servers field from upstream API
    # This ensures we don't have conflicting server declarations
    if "servers" in data:
        del data["servers"]

    # Add servers field with root path if root_path is set
    if self.root_path:
        data["servers"] = [{"url": self.root_path}]

    # Add security scheme
    components = ensure_type(data, "components", dict)
    securitySchemes = ensure_type(components, "securitySchemes", dict)
    securitySchemes[self.auth_scheme_name] = self.auth_scheme_override or {
        "type": "openIdConnect",
        "openIdConnectUrl": self.oidc_discovery_url,
    }

    # Add security to private endpoints and filtered endpoints
    for path, method_config in data["paths"].items():
        for method, config in method_config.items():
            if method == "options":
                # OPTIONS requests are not authenticated, https://fetch.spec.whatwg.org/#cors-protocol-and-credentials
                continue
            match = find_match(
                path,
                method,
                self.private_endpoints,
                self.public_endpoints,
                self.default_public,
                items_filter_path=self.items_filter_path,
                collections_filter_path=self.collections_filter_path,
            )
            if match.uses_auth:
                security = ensure_type(config, "security", list)
                security.append({self.auth_scheme_name: match.required_scopes})
    return data

RestoreRootPathMiddleware dataclass

Put back the root path RemoveRootPathMiddleware removed, so routing (including Mounts) and URLs built from root_path (docs, url_for) see the usual ASGI scope. Request handlers then route on, and the proxy forwards, the path the checks saw.

IMPORTANT: Must be the innermost middleware (first declared).

Parameters:

Name	Type	Description	Default
app	Callable[list, Awaitable[None]]		required

Source code in src/stac_auth_proxy/middleware/RemoveRootPathMiddleware.py

@dataclass
class RestoreRootPathMiddleware:
    """
    Put back the root path RemoveRootPathMiddleware removed, so routing (including
    Mounts) and URLs built from root_path (docs, url_for) see the usual ASGI scope.
    Request handlers then route on, and the proxy forwards, the path the checks saw.

    IMPORTANT: Must be the innermost middleware (first declared).
    """

    app: ASGIApp

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """Restore the original path and root_path."""
        if scope["type"] != "http" or _ORIGINAL not in scope:
            return await self.app(scope, receive, send)

        root_path, path, raw_path = scope[_ORIGINAL]
        if path == root_path:
            # Starlette would route "/stac" under root_path "/stac" as "", which
            # matches nothing and redirects; the checks treated it as "/"
            path = f"{path}/"
            raw_path = raw_path + b"/" if raw_path is not None else None
        inner = {**scope, "root_path": root_path, "path": path}
        if raw_path is not None:
            inner["raw_path"] = raw_path

        def share_new_keys() -> None:
            # Keys set or changed while routing (route, endpoint, path_params,
            # state) are visible to outer middleware, as without this middleware
            scope.update(
                (key, value)
                for key, value in inner.items()
                if key not in ("root_path", "path", "raw_path")
            )

        async def send_wrapper(message: Message) -> None:
            if message["type"] == "http.response.start":
                share_new_keys()
            await send(message)

        try:
            await self.app(inner, receive, send_wrapper)
        finally:
            share_new_keys()

__call__(scope: Scope, receive: Receive, send: Send) -> None async

Restore the original path and root_path.

Source code in src/stac_auth_proxy/middleware/RemoveRootPathMiddleware.py

async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
    """Restore the original path and root_path."""
    if scope["type"] != "http" or _ORIGINAL not in scope:
        return await self.app(scope, receive, send)

    root_path, path, raw_path = scope[_ORIGINAL]
    if path == root_path:
        # Starlette would route "/stac" under root_path "/stac" as "", which
        # matches nothing and redirects; the checks treated it as "/"
        path = f"{path}/"
        raw_path = raw_path + b"/" if raw_path is not None else None
    inner = {**scope, "root_path": root_path, "path": path}
    if raw_path is not None:
        inner["raw_path"] = raw_path

    def share_new_keys() -> None:
        # Keys set or changed while routing (route, endpoint, path_params,
        # state) are visible to outer middleware, as without this middleware
        scope.update(
            (key, value)
            for key, value in inner.items()
            if key not in ("root_path", "path", "raw_path")
        )

    async def send_wrapper(message: Message) -> None:
        if message["type"] == "http.response.start":
            share_new_keys()
        await send(message)

    try:
        await self.app(inner, receive, send_wrapper)
    finally:
        share_new_keys()