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

@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

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

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

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

@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."""
        headers = MutableHeaders(request.headers)
        headers.setdefault("Via", f"1.1 {self.proxy_name}")

        proxy_client = request.client.host if request.client else "unknown"
        proxy_proto = request.url.scheme
        proxy_host = request.url.netloc
        proxy_path = request.base_url.path
        headers.setdefault(
            "Forwarded",
            f"for={proxy_client};host={proxy_host};proto={proxy_proto};path={proxy_path}",
        )
        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)

        # 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."""
        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=request.url.path,
                query=request.url.query.encode("utf-8"),
            ),
            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()
        rp_resp = await self.client.send(rp_req, stream=True)
        proxy_time = time.perf_counter() - start_time

        logger.debug(
            f"Received response status {rp_resp.status_code!r} from {rp_req.url} in {proxy_time:.3f}s"
        )
        rp_resp.headers["X-Upstream-Time"] = f"{proxy_time:.3f}"

        # 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

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

async def proxy_request(self, request: Request) -> Response:
    """Proxy a request to the upstream STAC API."""
    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=request.url.path,
            query=request.url.query.encode("utf-8"),
        ),
        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()
    rp_resp = await self.client.send(rp_req, stream=True)
    proxy_time = time.perf_counter() - start_time

    logger.debug(
        f"Received response status {rp_resp.status_code!r} from {rp_req.url} in {proxy_time:.3f}s"
    )
    rp_resp.headers["X-Upstream-Time"] = f"{proxy_time:.3f}"

    # 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

@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

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,
    )