Skip to content

stac_auth_proxy.middleware.RejectAmbiguousPathMiddleware

Middleware to reject requests whose checked and routed paths may differ.

RejectAmbiguousPathMiddleware dataclass

Reject requests whose request.url.path differs from the routed path.

Auth, filter and transaction checks match on request.url.path, which Starlette rebuilds from the scope's scheme, server and path. Any request where its path differs from scope["path"] (e.g. "/search%23" truncated to "/search", or a scheme copied from a client's X-Forwarded-Proto) is rejected.

Paths that are only ambiguous once forwarded upstream are rejected by ReverseProxyHandler, so the app's own routes (non-proxy mode) accept them.

IMPORTANT: Must run before any middleware that uses the request path.

Parameters:

Name Type Description Default
app Callable[list, Awaitable[None]]
required
Source code in src/stac_auth_proxy/middleware/RejectAmbiguousPathMiddleware.py
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@dataclass(frozen=True)
class RejectAmbiguousPathMiddleware:
    """
    Reject requests whose ``request.url.path`` differs from the routed path.

    Auth, filter and transaction checks match on ``request.url.path``, which
    Starlette rebuilds from the scope's scheme, server and path. Any request where
    its path differs from ``scope["path"]`` (e.g. "/search%23" truncated to
    "/search", or a scheme copied from a client's X-Forwarded-Proto) is rejected.

    Paths that are only ambiguous once forwarded upstream are rejected by
    ``ReverseProxyHandler``, so the app's own routes (non-proxy mode) accept them.

    IMPORTANT: Must run before any middleware that uses the request path.
    """

    app: ASGIApp

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """Reject ambiguous request paths."""
        if scope["type"] == "http":
            error = None
            if not self.url_matches_scope(scope):
                error = "Invalid request path."
            elif not self.query_is_utf8(scope):
                # Starlette's request.url (used by every check) can't decode it
                error = "Invalid query string."
            if error:
                return await bad_request(error)(scope, receive, send)
        return await self.app(scope, receive, send)

    @staticmethod
    def query_is_utf8(scope: Scope) -> bool:
        """Whether the raw query string is valid UTF-8."""
        try:
            scope.get("query_string", b"").decode()
        except UnicodeDecodeError:
            return False
        return True

    @staticmethod
    def url_matches_scope(scope: Scope) -> bool:
        """Whether Starlette's request URL has the scope's path."""
        try:
            # Without the query string, which isn't compared (and may not be UTF-8)
            url = URL(scope={**scope, "query_string": b""})
        except (KeyError, ValueError, TypeError):  # e.g. unknown scheme
            return False
        # Only the path is used by the checks. The query string is forwarded and
        # parsed from scope["query_string"], never request.url.query (which a raw "#"
        # truncates), and the scheme and host may legitimately be missing (e.g. a
        # unix socket with no valid Host header).
        return url.path == scope["path"]

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

Reject ambiguous request paths.

Source code in src/stac_auth_proxy/middleware/RejectAmbiguousPathMiddleware.py
29
30
31
32
33
34
35
36
37
38
39
40
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
    """Reject ambiguous request paths."""
    if scope["type"] == "http":
        error = None
        if not self.url_matches_scope(scope):
            error = "Invalid request path."
        elif not self.query_is_utf8(scope):
            # Starlette's request.url (used by every check) can't decode it
            error = "Invalid query string."
        if error:
            return await bad_request(error)(scope, receive, send)
    return await self.app(scope, receive, send)

query_is_utf8(scope: Scope) -> bool staticmethod

Whether the raw query string is valid UTF-8.

Source code in src/stac_auth_proxy/middleware/RejectAmbiguousPathMiddleware.py
42
43
44
45
46
47
48
49
@staticmethod
def query_is_utf8(scope: Scope) -> bool:
    """Whether the raw query string is valid UTF-8."""
    try:
        scope.get("query_string", b"").decode()
    except UnicodeDecodeError:
        return False
    return True

url_matches_scope(scope: Scope) -> bool staticmethod

Whether Starlette's request URL has the scope's path.

Source code in src/stac_auth_proxy/middleware/RejectAmbiguousPathMiddleware.py
51
52
53
54
55
56
57
58
59
60
61
62
63
@staticmethod
def url_matches_scope(scope: Scope) -> bool:
    """Whether Starlette's request URL has the scope's path."""
    try:
        # Without the query string, which isn't compared (and may not be UTF-8)
        url = URL(scope={**scope, "query_string": b""})
    except (KeyError, ValueError, TypeError):  # e.g. unknown scheme
        return False
    # Only the path is used by the checks. The query string is forwarded and
    # parsed from scope["query_string"], never request.url.query (which a raw "#"
    # truncates), and the scheme and host may legitimately be missing (e.g. a
    # unix socket with no valid Host header).
    return url.path == scope["path"]