stac_auth_proxy

STAC Auth Proxy package.

This package contains the components for the STAC authentication and proxying system. It includes FastAPI routes for handling authentication, authorization, and interaction with some internal STAC API.

Settings

Bases: BaseSettings

Configuration settings for the STAC Auth Proxy.

Parameters:

Name	Type	Description	Default
upstream_url	HttpUrl		required
oidc_discovery_url	HttpUrl		required
oidc_discovery_internal_url	HttpUrl		required
root_path	str		''
override_host	bool		True
healthz_prefix	str		'/healthz'
wait_for_upstream	bool		True
check_conformance	bool		True
enable_compression	bool		True
openapi_spec_endpoint	str | None		None
openapi_auth_scheme_name	str		'oidcAuth'
openapi_auth_scheme_override	dict | None		None
swagger_ui_endpoint	str | None		None
swagger_ui_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'>
enable_authentication_extension	bool		True
default_public	bool		False
public_endpoints	dict[str, Sequence[Literal['GET', 'POST', 'PUT', 'DELETE', 'PATCH']]]		{'^/$': ['GET'], '^/api.html$': ['GET'], '^/api$': ['GET'], '^/docs/oauth2-redirect': ['GET'], '^/healthz': ['GET']}
private_endpoints	dict[str, Sequence[Union[Literal['GET', 'POST', 'PUT', 'DELETE', 'PATCH'], tuple[Literal['GET', 'POST', 'PUT', 'DELETE', 'PATCH'], str]]]]		{'^/collections$': ['POST'], '^/collections/([^/]+)$': ['PUT', 'PATCH', 'DELETE'], '^/collections/([^/]+)/items$': ['POST'], '^/collections/([^/]+)/items/([^/]+)$': ['PUT', 'PATCH', 'DELETE'], '^/collections/([^/]+)/bulk_items$': ['POST']}
items_filter	_ClassInput | None		None
items_filter_path	str		'^(/collections/([^/]+)/items(/[^/]+)?$|/search$)'
collections_filter	_ClassInput | None		None
collections_filter_path	str		'^/collections(/[^/]+)?$'

Source code in src/stac_auth_proxy/config.py

class Settings(BaseSettings):
    """Configuration settings for the STAC Auth Proxy."""

    # External URLs
    upstream_url: HttpUrl
    oidc_discovery_url: HttpUrl
    oidc_discovery_internal_url: HttpUrl

    root_path: str = ""
    override_host: bool = True
    healthz_prefix: str = Field(pattern=_PREFIX_PATTERN, default="/healthz")
    wait_for_upstream: bool = True
    check_conformance: bool = True
    enable_compression: bool = True

    # OpenAPI / Swagger UI
    openapi_spec_endpoint: Optional[str] = Field(pattern=_PREFIX_PATTERN, default=None)
    openapi_auth_scheme_name: str = "oidcAuth"
    openapi_auth_scheme_override: Optional[dict] = None
    swagger_ui_endpoint: Optional[str] = None
    swagger_ui_init_oauth: dict = Field(default_factory=dict)

    # Auth
    enable_authentication_extension: bool = True
    default_public: bool = False
    public_endpoints: EndpointMethods = {
        r"^/$": ["GET"],
        r"^/api.html$": ["GET"],
        r"^/api$": ["GET"],
        r"^/docs/oauth2-redirect": ["GET"],
        r"^/healthz": ["GET"],
    }
    private_endpoints: EndpointMethodsWithScope = {
        # https://github.com/stac-api-extensions/collection-transaction/blob/v1.0.0-beta.1/README.md#methods
        r"^/collections$": ["POST"],
        r"^/collections/([^/]+)$": ["PUT", "PATCH", "DELETE"],
        # https://github.com/stac-api-extensions/transaction/blob/v1.0.0-rc.3/README.md#methods
        r"^/collections/([^/]+)/items$": ["POST"],
        r"^/collections/([^/]+)/items/([^/]+)$": ["PUT", "PATCH", "DELETE"],
        # https://stac-utils.github.io/stac-fastapi/api/stac_fastapi/extensions/third_party/bulk_transactions/#bulktransactionextension
        r"^/collections/([^/]+)/bulk_items$": ["POST"],
    }

    # Filters
    items_filter: Optional[_ClassInput] = None
    items_filter_path: str = r"^(/collections/([^/]+)/items(/[^/]+)?$|/search$)"
    collections_filter: Optional[_ClassInput] = None
    collections_filter_path: str = r"^/collections(/[^/]+)?$"

    model_config = SettingsConfigDict(
        env_nested_delimiter="_",
    )

    @model_validator(mode="before")
    @classmethod
    def _default_oidc_discovery_internal_url(cls, data: Any) -> Any:
        """Set the internal OIDC discovery URL to the public URL if not set."""
        if not data.get("oidc_discovery_internal_url"):
            data["oidc_discovery_internal_url"] = data.get("oidc_discovery_url")
        return data

create_app(settings: Optional[Settings] = None) -> FastAPI

FastAPI Application Factory.

Source code in src/stac_auth_proxy/app.py

def create_app(settings: Optional[Settings] = None) -> FastAPI:
    """FastAPI Application Factory."""
    settings = settings or Settings()

    #
    # Application
    #

    @asynccontextmanager
    async def lifespan(app: FastAPI):
        assert settings

        # Wait for upstream servers to become available
        if settings.wait_for_upstream:
            logger.info("Running upstream server health checks...")
            urls = [settings.upstream_url, settings.oidc_discovery_internal_url]
            for url in urls:
                await check_server_health(url=url)
            logger.info(
                "Upstream servers are healthy:\n%s",
                "\n".join([f" - {url}" for url in urls]),
            )

        # Log all middleware connected to the app
        logger.info(
            "Connected middleware:\n%s",
            "\n".join([f" - {m.cls.__name__}" for m in app.user_middleware]),
        )

        if settings.check_conformance:
            await check_conformance(
                app.user_middleware,
                str(settings.upstream_url),
            )

        yield

    app = FastAPI(
        openapi_url=None,  # Disable OpenAPI schema endpoint, we want to serve upstream's schema
        lifespan=lifespan,
        root_path=settings.root_path,
    )
    if app.root_path:
        logger.debug("Mounted app at %s", app.root_path)

    #
    # Handlers (place catch-all proxy handler last)
    #

    if settings.swagger_ui_endpoint:
        assert (
            settings.openapi_spec_endpoint
        ), "openapi_spec_endpoint must be set when using swagger_ui_endpoint"
        app.add_route(
            settings.swagger_ui_endpoint,
            SwaggerUI(
                openapi_url=settings.openapi_spec_endpoint,
                init_oauth=settings.swagger_ui_init_oauth,
            ).route,
            include_in_schema=False,
        )
    if settings.healthz_prefix:
        app.include_router(
            HealthzHandler(upstream_url=str(settings.upstream_url)).router,
            prefix=settings.healthz_prefix,
        )

    app.add_api_route(
        "/{path:path}",
        ReverseProxyHandler(
            upstream=str(settings.upstream_url),
            override_host=settings.override_host,
        ).proxy_request,
        methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
    )

    #
    # Middleware (order is important, last added = first to run)
    #

    if settings.enable_authentication_extension:
        app.add_middleware(
            AuthenticationExtensionMiddleware,
            default_public=settings.default_public,
            public_endpoints=settings.public_endpoints,
            private_endpoints=settings.private_endpoints,
            oidc_discovery_url=str(settings.oidc_discovery_url),
        )

    if settings.openapi_spec_endpoint:
        app.add_middleware(
            OpenApiMiddleware,
            openapi_spec_path=settings.openapi_spec_endpoint,
            oidc_discovery_url=str(settings.oidc_discovery_url),
            public_endpoints=settings.public_endpoints,
            private_endpoints=settings.private_endpoints,
            default_public=settings.default_public,
            root_path=settings.root_path,
            auth_scheme_name=settings.openapi_auth_scheme_name,
            auth_scheme_override=settings.openapi_auth_scheme_override,
        )

    if settings.items_filter or settings.collections_filter:
        app.add_middleware(Cql2ValidateResponseBodyMiddleware)
        app.add_middleware(Cql2ApplyFilterBodyMiddleware)
        app.add_middleware(Cql2ApplyFilterQueryStringMiddleware)
        app.add_middleware(
            Cql2BuildFilterMiddleware,
            items_filter=settings.items_filter() if settings.items_filter else None,
            collections_filter=(
                settings.collections_filter() if settings.collections_filter else None
            ),
            collections_filter_path=settings.collections_filter_path,
            items_filter_path=settings.items_filter_path,
        )

    app.add_middleware(
        AddProcessTimeHeaderMiddleware,
    )

    app.add_middleware(
        EnforceAuthMiddleware,
        public_endpoints=settings.public_endpoints,
        private_endpoints=settings.private_endpoints,
        default_public=settings.default_public,
        oidc_discovery_url=settings.oidc_discovery_internal_url,
    )

    if settings.root_path or settings.upstream_url.path != "/":
        app.add_middleware(
            ProcessLinksMiddleware,
            upstream_url=str(settings.upstream_url),
            root_path=settings.root_path,
        )

    if settings.root_path:
        app.add_middleware(
            RemoveRootPathMiddleware,
            root_path=settings.root_path,
        )

    if settings.enable_compression:
        app.add_middleware(
            CompressionMiddleware,
        )

    return app