Skip to content

rs_server_catalog/middleware/catalog_middleware.md

<< Back to index

A BaseHTTPMiddleware to handle the user multi catalog.

The stac-fastapi software doesn't handle multi catalog. In the rs-server we need to handle user-based catalogs.

The rs-server uses only one catalog but the collections are prefixed by the user name. The middleware is used to hide this mechanism.

The middleware: * redirect the user-specific request to the common stac api endpoint * modifies the request to add the user prefix in the collection name * modifies the response to remove the user prefix in the collection name * modifies the response to update the links.

CatalogMiddleware

Bases: BaseHTTPMiddleware

Middleware entry point for the user-aware catalog facade.

A fresh UserCatalog handler is created for each request so mutable request_ids state is never shared across concurrent requests.

Source code in docs/rs-server/services/catalog/rs_server_catalog/middleware/catalog_middleware.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
class CatalogMiddleware(BaseHTTPMiddleware):  # pylint: disable=too-few-public-methods
    """
    Middleware entry point for the user-aware catalog facade.

    A fresh UserCatalog handler is created for each request so mutable
    `request_ids` state is never shared across concurrent requests.
    """

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        """Redirect the user catalog specific endpoint and adapt the response content."""
        # NOTE: maybe we could move the UserCatalog.dispatch here but I'm not sure it's thread-safe.
        # Maybe this is the reason why we init a new UserCatalog instance everytime. To be confirmed.
        return await UserCatalog(api.client).dispatch(request, call_next)

dispatch(request, call_next) async

Redirect the user catalog specific endpoint and adapt the response content.

Source code in docs/rs-server/services/catalog/rs_server_catalog/middleware/catalog_middleware.py 
56
57
58
59
60
async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
    """Redirect the user catalog specific endpoint and adapt the response content."""
    # NOTE: maybe we could move the UserCatalog.dispatch here but I'm not sure it's thread-safe.
    # Maybe this is the reason why we init a new UserCatalog instance everytime. To be confirmed.
    return await UserCatalog(api.client).dispatch(request, call_next)

UserCatalog

Per-request catalog middleware handler.

It collects caller/owner/collection/item identifiers, rewrites the incoming request to the internal pgstac route shape, delegates to stac-fastapi, then rewrites the response back to the public RS Server catalog API.

Source code in docs/rs-server/services/catalog/rs_server_catalog/middleware/catalog_middleware.py 
 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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
class UserCatalog:  # pylint: disable=too-few-public-methods
    """
    Per-request catalog middleware handler.

    It collects caller/owner/collection/item identifiers, rewrites the incoming
    request to the internal pgstac route shape, delegates to stac-fastapi, then
    rewrites the response back to the public RS Server catalog API.
    """

    def __init__(self, client: CoreCrudClient):
        """Constructor, called from the middleware"""
        self.request_ids: dict[Any, Any] = {}
        self.client = client

    async def dispatch(
        self,
        request: Request,
        call_next: RequestResponseEndpoint,
    ) -> Response:
        """
        Redirect the user catalog specific endpoint and adapt the response content.

        The dispatch flow has three phases:
        1. recover authentication and request identifiers,
        2. let CatalogRequestManager rewrite/authorize the request,
        3. let CatalogResponseManager adapt the backend response and side effects.

        Args:
            request (Request): Initial request
            call_next: next call to apply

        Returns:
            response (Response): Response to the current request
        """
        # Read mutable request bodies once at middleware level. The request
        # manager can later replace Starlette's cached body/json if it rewrites
        # owner, collection, timestamps or filters.
        request_body = None if request.method not in ["PATCH", "POST", "PUT"] else await request.json()
        auth_roles = user_login = owner_id = None
        logger.info("Catalog request received: %s %s", request.method, request.url.path)
        logger.debug("Catalog request query params: %s", dict(request.query_params))
        if request_body is not None:
            logger.debug("Catalog request body for %s %s: %s", request.method, request.url.path, request_body)

        # ---------- Management of  authentification (retrieve user_login + default owner_id)
        if common_settings.CLUSTER_MODE:  # Get the list of access and the user_login calling the endpoint.
            try:
                auth_roles = request.state.auth_roles
                user_login = request.state.user_login
            # Case of endpoints that do not call the authenticate function
            # Get the the user_login calling the endpoint. If this is not set (the authentication.authenticate function
            # is not called), the local user shall be used (later on, in rereoute_url)
            # The common_settings.CLUSTER_MODE may not be used because for some endpoints like /api
            # the authenticate is not called even if common_settings.CLUSTER_MODE is True. Thus, the presence of
            # user_login has to be checked instead
            except (NameError, AttributeError):
                auth_roles = []
                user_login = get_user(None, None)  # Get default local or cluster user
                logger.debug("Catalog request has no authenticated state; using fallback user %s", user_login)
        elif common_settings.LOCAL_MODE:
            user_login = get_user(None, None)
        owner_id = ""  # Default owner_id is empty
        logger.debug(
            f"Received {request.method} from '{user_login}' | {request.url.path}?{request.query_params}",
        )

        # ---------- Request rerouting
        # Dictionary to carry the identifiers recovered from auth/path/body
        # through request and response processing.
        self.request_ids = {
            "auth_roles": auth_roles,
            "user_login": user_login,
            "owner_id": owner_id,
            "collection_ids": [],
            "item_id": "",
        }
        reroute_url(request, self.request_ids)
        if not request.scope["path"]:  # Invalid endpoint
            logger.error("Invalid catalog endpoint for request %s %s", request.method, request.url.path)
            raise HTTPException(status_code=HTTP_400_BAD_REQUEST, detail="Invalid endpoint.")
        logger.debug(f"path = {request.scope['path']} | requests_ids = {self.request_ids}")
        logger.info(
            "Catalog request routed: %s %s -> %s; owner=%s, collections=%s, item=%s",
            request.method,
            request.url.path,
            request.scope["path"],
            self.request_ids["owner_id"],
            self.request_ids["collection_ids"],
            self.request_ids["item_id"],
        )

        # Ensure that user_login is not null after rerouting
        if not self.request_ids["user_login"]:
            logger.error("Catalog request has no user_login after rerouting: %s", self.request_ids)
            raise HTTPException(
                status_code=HTTP_500_INTERNAL_SERVER_ERROR,
                detail="user_login is not defined !",
            )

        # ---------- Body data recovery
        # Some endpoints carry owner/collection/item identifiers only in the body
        # (for example POST collection or POST item). Fill missing values before
        # authorization and route-specific rewriting.
        if request_body:
            # Edit owner_id with the corresponding body content if exist
            if not self.request_ids["owner_id"]:
                self.request_ids["owner_id"] = request_body.get("owner")
            # received a POST/PUT/PATCH for a STAC item or
            # a STAC collection is created
            if len(self.request_ids["collection_ids"]) == 0:
                collections = request_body.get("collections") or request_body.get("id")
                if collections:
                    self.request_ids["collection_ids"] = collections if isinstance(collections, list) else [collections]

            if not self.request_ids["item_id"] and request_body.get("type") == "Feature":
                self.request_ids["item_id"] = request_body.get("id")
            logger.debug("Catalog request ids after body recovery: %s", self.request_ids)

        # ---------- Apply specific changes for each endpoint
        request_manager = CatalogRequestManager(self.client, self.request_ids)
        request = await request_manager.manage_requests(request)
        # If the request manager returns a response, it usually means the user is not authorized
        # to do the operation received, so we directly return the response
        if isinstance(request, Response):
            logger.info(
                "Catalog request %s %s returned early with status %s",
                request_manager.request_ids,
                getattr(request, "media_type", None),
                request.status_code,
            )
            return request

        response = await call_next(request)
        logger.info(
            "Catalog backend response for %s %s has status %s",
            request.method,
            request.scope["path"],
            response.status_code,
        )

        response_manager = CatalogResponseManager(
            request_manager.client,
            request_manager.request_ids,
            request_manager.s3_files_to_be_deleted,
        )
        managed_response = await response_manager.manage_responses(request, cast(StreamingResponse, response))
        logger.info(
            "Catalog response completed for %s %s with status %s",
            request.method,
            request.scope["path"],
            managed_response.status_code,
        )
        return managed_response

__init__(client)

Constructor, called from the middleware

Source code in docs/rs-server/services/catalog/rs_server_catalog/middleware/catalog_middleware.py 
72
73
74
75
def __init__(self, client: CoreCrudClient):
    """Constructor, called from the middleware"""
    self.request_ids: dict[Any, Any] = {}
    self.client = client

dispatch(request, call_next) async

Redirect the user catalog specific endpoint and adapt the response content.

The dispatch flow has three phases: 1. recover authentication and request identifiers, 2. let CatalogRequestManager rewrite/authorize the request, 3. let CatalogResponseManager adapt the backend response and side effects.

Parameters:

Name Type Description Default
request Request

Initial request

 required
call_next RequestResponseEndpoint

next call to apply

 required

Returns:

Name Type Description
response Response

Response to the current request
Source code in docs/rs-server/services/catalog/rs_server_catalog/middleware/catalog_middleware.py 
 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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
async def dispatch(
    self,
    request: Request,
    call_next: RequestResponseEndpoint,
) -> Response:
    """
    Redirect the user catalog specific endpoint and adapt the response content.

    The dispatch flow has three phases:
    1. recover authentication and request identifiers,
    2. let CatalogRequestManager rewrite/authorize the request,
    3. let CatalogResponseManager adapt the backend response and side effects.

    Args:
        request (Request): Initial request
        call_next: next call to apply

    Returns:
        response (Response): Response to the current request
    """
    # Read mutable request bodies once at middleware level. The request
    # manager can later replace Starlette's cached body/json if it rewrites
    # owner, collection, timestamps or filters.
    request_body = None if request.method not in ["PATCH", "POST", "PUT"] else await request.json()
    auth_roles = user_login = owner_id = None
    logger.info("Catalog request received: %s %s", request.method, request.url.path)
    logger.debug("Catalog request query params: %s", dict(request.query_params))
    if request_body is not None:
        logger.debug("Catalog request body for %s %s: %s", request.method, request.url.path, request_body)

    # ---------- Management of  authentification (retrieve user_login + default owner_id)
    if common_settings.CLUSTER_MODE:  # Get the list of access and the user_login calling the endpoint.
        try:
            auth_roles = request.state.auth_roles
            user_login = request.state.user_login
        # Case of endpoints that do not call the authenticate function
        # Get the the user_login calling the endpoint. If this is not set (the authentication.authenticate function
        # is not called), the local user shall be used (later on, in rereoute_url)
        # The common_settings.CLUSTER_MODE may not be used because for some endpoints like /api
        # the authenticate is not called even if common_settings.CLUSTER_MODE is True. Thus, the presence of
        # user_login has to be checked instead
        except (NameError, AttributeError):
            auth_roles = []
            user_login = get_user(None, None)  # Get default local or cluster user
            logger.debug("Catalog request has no authenticated state; using fallback user %s", user_login)
    elif common_settings.LOCAL_MODE:
        user_login = get_user(None, None)
    owner_id = ""  # Default owner_id is empty
    logger.debug(
        f"Received {request.method} from '{user_login}' | {request.url.path}?{request.query_params}",
    )

    # ---------- Request rerouting
    # Dictionary to carry the identifiers recovered from auth/path/body
    # through request and response processing.
    self.request_ids = {
        "auth_roles": auth_roles,
        "user_login": user_login,
        "owner_id": owner_id,
        "collection_ids": [],
        "item_id": "",
    }
    reroute_url(request, self.request_ids)
    if not request.scope["path"]:  # Invalid endpoint
        logger.error("Invalid catalog endpoint for request %s %s", request.method, request.url.path)
        raise HTTPException(status_code=HTTP_400_BAD_REQUEST, detail="Invalid endpoint.")
    logger.debug(f"path = {request.scope['path']} | requests_ids = {self.request_ids}")
    logger.info(
        "Catalog request routed: %s %s -> %s; owner=%s, collections=%s, item=%s",
        request.method,
        request.url.path,
        request.scope["path"],
        self.request_ids["owner_id"],
        self.request_ids["collection_ids"],
        self.request_ids["item_id"],
    )

    # Ensure that user_login is not null after rerouting
    if not self.request_ids["user_login"]:
        logger.error("Catalog request has no user_login after rerouting: %s", self.request_ids)
        raise HTTPException(
            status_code=HTTP_500_INTERNAL_SERVER_ERROR,
            detail="user_login is not defined !",
        )

    # ---------- Body data recovery
    # Some endpoints carry owner/collection/item identifiers only in the body
    # (for example POST collection or POST item). Fill missing values before
    # authorization and route-specific rewriting.
    if request_body:
        # Edit owner_id with the corresponding body content if exist
        if not self.request_ids["owner_id"]:
            self.request_ids["owner_id"] = request_body.get("owner")
        # received a POST/PUT/PATCH for a STAC item or
        # a STAC collection is created
        if len(self.request_ids["collection_ids"]) == 0:
            collections = request_body.get("collections") or request_body.get("id")
            if collections:
                self.request_ids["collection_ids"] = collections if isinstance(collections, list) else [collections]

        if not self.request_ids["item_id"] and request_body.get("type") == "Feature":
            self.request_ids["item_id"] = request_body.get("id")
        logger.debug("Catalog request ids after body recovery: %s", self.request_ids)

    # ---------- Apply specific changes for each endpoint
    request_manager = CatalogRequestManager(self.client, self.request_ids)
    request = await request_manager.manage_requests(request)
    # If the request manager returns a response, it usually means the user is not authorized
    # to do the operation received, so we directly return the response
    if isinstance(request, Response):
        logger.info(
            "Catalog request %s %s returned early with status %s",
            request_manager.request_ids,
            getattr(request, "media_type", None),
            request.status_code,
        )
        return request

    response = await call_next(request)
    logger.info(
        "Catalog backend response for %s %s has status %s",
        request.method,
        request.scope["path"],
        response.status_code,
    )

    response_manager = CatalogResponseManager(
        request_manager.client,
        request_manager.request_ids,
        request_manager.s3_files_to_be_deleted,
    )
    managed_response = await response_manager.manage_responses(request, cast(StreamingResponse, response))
    logger.info(
        "Catalog response completed for %s %s with status %s",
        request.method,
        request.scope["path"],
        managed_response.status_code,
    )
    return managed_response