buildout working example

Author: alukach

src/stac_auth_proxy/app.py

@@ -16,6 +16,7 @@
 from .middleware import (
     AddProcessTimeHeaderMiddleware,
     ApplyCql2FilterMiddleware,
+    AuthenticationExtensionMiddleware,
     BuildCql2FilterMiddleware,
     EnforceAuthMiddleware,
     OpenApiMiddleware,
@@ -61,6 +62,11 @@ def create_app(settings: Optional[Settings] = None) -> FastAPI:
             S3AssetSigner(bucket_pattern=settings.signer_asset_expression).endpoint,
             methods=["POST"],
         )
+        app.add_middleware(
+            AuthenticationExtensionMiddleware,
+            endpoint=settings.signer_endpoint,
+            asset_expression=settings.signer_asset_expression,
+        )
 
     app.add_api_route(
         "/{path:path}",
@@ -81,11 +87,6 @@ def create_app(settings: Optional[Settings] = None) -> FastAPI:
             default_public=settings.default_public,
         )
 
-        # signers={
-        #     schema: endpoint
-        #     for schema, endpoint in {"s3": settings.signer_endpoint}.items()
-        #     if endpoint
-        # },
     if settings.items_filter:
         app.add_middleware(
             ApplyCql2FilterMiddleware,

src/stac_auth_proxy/middleware/AuthenticationExtensionMiddleware.py

@@ -1,11 +1,13 @@
-"""Middleware to add auth information to the OpenAPI spec served by upstream API."""
+"""Middleware to add auth information to item response served by upstream API."""
 
-from dataclasses import dataclass, field
+import re
+from dataclasses import dataclass
 from typing import Any
 
 from starlette.requests import Request
 from starlette.types import ASGIApp
 
+from ..utils.filters import is_item_endpoint
 from ..utils.middleware import JsonResponseMiddleware
 
 
@@ -14,14 +16,58 @@ class AuthenticationExtensionMiddleware(JsonResponseMiddleware):
     """Middleware to add the authentication extension to the response."""
 
     app: ASGIApp
-    signers: dict[str, str] = field(default_factory=dict)
+    endpoint: str
+    asset_expression: str
 
     def should_transform_response(self, request: Request) -> bool:
         """Only transform responses for STAC Items."""
-        # TODO: Implement proper path matching for STAC Items
-        return True
+        return is_item_endpoint(request.url.path)
 
     def transform_json(self, item: dict[str, Any]) -> dict[str, Any]:
         """Augment the STAC Item with auth information."""
-        # TODO: Implement STAC Item augmentation
+        extension = (
+            "https://stac-extensions.github.io/authentication/v1.1.0/schema.json"
+        )
+        extensions = item.setdefault("stac_extensions", [])
+        if extension not in extensions:
+            extensions.append(extension)
+
+        # TODO: Should we add this to items even if the assets don't match the asset expression?
+        schemes = item["properties"].setdefault("auth:schemes", {})
+        scheme = "signed_url_auth"
+        schemes[scheme] = {
+            "type": "signedUrl",
+            "description": "Requires an authentication API",
+            "flows": {
+                "authorizationCode": {
+                    "authorizationApi": self.endpoint,
+                    "method": "POST",
+                    "parameters": {
+                        "bucket": {
+                            "in": "body",
+                            "required": True,
+                            "description": "asset bucket",
+                            "schema": {
+                                "type": "string",
+                                "examples": "example-bucket",
+                            },
+                        },
+                        "key": {
+                            "in": "body",
+                            "required": True,
+                            "description": "asset key",
+                            "schema": {
+                                "type": "string",
+                                "examples": "path/to/example/asset.xyz",
+                            },
+                        },
+                    },
+                    "responseField": "signed_url",
+                }
+            },
+        }
+
+        for asset in item["assets"].values():
+            if re.match(self.asset_expression, asset.get("href", "")):
+                asset.setdefault("auth:refs", []).append(scheme)
         return item