Skip to content

stac_auth_proxy.handlers

Handlers to process requests.

HealthzHandler dataclass

Handler for health check endpoints.

Parameters:

Name Type Description Default
upstream_url str
required

Attributes:

Name Type Description
router APIRouter
Source code in src/stac_auth_proxy/handlers/healthz.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@dataclass
class HealthzHandler:
    """Handler for health check endpoints."""

    upstream_url: str
    router: APIRouter = field(init=False)

    def __post_init__(self):
        """Initialize the router."""
        self.router = APIRouter()
        self.router.add_api_route("", self.healthz, methods=["GET"])
        self.router.add_api_route("/upstream", self.healthz_upstream, methods=["GET"])

    async def healthz(self):
        """Return health of this API."""
        return {"status": "ok"}

    async def healthz_upstream(self):
        """Return health of upstream STAC API."""
        async with AsyncClient() as client:
            response = await client.get(self.upstream_url)
            response.raise_for_status()
            return {"status": "ok", "code": response.status_code}

__post_init__()

Initialize the router.

Source code in src/stac_auth_proxy/handlers/healthz.py 
16
17
18
19
20
def __post_init__(self):
    """Initialize the router."""
    self.router = APIRouter()
    self.router.add_api_route("", self.healthz, methods=["GET"])
    self.router.add_api_route("/upstream", self.healthz_upstream, methods=["GET"])

healthz() async

Return health of this API.

Source code in src/stac_auth_proxy/handlers/healthz.py 
22
23
24
async def healthz(self):
    """Return health of this API."""
    return {"status": "ok"}

healthz_upstream() async

Return health of upstream STAC API.

Source code in src/stac_auth_proxy/handlers/healthz.py 
26
27
28
29
30
31
async def healthz_upstream(self):
    """Return health of upstream STAC API."""
    async with AsyncClient() as client:
        response = await client.get(self.upstream_url)
        response.raise_for_status()
        return {"status": "ok", "code": response.status_code}

ReverseProxyHandler dataclass

Reverse proxy functionality.

Parameters:

Name Type Description Default
upstream str
required
client AsyncClient
None
timeout Timeout
Timeout(timeout=15.0)
proxy_name str
'stac-auth-proxy'
override_host bool
True
legacy_forwarded_headers bool
False
Source code in src/stac_auth_proxy/handlers/reverse_proxy.py 
 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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
@dataclass
class ReverseProxyHandler:
    """Reverse proxy functionality."""

    upstream: str
    client: httpx.AsyncClient = None
    timeout: httpx.Timeout = field(default_factory=lambda: httpx.Timeout(timeout=15.0))

    proxy_name: str = "stac-auth-proxy"
    override_host: bool = True
    legacy_forwarded_headers: bool = False

    def __post_init__(self):
        """Initialize the HTTP client."""
        self.client = self.client or httpx.AsyncClient(
            base_url=self.upstream,
            timeout=self.timeout,
            http2=True,
        )

    def _prepare_headers(self, request: Request) -> MutableHeaders:
        """
        Prepare headers for the proxied request. Construct a Forwarded header to inform
        the upstream API about the original request context, which will allow it to
        properly construct URLs in responses (namely, in the Links). If there are
        existing X-Forwarded-*/Forwarded headers (typically, in situations where the
        STAC Auth Proxy is behind a proxy like Traefik or NGINX), we use those values.
        """
        headers = MutableHeaders(request.headers)
        headers.setdefault("Via", f"1.1 {self.proxy_name}")

        proxy_client = headers.get(
            "X-Forwarded-For", request.client.host if request.client else "unknown"
        )
        proxy_proto = headers.get("X-Forwarded-Proto", request.url.scheme)
        proxy_host = headers.get("X-Forwarded-Host", request.url.netloc)
        proxy_port = str(headers.get("X-Forwarded-Port", request.url.port))
        proxy_path = headers.get("X-Forwarded-Path", request.base_url.path)

        # NOTE: If we don't include a port, it's possible that the upstream server may
        # mistakenly use the port from the Host header (which may be the internal port
        # of the upstream server) when constructing URLs.
        forwarded_host = proxy_host
        if proxy_port:
            forwarded_host = f"{forwarded_host}:{proxy_port}"

        headers.setdefault(
            "Forwarded",
            f"for={proxy_client};host={forwarded_host};proto={proxy_proto};path={proxy_path}",
        )

        # NOTE: This is useful if the upstream API does not support the Forwarded header
        # and there were no existing X-Forwarded-* headers on the incoming request.
        if self.legacy_forwarded_headers:
            headers.setdefault("X-Forwarded-For", proxy_client)
            headers.setdefault("X-Forwarded-Host", proxy_host)
            headers.setdefault("X-Forwarded-Path", proxy_path)
            headers.setdefault("X-Forwarded-Proto", proxy_proto)
            headers.setdefault("X-Forwarded-Port", proxy_port)

        # Set host to the upstream host
        if self.override_host:
            headers["Host"] = self.client.base_url.netloc.decode("utf-8")

        return headers

    async def proxy_request(self, request: Request) -> Response:
        """Proxy a request to the upstream STAC API."""
        path = checked_path(request.scope)
        if not is_forwardable(path):
            return bad_request("Invalid request path.")
        headers = self._prepare_headers(request)

        # https://github.com/fastapi/fastapi/discussions/7382#discussioncomment-5136466
        rp_req = self.client.build_request(
            request.method,
            url=httpx.URL(
                path=path,
                # The raw query string, as filter factories saw it: request.url.query
                # is truncated at a raw "#". Escape only what httpx rejects ("#",
                # space, non-ASCII), leaving existing escapes and delimiters as-is.
                query=quote_from_bytes(
                    request.scope.get("query_string", b""),
                    safe="!$&'()*+,;=:@/?%[]",
                ).encode("ascii"),
            ),
            headers=headers,
            content=request.stream(),
        )

        # NOTE: HTTPX adds headers, so we need to trim them before sending request
        for h in rp_req.headers:
            if h not in headers:
                del rp_req.headers[h]

        logger.debug(f"Proxying request to {rp_req.url}")

        start_time = time.perf_counter()
        try:
            rp_resp = await self.client.send(rp_req, stream=True)
        except httpx.TimeoutException:
            return Response(status_code=504, content=b"Upstream timed out")
        except httpx.ConnectError:
            return Response(status_code=502, content=b"Upstream unreachable")
        proxy_time = time.perf_counter() - start_time
        rp_resp.headers["Server-Timing"] = build_server_timing_header(
            rp_resp.headers.get("Server-Timing"),
            name="upstream",
            dur=proxy_time,
            desc="Upstream processing time",
        )
        logger.debug(
            f"Received response status {rp_resp.status_code!r} from {rp_req.url} in {proxy_time:.3f}s"
        )

        # We read the content here to make use of HTTPX's decompression, ensuring we have
        # non-compressed content for the middleware to work with.
        content = await rp_resp.aread()
        if rp_resp.headers.get("Content-Encoding"):
            del rp_resp.headers["Content-Encoding"]

        return Response(
            content=content,
            status_code=rp_resp.status_code,
            headers=dict(rp_resp.headers),
        )

__post_init__()

Initialize the HTTP client.

Source code in src/stac_auth_proxy/handlers/reverse_proxy.py 
60
61
62
63
64
65
66
def __post_init__(self):
    """Initialize the HTTP client."""
    self.client = self.client or httpx.AsyncClient(
        base_url=self.upstream,
        timeout=self.timeout,
        http2=True,
    )

proxy_request(request: Request) -> Response async

Proxy a request to the upstream STAC API.

Source code in src/stac_auth_proxy/handlers/reverse_proxy.py 
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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
async def proxy_request(self, request: Request) -> Response:
    """Proxy a request to the upstream STAC API."""
    path = checked_path(request.scope)
    if not is_forwardable(path):
        return bad_request("Invalid request path.")
    headers = self._prepare_headers(request)

    # https://github.com/fastapi/fastapi/discussions/7382#discussioncomment-5136466
    rp_req = self.client.build_request(
        request.method,
        url=httpx.URL(
            path=path,
            # The raw query string, as filter factories saw it: request.url.query
            # is truncated at a raw "#". Escape only what httpx rejects ("#",
            # space, non-ASCII), leaving existing escapes and delimiters as-is.
            query=quote_from_bytes(
                request.scope.get("query_string", b""),
                safe="!$&'()*+,;=:@/?%[]",
            ).encode("ascii"),
        ),
        headers=headers,
        content=request.stream(),
    )

    # NOTE: HTTPX adds headers, so we need to trim them before sending request
    for h in rp_req.headers:
        if h not in headers:
            del rp_req.headers[h]

    logger.debug(f"Proxying request to {rp_req.url}")

    start_time = time.perf_counter()
    try:
        rp_resp = await self.client.send(rp_req, stream=True)
    except httpx.TimeoutException:
        return Response(status_code=504, content=b"Upstream timed out")
    except httpx.ConnectError:
        return Response(status_code=502, content=b"Upstream unreachable")
    proxy_time = time.perf_counter() - start_time
    rp_resp.headers["Server-Timing"] = build_server_timing_header(
        rp_resp.headers.get("Server-Timing"),
        name="upstream",
        dur=proxy_time,
        desc="Upstream processing time",
    )
    logger.debug(
        f"Received response status {rp_resp.status_code!r} from {rp_req.url} in {proxy_time:.3f}s"
    )

    # We read the content here to make use of HTTPX's decompression, ensuring we have
    # non-compressed content for the middleware to work with.
    content = await rp_resp.aread()
    if rp_resp.headers.get("Content-Encoding"):
        del rp_resp.headers["Content-Encoding"]

    return Response(
        content=content,
        status_code=rp_resp.status_code,
        headers=dict(rp_resp.headers),
    )

SwaggerUI dataclass

Swagger UI handler.

Parameters:

Name Type Description Default
openapi_url str
required
title str | None
'STAC API'
init_oauth dict

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

 <class 'dict'>
parameters dict

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

 <class 'dict'>
oauth2_redirect_url str
'/docs/oauth2-redirect'
Source code in src/stac_auth_proxy/handlers/swagger_ui.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
@dataclass
class SwaggerUI:
    """Swagger UI handler."""

    openapi_url: str
    title: Optional[str] = "STAC API"
    init_oauth: dict = field(default_factory=dict)
    parameters: dict = field(default_factory=dict)
    oauth2_redirect_url: str = "/docs/oauth2-redirect"

    async def route(self, req: Request) -> HTMLResponse:
        """Route handler."""
        root_path = req.scope.get("root_path", "").rstrip("/")
        openapi_url = root_path + self.openapi_url
        oauth2_redirect_url = self.oauth2_redirect_url
        if oauth2_redirect_url:
            oauth2_redirect_url = root_path + oauth2_redirect_url
        return get_swagger_ui_html(
            openapi_url=openapi_url,
            title=f"{self.title} - Swagger UI",
            oauth2_redirect_url=oauth2_redirect_url,
            init_oauth=self.init_oauth,
            swagger_ui_parameters=self.parameters,
        )

route(req: Request) -> HTMLResponse async

Route handler.

Source code in src/stac_auth_proxy/handlers/swagger_ui.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
async def route(self, req: Request) -> HTMLResponse:
    """Route handler."""
    root_path = req.scope.get("root_path", "").rstrip("/")
    openapi_url = root_path + self.openapi_url
    oauth2_redirect_url = self.oauth2_redirect_url
    if oauth2_redirect_url:
        oauth2_redirect_url = root_path + oauth2_redirect_url
    return get_swagger_ui_html(
        openapi_url=openapi_url,
        title=f"{self.title} - Swagger UI",
        oauth2_redirect_url=oauth2_redirect_url,
        init_oauth=self.init_oauth,
        swagger_ui_parameters=self.parameters,
    )