Skip to content

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
allowed_jwt_audiences Sequence[str] | None
None
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
'/api'
openapi_auth_scheme_name str
'oidcAuth'
openapi_auth_scheme_override dict | None
None
swagger_ui_endpoint str | None
'/api.html'
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'], '^/conformance$': ['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 
 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
class Settings(BaseSettings):
    """Configuration settings for the STAC Auth Proxy."""

    # External URLs
    upstream_url: HttpUrl
    oidc_discovery_url: HttpUrl
    oidc_discovery_internal_url: HttpUrl
    allowed_jwt_audiences: Optional[Sequence[str]] = None

    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="/api"
    )
    openapi_auth_scheme_name: str = "oidcAuth"
    openapi_auth_scheme_override: Optional[dict] = None
    swagger_ui_endpoint: Optional[str] = Field(
        pattern=_PREFIX_PATTERN, default="/api.html"
    )
    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"^/conformance$": ["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

    @field_validator("allowed_jwt_audiences", mode="before")
    @classmethod
    def parse_audience(cls, v) -> Optional[Sequence[str]]:
        """Parse a comma separated string list of audiences into a list."""
        return str2list(v)

parse_audience(v) -> Optional[Sequence[str]] classmethod

Parse a comma separated string list of audiences into a list.

Source code in src/stac_auth_proxy/config.py 
110
111
112
113
114
@field_validator("allowed_jwt_audiences", mode="before")
@classmethod
def parse_audience(cls, v) -> Optional[Sequence[str]]:
    """Parse a comma separated string list of audiences into a list."""
    return str2list(v)

build_lifespan(settings: Settings | None = None, **settings_kwargs: Any)

Create a lifespan handler that runs startup checks.

Parameters

settings : Settings | None, optional Pre-built settings instance. If omitted, a new one is constructed from settings_kwargs. **settings_kwargs : Any Keyword arguments used to configure the health and conformance checks if settings is not provided.

Returns

Callable[[FastAPI], AsyncContextManager[Any]] A callable suitable for the lifespan parameter of FastAPI.

Source code in src/stac_auth_proxy/lifespan.py 
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
def build_lifespan(settings: Settings | None = None, **settings_kwargs: Any):
    """
    Create a lifespan handler that runs startup checks.

    Parameters
    ----------
    settings : Settings | None, optional
        Pre-built settings instance. If omitted, a new one is constructed from
        ``settings_kwargs``.
    **settings_kwargs : Any
        Keyword arguments used to configure the health and conformance checks if
        ``settings`` is not provided.

    Returns
    -------
    Callable[[FastAPI], AsyncContextManager[Any]]
        A callable suitable for the ``lifespan`` parameter of ``FastAPI``.
    """
    if settings is None:
        settings = Settings(**settings_kwargs)

    @asynccontextmanager
    async def lifespan(app: "FastAPI"):
        assert settings is not None  # Required for type checking

        # Wait for upstream servers to become available
        if settings.wait_for_upstream:
            await check_server_healths(
                settings.upstream_url, settings.oidc_discovery_internal_url
            )

        # 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

    return lifespan

configure_app(app: FastAPI, settings: Optional[Settings] = None, **settings_kwargs: Any) -> FastAPI

Apply routes and middleware to a FastAPI app.

Parameters

app : FastAPI The FastAPI app to configure. settings : Settings | None, optional Pre-built settings instance. If omitted, a new one is constructed from settings_kwargs. **settings_kwargs : Any Keyword arguments used to configure the health and conformance checks if settings is not provided.

Source code in src/stac_auth_proxy/app.py 
 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
149
150
151
152
153
154
def configure_app(
    app: FastAPI,
    settings: Optional[Settings] = None,
    **settings_kwargs: Any,
) -> FastAPI:
    """
    Apply routes and middleware to a FastAPI app.

    Parameters
    ----------
    app : FastAPI
        The FastAPI app to configure.
    settings : Settings | None, optional
        Pre-built settings instance. If omitted, a new one is constructed from
        ``settings_kwargs``.
    **settings_kwargs : Any
        Keyword arguments used to configure the health and conformance checks if
        ``settings`` is not provided.
    """
    settings = settings or Settings(**settings_kwargs)

    #
    # Route Handlers
    #

    # If we have customized Swagger UI Init settings (e.g. a provided client_id)
    # then we need to serve our own Swagger UI in place of the upstream's. This requires
    # that we know the Swagger UI endpoint and the OpenAPI spec endpoint.
    if all(
        [
            settings.swagger_ui_endpoint,
            settings.openapi_spec_endpoint,
            settings.swagger_ui_init_oauth,
        ]
    ):
        app.add_route(
            settings.swagger_ui_endpoint,
            SwaggerUI(
                openapi_url=settings.openapi_spec_endpoint,  # type: ignore
                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,
        )

    #
    # 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,
        allowed_jwt_audiences=settings.allowed_jwt_audiences,
    )

    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

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

FastAPI Application Factory.

Source code in src/stac_auth_proxy/app.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def create_app(settings: Optional[Settings] = None) -> FastAPI:
    """FastAPI Application Factory."""
    settings = settings or Settings()

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

    configure_app(app, settings)

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

    return app