Zapros has an Unbounded Content-Encoding decompression chain that allows denial of service

from typing import cast

from zapros import (
    AsyncBaseHandler,
    AsyncBaseMiddleware,
    BaseHandler,
    BaseMiddleware,
    Client,
    DecodingError,
    Request,
    Response,
)

MAX_DECODE_LAYERS = 5


class ContentEncodingCheckMiddleware(BaseMiddleware, AsyncBaseMiddleware):
    def __init__(
        self,
        next_handler: BaseHandler | AsyncBaseHandler,
        *,
        max_layers: int = MAX_DECODE_LAYERS,
    ) -> None:
        self.next = cast(BaseHandler, next_handler)
        self.async_next = cast(AsyncBaseHandler, next_handler)
        self._max_layers = max_layers

    def _check(self, response: Response) -> None:
        encoding_header = response.headers.get("Content-Encoding")
        if not encoding_header:
            return

        layers = [enc.strip().lower() for enc in encoding_header.split(",") if enc.strip()]
        if len(layers) > self._max_layers:
            raise DecodingError(f"Too many Content-Encoding layers ({len(layers)}), maximum is {self._max_layers}")

    def handle(self, request: Request) -> Response:
        response = self.next.handle(request)
        self._check(response)
        return response

    async def ahandle(self, request: Request) -> Response:
        response = await self.async_next.ahandle(request)
        self._check(response)
        return response


with Client().wrap_with_middleware(lambda next: ContentEncodingCheckMiddleware(next)) as client:
    ...