Revert "rm asset signing logic"

This reverts commit 9ce9cad.

Author: alukach

src/stac_auth_proxy/app.py

@@ -13,7 +13,7 @@
 from starlette_cramjam.middleware import CompressionMiddleware
 
 from .config import Settings
-from .handlers import HealthzHandler, ReverseProxyHandler
+from .handlers import HealthzHandler, ReverseProxyHandler, S3AssetSigner
 from .middleware import (
     AddProcessTimeHeaderMiddleware,
     ApplyCql2FilterMiddleware,
@@ -60,6 +60,14 @@ async def lifespan(app: FastAPI):
             prefix=settings.healthz_prefix,
         )
 
+    if settings.signer_endpoint:
+        # TODO: Warn/error if endpoint is public
+        app.add_api_route(
+            settings.signer_endpoint,
+            S3AssetSigner(bucket_pattern=settings.signer_asset_expression).endpoint,
+            methods=["POST"],
+        )
+
     app.add_api_route(
         "/{path:path}",
         ReverseProxyHandler(upstream=str(settings.upstream_url)).proxy_request,
@@ -71,6 +79,8 @@ async def lifespan(app: FastAPI):
     #
     app.add_middleware(
         AuthenticationExtensionMiddleware,
+        signing_endpoint=settings.signer_endpoint,
+        signed_asset_expression=settings.signer_asset_expression,
         default_public=settings.default_public,
         public_endpoints=settings.public_endpoints,
         private_endpoints=settings.private_endpoints,

src/stac_auth_proxy/config.py

@@ -44,6 +44,9 @@ class Settings(BaseSettings):
     healthz_prefix: str = Field(pattern=_PREFIX_PATTERN, default="/healthz")
     openapi_spec_endpoint: Optional[str] = Field(pattern=_PREFIX_PATTERN, default=None)
 
+    signer_endpoint: Optional[str] = Field(pattern=_PREFIX_PATTERN, default=None)
+    signer_asset_expression: str = Field(default=r"^s3://.*$")
+
     # Auth
     default_public: bool = False
     public_endpoints: EndpointMethodsNoScope = {

src/stac_auth_proxy/middleware/AuthenticationExtensionMiddleware.py

@@ -25,11 +25,15 @@ class AuthenticationExtensionMiddleware(JsonResponseMiddleware):
 
     app: ASGIApp
 
+    signing_endpoint: Optional[str]
+    signed_asset_expression: str
+
     default_public: bool
     private_endpoints: EndpointMethods
     public_endpoints: EndpointMethods
 
     oidc_config_url: Optional[HttpUrl] = None
+    signing_scheme_name: str = "signed_url_auth"
     auth_scheme_name: str = "oauth"
     auth_scheme: dict[str, Any] = field(default_factory=dict)
     extension_url: str = (
@@ -84,9 +88,60 @@ def transform_json(self, doc: dict[str, Any]) -> dict[str, Any]:
         scheme_loc = doc["properties"] if "properties" in doc else doc
         schemes = scheme_loc.setdefault("auth:schemes", {})
         schemes[self.auth_scheme_name] = self.auth_scheme
+        if self.signing_endpoint:
+            schemes[self.signing_scheme_name] = {
+                "type": "signedUrl",
+                "description": "Requires an authentication API",
+                "flows": {
+                    "authorizationCode": {
+                        "authorizationApi": self.signing_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",
+                    }
+                },
+            }
 
         # auth:refs
         # ---
+        # Annotate assets with "auth:refs": [signing_scheme]
+        if self.signing_endpoint:
+            assets = chain(
+                # Item
+                doc.get("assets", {}).values(),
+                # Items/Search
+                (
+                    asset
+                    for item in doc.get("features", [])
+                    for asset in item.get("assets", {}).values()
+                ),
+            )
+            for asset in assets:
+                if "href" not in asset:
+                    logger.warning("Asset %s has no href", asset)
+                    continue
+                if re.match(self.signed_asset_expression, asset["href"]):
+                    asset.setdefault("auth:refs", []).append(self.signing_scheme_name)
+
         # Annotate links with "auth:refs": [auth_scheme]
         links = chain(
             # Item/Collection