Skip to content

stac_auth_proxy.middleware.ProcessLinksMiddleware

Middleware to remove the application root path from incoming requests and update links in responses.

ProcessLinksMiddleware dataclass

Bases: JsonResponseMiddleware

Middleware to update links in responses, removing the upstream_url path and adding the root_path if it exists.

Parameters:

Name Type Description Default
app Callable[list, Awaitable[None]]
required
upstream_url str
required
root_path str | None
None
json_content_type_expr str
'application/(geo\\+)?json'
root_path_skip_prefixes Sequence[str]
()
Source code in src/stac_auth_proxy/middleware/ProcessLinksMiddleware.py 
 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
 81
 82
 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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@dataclass
class ProcessLinksMiddleware(JsonResponseMiddleware):
    """
    Middleware to update links in responses, removing the upstream_url path and adding
    the root_path if it exists.
    """

    app: ASGIApp
    upstream_url: str
    root_path: Optional[str] = None

    json_content_type_expr: str = r"application/(geo\+)?json"
    root_path_skip_prefixes: Sequence[str] = ()

    def __post_init__(self) -> None:
        """Normalize skip prefixes, dropping empty entries and trailing slashes."""
        self.root_path_skip_prefixes = tuple(
            prefix.rstrip("/")
            for prefix in self.root_path_skip_prefixes
            if prefix.rstrip("/")
        )

    def should_transform_response(self, request: Request, scope: Scope) -> bool:
        """Only transform responses with JSON content type."""
        return bool(
            re.match(
                self.json_content_type_expr,
                Headers(scope=scope).get("content-type", ""),
            )
        )

    def transform_json(self, data: dict[str, Any], request: Request) -> dict[str, Any]:
        """Update links in the response to include root_path."""
        # Get the client's actual base URL (accounting for load balancers/proxies)
        req_base_url = get_base_url(request)
        parsed_req_url = urlparse(req_base_url)
        parsed_upstream_url = urlparse(self.upstream_url)

        for link in get_links(data):
            try:
                self._update_link(link, parsed_req_url, parsed_upstream_url)
            except Exception as e:
                logger.error(
                    "Failed to parse link href %r, (ignoring): %s",
                    link.get("href"),
                    str(e),
                )
        return data

    def _matches_skip_prefix(self, path: str) -> bool:
        """Return whether path is exactly a skip prefix or under one of them."""
        return any(
            is_under_prefix(path, prefix) for prefix in self.root_path_skip_prefixes
        )

    def _update_link(
        self, link: dict[str, Any], request_url: ParseResult, upstream_url: ParseResult
    ) -> None:
        """
        Ensure that link hrefs that are local to upstream url are rewritten as local to
        the proxy.
        """
        if "href" not in link:
            logger.warning("Link %r has no href", link)
            return

        parsed_link = urlparse(link["href"])

        if parsed_link.netloc not in [
            request_url.netloc,
            upstream_url.netloc,
        ]:
            logger.debug(
                "Ignoring link %s because it is not for an endpoint behind this proxy (%s or %s)",
                link["href"],
                request_url.netloc,
                upstream_url.netloc,
            )
            return

        # Skip links outside the upstream path, unless they match a skip prefix
        # (those still need host rewrite, just not root_path).
        if upstream_url.path != "/" and not is_under_prefix(
            parsed_link.path, upstream_url.path
        ):
            if not self._matches_skip_prefix(parsed_link.path):
                logger.debug(
                    "Ignoring link %s because it is not descendant of upstream path (%s)",
                    link["href"],
                    upstream_url.path,
                )
                return

        # Replace the upstream host with the client's host
        if parsed_link.netloc == upstream_url.netloc:
            parsed_link = parsed_link._replace(netloc=request_url.netloc)._replace(
                scheme=request_url.scheme
            )

        # Remove the upstream prefix from the link path
        if upstream_url.path != "/" and is_under_prefix(
            parsed_link.path, upstream_url.path
        ):
            parsed_link = parsed_link._replace(
                path=parsed_link.path[len(upstream_url.path) :]
            )

        # Add root_path unless this link is for another app on the same host
        # (listed in root_path_skip_prefixes), or it already has root_path.
        path = parsed_link.path
        if (
            self.root_path
            and not self._matches_skip_prefix(path)
            and not is_under_prefix(path, self.root_path)
        ):
            parsed_link = parsed_link._replace(path=f"{self.root_path}{path}")

        updated_href = urlunparse(parsed_link)
        if updated_href == link["href"]:
            return

        logger.debug(
            "Rewriting %r link %r to %r",
            link.get("rel"),
            link["href"],
            updated_href,
        )

        link["href"] = updated_href

__post_init__() -> None

Normalize skip prefixes, dropping empty entries and trailing slashes.

Source code in src/stac_auth_proxy/middleware/ProcessLinksMiddleware.py 
34
35
36
37
38
39
40
def __post_init__(self) -> None:
    """Normalize skip prefixes, dropping empty entries and trailing slashes."""
    self.root_path_skip_prefixes = tuple(
        prefix.rstrip("/")
        for prefix in self.root_path_skip_prefixes
        if prefix.rstrip("/")
    )

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

Only transform responses with JSON content type.

Source code in src/stac_auth_proxy/middleware/ProcessLinksMiddleware.py 
42
43
44
45
46
47
48
49
def should_transform_response(self, request: Request, scope: Scope) -> bool:
    """Only transform responses with JSON content type."""
    return bool(
        re.match(
            self.json_content_type_expr,
            Headers(scope=scope).get("content-type", ""),
        )
    )

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

Update links in the response to include root_path.

Source code in src/stac_auth_proxy/middleware/ProcessLinksMiddleware.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def transform_json(self, data: dict[str, Any], request: Request) -> dict[str, Any]:
    """Update links in the response to include root_path."""
    # Get the client's actual base URL (accounting for load balancers/proxies)
    req_base_url = get_base_url(request)
    parsed_req_url = urlparse(req_base_url)
    parsed_upstream_url = urlparse(self.upstream_url)

    for link in get_links(data):
        try:
            self._update_link(link, parsed_req_url, parsed_upstream_url)
        except Exception as e:
            logger.error(
                "Failed to parse link href %r, (ignoring): %s",
                link.get("href"),
                str(e),
            )
    return data