Skip to content

stac_auth_proxy.middleware.RemoveRootPathMiddleware

Middleware to remove ROOT_PATH from incoming requests and update links in responses.

RemoveRootPathMiddleware dataclass

Middleware to remove the root path of the request before the checks and the upstream see it.

IMPORTANT: This middleware must be placed early in the middleware chain (ie late in the order of declaration) so that it trims the root_path from the request path before any middleware that may need to use the request path (e.g. EnforceAuthMiddleware). RestoreRootPathMiddleware puts it back before routing.

Parameters:

Name Type Description Default
app Callable[list, Awaitable[None]]
required
root_path str
''
Source code in src/stac_auth_proxy/middleware/RemoveRootPathMiddleware.py 
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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@dataclass
class RemoveRootPathMiddleware:
    """
    Middleware to remove the root path of the request before the checks and the upstream
    see it.

    IMPORTANT: This middleware must be placed early in the middleware chain (ie late in
    the order of declaration) so that it trims the root_path from the request path before
    any middleware that may need to use the request path (e.g. EnforceAuthMiddleware).
    RestoreRootPathMiddleware puts it back before routing.
    """

    app: ASGIApp
    # ROOT_PATH. When unset, the app's own root path (scope["root_path"], e.g. from
    # FastAPI(root_path=...) or uvicorn --root-path) is removed instead, as
    # Starlette does when routing, so the checks see the path that is routed.
    root_path: str = ""

    def __post_init__(self) -> None:
        """Normalize the root path, as Settings does."""
        self.root_path = self.root_path.rstrip("/")

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """Remove ROOT_PATH from the request path if it exists."""
        if scope["type"] != "http":
            return await self.app(scope, receive, send)

        path = scope["path"]
        app_root_path = scope.get("root_path", "").rstrip("/")
        root_path = self.root_path or app_root_path
        # Only match the root path at a segment boundary ("/stac../x" is not under "/stac")
        under_root_path = bool(root_path) and is_under_prefix(path, root_path)

        # If root_path is set and path isn't under it, return 404
        if self.root_path and not under_root_path:
            response = Response("Not Found", status_code=404)
            logger.error(
                f"Root path {self.root_path!r} not found in path {scope['path']!r}"
            )
            await response(scope, receive, send)
            return

        if under_root_path:
            scope[_ORIGINAL] = (root_path, path, scope.get("raw_path"))
            scope["raw_path"] = path.encode()
            scope["path"] = strip_prefix(path, root_path)

        # "/stac/stac/x" is checked as "/stac/x". An upstream sharing the root path
        # (e.g. stac-fastapi reading the same ROOT_PATH variable) would remove it again
        # and act on "/x", so reject paths still under a root path.
        # Case-insensitively, as some upstreams (Express) route that way
        if any(
            prefix and is_under_prefix(scope["path"].lower(), prefix.lower())
            for prefix in (root_path, app_root_path)
        ):
            response = Response("Not Found", status_code=404)
            # A client error, so not logged at error level
            logger.info("Path %r still starts with a root path", scope["path"])
            await response(scope, receive, send)
            return

        scope[CHECKED_PATH] = scope["path"]
        return await self.app(scope, receive, send)

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

Remove ROOT_PATH from the request path if it exists.

Source code in src/stac_auth_proxy/middleware/RemoveRootPathMiddleware.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
    """Remove ROOT_PATH from the request path if it exists."""
    if scope["type"] != "http":
        return await self.app(scope, receive, send)

    path = scope["path"]
    app_root_path = scope.get("root_path", "").rstrip("/")
    root_path = self.root_path or app_root_path
    # Only match the root path at a segment boundary ("/stac../x" is not under "/stac")
    under_root_path = bool(root_path) and is_under_prefix(path, root_path)

    # If root_path is set and path isn't under it, return 404
    if self.root_path and not under_root_path:
        response = Response("Not Found", status_code=404)
        logger.error(
            f"Root path {self.root_path!r} not found in path {scope['path']!r}"
        )
        await response(scope, receive, send)
        return

    if under_root_path:
        scope[_ORIGINAL] = (root_path, path, scope.get("raw_path"))
        scope["raw_path"] = path.encode()
        scope["path"] = strip_prefix(path, root_path)

    # "/stac/stac/x" is checked as "/stac/x". An upstream sharing the root path
    # (e.g. stac-fastapi reading the same ROOT_PATH variable) would remove it again
    # and act on "/x", so reject paths still under a root path.
    # Case-insensitively, as some upstreams (Express) route that way
    if any(
        prefix and is_under_prefix(scope["path"].lower(), prefix.lower())
        for prefix in (root_path, app_root_path)
    ):
        response = Response("Not Found", status_code=404)
        # A client error, so not logged at error level
        logger.info("Path %r still starts with a root path", scope["path"])
        await response(scope, receive, send)
        return

    scope[CHECKED_PATH] = scope["path"]
    return await self.app(scope, receive, send)

__post_init__() -> None

Normalize the root path, as Settings does.

Source code in src/stac_auth_proxy/middleware/RemoveRootPathMiddleware.py 
36
37
38
def __post_init__(self) -> None:
    """Normalize the root path, as Settings does."""
    self.root_path = self.root_path.rstrip("/")

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 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
@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 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
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()