Cql2ApplyFilterBodyMiddleware

Middleware to augment the request body with a CQL2 filter for search requests.

Cql2ApplyFilterBodyMiddleware dataclass

Middleware to augment the request body with a CQL2 filter for search requests.

Parameters:

Name	Type	Description	Default
app	Callable[list, Awaitable[None]]		required
state_key	str		'cql2_filter'

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

@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",
)
@dataclass(frozen=True)
class Cql2ApplyFilterBodyMiddleware:
    """Middleware to augment the request body with a CQL2 filter for search requests."""

    app: ASGIApp
    state_key: str = "cql2_filter"

    search_body_endpoints = [
        r"^/search$",
    ]

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """Apply the CQL2 filter to the request body."""
        if scope["type"] != "http":
            return await self.app(scope, receive, send)

        request = Request(scope)
        cql2_filter: Optional[Expr] = getattr(request.state, self.state_key, None)
        if not cql2_filter:
            return await self.app(scope, receive, send)

        if request.method not in ["POST", "PUT", "PATCH"]:
            return await self.app(scope, receive, send)

        if not any(
            re.match(expr, request.url.path) for expr in self.search_body_endpoints
        ):
            return await self.app(scope, receive, send)

        body = b""
        more_body = True
        while more_body:
            message = await receive()
            if message["type"] == "http.request":
                body += message.get("body", b"")
                more_body = message.get("more_body", False)

        try:
            body_json = json.loads(body) if body else {}
        except json.JSONDecodeError:
            logger.warning("Failed to parse request body as JSON")
            from starlette.responses import JSONResponse

            response = JSONResponse(
                {
                    "code": "ParseError",
                    "description": "Request body must be valid JSON.",
                },
                status_code=400,
            )
            await response(scope, receive, send)
            return

        if not isinstance(body_json, dict):
            logger.warning("Request body must be a JSON object")
            from starlette.responses import JSONResponse

            response = JSONResponse(
                {
                    "code": "TypeError",
                    "description": "Request body must be a JSON object.",
                },
                status_code=400,
            )
            await response(scope, receive, send)
            return

        new_body = json.dumps(
            filters.append_body_filter(body_json, cql2_filter)
        ).encode("utf-8")

        # Patch content-length in the headers
        headers = dict(scope["headers"])
        headers[b"content-length"] = str(len(new_body)).encode("latin1")
        scope = dict(scope)
        scope["headers"] = list(headers.items())

        async def new_receive():
            return {
                "type": "http.request",
                "body": new_body,
                "more_body": False,
            }

        await self.app(scope, new_receive, send)

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

Apply the CQL2 filter to the request body.

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

async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
    """Apply the CQL2 filter to the request body."""
    if scope["type"] != "http":
        return await self.app(scope, receive, send)

    request = Request(scope)
    cql2_filter: Optional[Expr] = getattr(request.state, self.state_key, None)
    if not cql2_filter:
        return await self.app(scope, receive, send)

    if request.method not in ["POST", "PUT", "PATCH"]:
        return await self.app(scope, receive, send)

    if not any(
        re.match(expr, request.url.path) for expr in self.search_body_endpoints
    ):
        return await self.app(scope, receive, send)

    body = b""
    more_body = True
    while more_body:
        message = await receive()
        if message["type"] == "http.request":
            body += message.get("body", b"")
            more_body = message.get("more_body", False)

    try:
        body_json = json.loads(body) if body else {}
    except json.JSONDecodeError:
        logger.warning("Failed to parse request body as JSON")
        from starlette.responses import JSONResponse

        response = JSONResponse(
            {
                "code": "ParseError",
                "description": "Request body must be valid JSON.",
            },
            status_code=400,
        )
        await response(scope, receive, send)
        return

    if not isinstance(body_json, dict):
        logger.warning("Request body must be a JSON object")
        from starlette.responses import JSONResponse

        response = JSONResponse(
            {
                "code": "TypeError",
                "description": "Request body must be a JSON object.",
            },
            status_code=400,
        )
        await response(scope, receive, send)
        return

    new_body = json.dumps(
        filters.append_body_filter(body_json, cql2_filter)
    ).encode("utf-8")

    # Patch content-length in the headers
    headers = dict(scope["headers"])
    headers[b"content-length"] = str(len(new_body)).encode("latin1")
    scope = dict(scope)
    scope["headers"] = list(headers.items())

    async def new_receive():
        return {
            "type": "http.request",
            "body": new_body,
            "more_body": False,
        }

    await self.app(scope, new_receive, send)