stac_auth_proxy.utils.requests

Utility functions for working with HTTP requests.

MatchResult dataclass

Result of a match between a path and method and a set of endpoints.

Parameters:

Name	Type	Description	Default
uses_auth	bool		required
required_scopes	Sequence[str]	Built-in mutable sequence. If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.	<dynamic>

Source code in src/stac_auth_proxy/utils/requests.py

@dataclass
class MatchResult:
    """Result of a match between a path and method and a set of endpoints."""

    uses_auth: bool
    required_scopes: Sequence[str] = field(default_factory=list)

build_server_timing_header(current_value: Optional[str] = None, *, name: str, desc: str, dur: float)

Append a timing header to headers.

Source code in src/stac_auth_proxy/utils/requests.py

def build_server_timing_header(
    current_value: Optional[str] = None, *, name: str, desc: str, dur: float
):
    """Append a timing header to headers."""
    metric = f'{name};desc="{desc}";dur={dur:.3f}'
    if current_value:
        return f"{current_value}, {metric}"
    return metric

checked_path(scope: dict) -> str

Return the path that auth, filter and transaction checks matched on.

Recorded once rather than re-derived from root_path, which Starlette also extends for Mounts: a proxy mounted under "/proxy" must forward "/proxy/x", the path that was checked, not "/x".

Source code in src/stac_auth_proxy/utils/requests.py

def checked_path(scope: dict) -> str:
    """
    Return the path that auth, filter and transaction checks matched on.

    Recorded once rather than re-derived from root_path, which Starlette also
    extends for Mounts: a proxy mounted under "/proxy" must forward "/proxy/x",
    the path that was checked, not "/x".
    """
    return scope.get(CHECKED_PATH, scope["path"])

dict_to_bytes(d: dict) -> bytes

Convert a dictionary to a body.

Source code in src/stac_auth_proxy/utils/requests.py

def dict_to_bytes(d: dict) -> bytes:
    """Convert a dictionary to a body."""
    return json.dumps(d, separators=(",", ":")).encode("utf-8")

extract_variables(url: str) -> dict

Extract variables from a URL path. Being that we use a catch-all endpoint for the proxy, we can't rely on the path parameters that FastAPI provides.

Source code in src/stac_auth_proxy/utils/requests.py

def extract_variables(url: str) -> dict:
    """
    Extract variables from a URL path. Being that we use a catch-all endpoint for the proxy,
    we can't rely on the path parameters that FastAPI provides.
    """
    path = urlparse(url).path
    # This allows either /queryables or /items or /bulk_items, with an optional item_id following.
    pattern = r"^/collections/(?P<collection_id>[^/]+)(?:/(?:items|bulk_items|queryables)(?:/(?P<item_id>[^/]+))?)?/?$"
    match = match_path(pattern, path)
    return {k: v for k, v in match.groupdict().items() if v} if match else {}

find_match(path: str, method: str, private_endpoints: EndpointMethods, public_endpoints: EndpointMethods, default_public: bool, items_filter_path: Optional[str] = None, collections_filter_path: Optional[str] = None) -> MatchResult

Check if the given path and method match any of the regex patterns and methods in the endpoints.

Some upstreams (e.g. stac-server/Express) route case-insensitively, so private endpoints and filter paths match case-insensitively. Public endpoints stay case-sensitive so that a case variation can only ever require more auth.

Source code in src/stac_auth_proxy/utils/requests.py

def find_match(
    path: str,
    method: str,
    private_endpoints: EndpointMethods,
    public_endpoints: EndpointMethods,
    default_public: bool,
    items_filter_path: Optional[str] = None,
    collections_filter_path: Optional[str] = None,
) -> "MatchResult":
    """
    Check if the given path and method match any of the regex patterns and methods in the endpoints.

    Some upstreams (e.g. stac-server/Express) route case-insensitively, so private
    endpoints and filter paths match case-insensitively. Public endpoints stay
    case-sensitive so that a case variation can only ever require more auth.
    """
    primary_endpoints = private_endpoints if default_public else public_endpoints
    matched, required_scopes = _check_endpoint_match(
        path, method, primary_endpoints, widen=default_public
    )
    if matched:
        return MatchResult(
            uses_auth=default_public,
            required_scopes=required_scopes,
        )

    # If we have filter paths configured, check those as well (these are always considered to use auth if they match, regardless of default_public)
    for filter_path in [items_filter_path, collections_filter_path]:
        if filter_path and match_path(filter_path, path):
            # With default_public, private endpoints were already checked above
            required_scopes = (
                []
                if default_public
                else _check_endpoint_match(path, method, private_endpoints)[1]
            )
            return MatchResult(uses_auth=True, required_scopes=required_scopes)

    # If default_public and no match found in private_endpoints, it's public
    if default_public:
        return MatchResult(uses_auth=False)

    # If not default_public, check private_endpoints for required scopes
    matched, required_scopes = _check_endpoint_match(path, method, private_endpoints)
    if matched:
        return MatchResult(uses_auth=True, required_scopes=required_scopes)

    # Default case: if not default_public and no explicit match, it's private
    return MatchResult(uses_auth=True)

get_base_url(request: Request) -> str

Get the request's base URL, accounting for forwarded headers from load balancers/proxies.

This function handles both the standard Forwarded header (RFC 7239) and legacy X-Forwarded-* headers to reconstruct the original client URL when the service is deployed behind load balancers or reverse proxies.

Parameters:

Name	Type	Description	Default
request	Request	The Starlette request object	required

Returns:

Type	Description
str	The reconstructed client base URL

Example

With Forwarded header

request.headers = {"Forwarded": "for=192.0.2.43; proto=https; host=api.example.com"} get_base_url(request) "api.example.com/"

With X-Forwarded-* headers

request.headers = {"X-Forwarded-Host": "api.example.com", "X-Forwarded-Proto": "https"} get_base_url(request) "api.example.com/"

Source code in src/stac_auth_proxy/utils/requests.py

def get_base_url(request: Request) -> str:
    """
    Get the request's base URL, accounting for forwarded headers from load balancers/proxies.

    This function handles both the standard Forwarded header (RFC 7239) and legacy
    X-Forwarded-* headers to reconstruct the original client URL when the service
    is deployed behind load balancers or reverse proxies.

    Args:
        request: The Starlette request object

    Returns:
        The reconstructed client base URL

    Example:
        >>> # With Forwarded header
        >>> request.headers = {"Forwarded": "for=192.0.2.43; proto=https; host=api.example.com"}
        >>> get_base_url(request)
        "https://api.example.com/"

        >>> # With X-Forwarded-* headers
        >>> request.headers = {"X-Forwarded-Host": "api.example.com", "X-Forwarded-Proto": "https"}
        >>> get_base_url(request)
        "https://api.example.com/"

    """
    # Check for standard Forwarded header first (RFC 7239)
    forwarded_header = request.headers.get("Forwarded")
    if forwarded_header:
        try:
            forwarded_info = parse_forwarded_header(forwarded_header)
            # Only use Forwarded header if we successfully parsed it and got useful info
            if forwarded_info and (
                "proto" in forwarded_info or "host" in forwarded_info
            ):
                scheme = forwarded_info.get("proto", request.url.scheme)
                host = forwarded_info.get("host", request.url.netloc)
                # Note: Forwarded header doesn't include path, so we use request.base_url.path
                path = request.base_url.path
                return f"{scheme}://{host}{path}"
        except Exception as e:
            logger.warning(f"Failed to parse Forwarded header: {e}")

    # Fall back to legacy X-Forwarded-* headers
    forwarded_host = request.headers.get("X-Forwarded-Host")
    forwarded_proto = request.headers.get("X-Forwarded-Proto")
    forwarded_path = request.headers.get("X-Forwarded-Path")

    if forwarded_host:
        # Use forwarded headers to reconstruct the original client URL
        scheme = forwarded_proto or request.url.scheme
        netloc = forwarded_host
        # Use forwarded path if available, otherwise use request base URL path
        path = forwarded_path or request.base_url.path
    else:
        # Fall back to the request's base URL if no forwarded headers
        scheme = request.url.scheme
        netloc = request.url.netloc
        path = request.base_url.path

    return f"{scheme}://{netloc}{path}"

is_under_prefix(path: str, prefix: str) -> bool

Whether path is prefix or below it, at a segment boundary ("/stacx" isn't under "/stac").

Source code in src/stac_auth_proxy/utils/requests.py

def is_under_prefix(path: str, prefix: str) -> bool:
    """Whether path is prefix or below it, at a segment boundary ("/stacx" isn't under "/stac")."""
    prefix = prefix.rstrip("/")
    return path == prefix or path.startswith(f"{prefix}/")

match_path(pattern: str, path: str, widen: bool = True) -> Optional[re.Match[str]]

Match a path-based access rule against the path an upstream may route as.

Some upstreams route case-insensitively (stac-server/Express) or ignore a trailing slash (Express, pygeoapi). So a rule matches if it matches the path as given or, with widen, ignoring case or without one trailing slash. Each variation only adds matches: e.g. "^/admin/" still matches "/admin/", and "(?!public-)" still matches "PUBLIC-x" as written. Rules that grant access (public endpoints) must not be widened.

Source code in src/stac_auth_proxy/utils/requests.py

def match_path(pattern: str, path: str, widen: bool = True) -> Optional[re.Match[str]]:
    """
    Match a path-based access rule against the path an upstream may route as.

    Some upstreams route case-insensitively (stac-server/Express) or ignore a
    trailing slash (Express, pygeoapi). So a rule matches if it matches the path as
    given or, with ``widen``, ignoring case or without one trailing slash. Each
    variation only adds matches: e.g. "^/admin/" still matches "/admin/", and
    "(?!public-)" still matches "PUBLIC-x" as written. Rules that grant access
    (public endpoints) must not be widened.
    """
    if not widen:
        return re.match(pattern, path)
    paths = [path]
    if len(path) > 1 and path.endswith("/"):
        paths.append(path[:-1])
    for flags in (0, re.IGNORECASE):
        for p in paths:
            if match := re.match(pattern, p, flags):
                return match
    return None

parse_forwarded_header(forwarded_header: str) -> Dict[str, str]

Parse the Forwarded header according to RFC 7239.

Parameters:

Name	Type	Description	Default
forwarded_header	str	The Forwarded header value	required

Returns:

Type	Description
Dict[str, str]	Dictionary containing parsed forwarded information (proto, host, for, by, etc.)

Example

parse_forwarded_header("for=192.0.2.43; by=203.0.113.60; proto=https; host=api.example.com")

Source code in src/stac_auth_proxy/utils/requests.py

def parse_forwarded_header(forwarded_header: str) -> Dict[str, str]:
    """
    Parse the Forwarded header according to RFC 7239.

    Args:
        forwarded_header: The Forwarded header value

    Returns:
        Dictionary containing parsed forwarded information (proto, host, for, by, etc.)

    Example:
        >>> parse_forwarded_header("for=192.0.2.43; by=203.0.113.60; proto=https; host=api.example.com")
        {'for': '192.0.2.43', 'by': '203.0.113.60', 'proto': 'https', 'host': 'api.example.com'}

    """
    # Forwarded header format: "for=192.0.2.43, for=198.51.100.17; by=203.0.113.60; proto=https; host=example.com"
    # The format is: for=value1, for=value2; by=value; proto=value; host=value
    # We need to parse all the key=value pairs, taking the first 'for' value
    forwarded_info = {}

    try:
        # Parse all key=value pairs separated by semicolons
        for pair in forwarded_header.split(";"):
            pair = pair.strip()
            if "=" in pair:
                key, value = pair.split("=", 1)
                key = key.strip()
                value = value.strip().strip('"')

                # For 'for' field, only take the first value if there are multiple
                if key == "for" and key not in forwarded_info:
                    # Extract the first for value (before comma if present)
                    first_for_value = value.split(",")[0].strip()
                    forwarded_info[key] = first_for_value
                elif key != "for":
                    # For other fields, just use the value as-is
                    forwarded_info[key] = value
    except Exception as e:
        logger.warning(f"Failed to parse Forwarded header '{forwarded_header}': {e}")
        return {}

    return forwarded_info

strip_prefix(path: str, prefix: str) -> str

Remove a root path prefix from path, at a segment boundary, if path is under it.

Source code in src/stac_auth_proxy/utils/requests.py

def strip_prefix(path: str, prefix: str) -> str:
    """Remove a root path prefix from path, at a segment boundary, if path is under it."""
    prefix = prefix.rstrip("/")
    if prefix and is_under_prefix(path, prefix):
        return path[len(prefix) :] or "/"
    return path