rm asset signing logic

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, S3AssetSigner
+from .handlers import HealthzHandler, ReverseProxyHandler
 from .middleware import (
     AddProcessTimeHeaderMiddleware,
     ApplyCql2FilterMiddleware,
@@ -60,14 +60,6 @@ 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,
@@ -79,8 +71,6 @@ 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,9 +44,6 @@ 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,15 +25,11 @@ 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 = (
@@ -88,60 +84,9 @@ 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