Support opt-in strict content negotiation

docs/source/configuration.rst

@@ -454,6 +454,17 @@ Some API rules state that trailing slashes at the end of a URL are not allowed i
 In that case, you may wish to set this property to ``true``. Doing so will result in a ``404 Not Found`` if a user adds a ``/`` to the end of a URL.
 If omitted or ``false`` (default), it does not matter whether the user omits or adds the ``/`` to the end of the URL.
 
+``strict_content_negotiation``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+If set to ``true``, requests with an ``Accept`` header that does not match any
+representation supported by the requested resource receive a ``406 Not
+Acceptable`` response. Wildcards such as ``*/*`` and ``image/*`` match the
+default or first compatible representation, respectively.
+
+If omitted or ``false`` (default), pygeoapi returns its default representation
+when none of the requested media types are supported.
+
 ``url_prefix``
 ^^^^^^^^^^^^^^
 

pygeoapi/api/__init__.py

@@ -185,11 +185,16 @@ def example_method(api: API, request: APIRequest, custom_arg):
 
     :param request:             The web platform specific Request instance.
     :param supported_locales:   List or set of supported Locale instances.
+    :param strict_content_negotiation: Return an invalid format for an
+                                       unsupported Accept header instead of
+                                       falling back to the default format.
     """
 
-    def __init__(self, request, supported_locales):
+    def __init__(self, request, supported_locales,
+                 strict_content_negotiation: bool = False):
         # Set default request data
         self._data = b''
+        self._strict_content_negotiation = strict_content_negotiation
 
         # Copy request query parameters
         self._args = self._get_params(request)
@@ -213,24 +218,28 @@ def __init__(self, request, supported_locales):
         self._headers = self.get_request_headers(request.headers)
 
     @classmethod
-    def from_flask(cls, request, supported_locales) -> 'APIRequest':
+    def from_flask(cls, request, supported_locales,
+                   strict_content_negotiation: bool = False) -> 'APIRequest':
         """Factory class similar to with_data, but only for flask requests"""
-        api_req = cls(request, supported_locales)
+        api_req = cls(request, supported_locales, strict_content_negotiation)
         api_req._data = request.data
         return api_req
 
     @classmethod
-    async def from_starlette(cls, request, supported_locales) -> 'APIRequest':
+    async def from_starlette(
+            cls, request, supported_locales,
+            strict_content_negotiation: bool = False) -> 'APIRequest':
         """Factory class similar to with_data, but only for starlette requests
         """
-        api_req = cls(request, supported_locales)
+        api_req = cls(request, supported_locales, strict_content_negotiation)
         api_req._data = await request.body()
         return api_req
 
     @classmethod
-    def from_django(cls, request, supported_locales) -> 'APIRequest':
+    def from_django(cls, request, supported_locales,
+                    strict_content_negotiation: bool = False) -> 'APIRequest':
         """Factory class similar to with_data, but only for django requests"""
-        api_req = cls(request, supported_locales)
+        api_req = cls(request, supported_locales, strict_content_negotiation)
         api_req._data = request.body
         return api_req
 
@@ -320,6 +329,10 @@ def _get_format(self, headers: dict,
         types_ = get_choice_from_headers(headers, 'accept', all=True)
         if types_ is None:
             return
+        if not types_:
+            if self._strict_content_negotiation:
+                return get_from_headers(headers, 'accept')
+            return
 
         merged_format_types = FORMAT_TYPES | extra_formats
 
@@ -331,6 +344,23 @@ def _get_format(self, headers: dict,
                 idx_ = mimes2.index(type_)
                 return fmts[idx_]
 
+            # A wildcard accepts the default representation.  A type
+            # wildcard accepts the first configured representation in that
+            # media type (for example, ``image/*`` accepts ``image/png``).
+            if type_ == '*/*':
+                return
+            if type_.endswith('/*'):
+                media_type = type_.split('/', 1)[0]
+                for idx_, mimetype in enumerate(mimes2):
+                    if mimetype.startswith(f'{media_type}/'):
+                        return fmts[idx_]
+
+        # In strict mode, keep an unmatched Accept value distinct from an
+        # absent Accept header. This lets adapters return HTTP 406 instead of
+        # silently serving the default format.
+        if self._strict_content_negotiation:
+            return types_[0]
+
     @property
     def data(self) -> bytes:
         """Returns the additional data send with the Request (bytes)"""
@@ -524,7 +554,8 @@ def __init__(self, config: dict, openapi: dict,
         self.config = config
         self.openapi = openapi
         self.asyncapi = asyncapi
-        self.api_headers = get_api_rules(self.config).response_headers
+        self.api_rules = get_api_rules(self.config)
+        self.api_headers = self.api_rules.response_headers
         self.base_url = get_base_url(self.config)
         self.prefetcher = UrlPrefetcher()
         self.pubsub_client = None
@@ -612,11 +643,18 @@ def get_format_exception(self,
         # Content-Language is in the system locale (ignore language settings)
         headers = request.get_response_headers(SYSTEM_LOCALE,
                                                **self.api_headers)
-        msg = 'Invalid format requested'
+        accept = get_from_headers(request.headers, 'accept')
+        if 'f' not in request.params and accept:
+            status = HTTPStatus.NOT_ACCEPTABLE
+            code = 'NotAcceptable'
+            msg = 'Requested media type is not supported'
+        else:
+            status = HTTPStatus.BAD_REQUEST
+            code = 'InvalidParameterValue'
+            msg = 'Invalid format requested'
         LOGGER.error(f'{msg}: {request.format}')
         return self.get_exception(
-            HTTPStatus.BAD_REQUEST, headers,
-            request.format, 'InvalidParameterValue', msg)
+            status, headers, request.format, code, msg)
 
     def get_collections_url(self) -> str:
         return f"{self.base_url}/collections"

pygeoapi/django_/views.py

@@ -581,7 +581,8 @@ def execute_from_django(api_function, request: HttpRequest, *args,
     else:
         api_ = API(settings.PYGEOAPI_CONFIG, settings.OPENAPI_DOCUMENT)
 
-    api_request = APIRequest.from_django(request, api_.locales)
+    api_request = APIRequest.from_django(
+        request, api_.locales, api_.api_rules.strict_content_negotiation)
     content: Union[str, bytes]
     if not skip_valid_check and not api_request.is_valid():
         headers, status, content = api_.get_format_exception(api_request)

pygeoapi/flask_app.py

@@ -145,7 +145,9 @@ def execute_from_flask(api_function: Callable, request: Request, *args,
 
     actual_api = api_ if alternative_api is None else alternative_api
 
-    api_request = APIRequest.from_flask(request, actual_api.locales)
+    api_request = APIRequest.from_flask(
+        request, actual_api.locales,
+        actual_api.api_rules.strict_content_negotiation)
 
     content: Union[str, bytes]
 

pygeoapi/models/config.py

@@ -67,6 +67,12 @@ class APIRules(BaseModel):
         description="If False (default), URL trailing slashes are allowed. "
                     "If True, pygeoapi will return a 404."
     )
+    strict_content_negotiation: bool = Field(
+        False,
+        description="If False (default), an unsupported Accept header falls "
+                    "back to the default representation. If True, pygeoapi "
+                    "will return a 406 Not Acceptable response."
+    )
 
     @staticmethod
     def create(**rules_config) -> 'APIRules':

pygeoapi/resources/schemas/config/pygeoapi-config-0.x.yml

@@ -153,6 +153,9 @@ properties:
                     strict_slashes:
                         type: boolean
                         description: whether trailing slashes are allowed in URLs (disallow = True)
+                    strict_content_negotiation:
+                        type: boolean
+                        description: whether unsupported Accept headers result in a 406 response
                     url_prefix:
                         type: string
                         description: |-
@@ -753,4 +756,4 @@ required:
     - server
     - logging
     - metadata
-    - resources
+    - resources

pygeoapi/starlette_app.py

@@ -127,7 +127,9 @@ async def execute_from_starlette(api_function, request: Request, *args,
                                  alternative_api: API | None = None
                                  ) -> Response:
     actual_api = api_ if alternative_api is None else alternative_api
-    api_request = await APIRequest.from_starlette(request, actual_api.locales)
+    api_request = await APIRequest.from_starlette(
+        request, actual_api.locales,
+        actual_api.api_rules.strict_content_negotiation)
     content: Union[str, bytes]
     if not skip_valid_check and not api_request.is_valid():
         headers, status, content = actual_api.get_format_exception(api_request)

tests/api/test_api.py

@@ -166,6 +166,35 @@ def test_apirequest(api_):
     assert apireq.get_linkrel(F_JSON) == 'self'
     assert apireq.get_linkrel(F_HTML) == 'alternate'
 
+    # Unsupported media types fall back to the default representation unless
+    # strict content negotiation is enabled
+    req = mock_request(HTTP_ACCEPT='application/vnd.lala')
+    apireq = APIRequest(req, api_.locales)
+    assert apireq.is_valid()
+    assert apireq.format is None
+
+    apireq = APIRequest(req, api_.locales, strict_content_negotiation=True)
+    assert not apireq.is_valid()
+    assert apireq.format == 'application/vnd.lala'
+    _, status, content = api_.get_format_exception(apireq)
+    assert status == HTTPStatus.NOT_ACCEPTABLE
+    assert 'Requested media type is not supported' in content
+
+    # Wildcards accept the default or first matching representation
+    req = mock_request(HTTP_ACCEPT='*/*')
+    apireq = APIRequest(req, api_.locales)
+    assert apireq.is_valid()
+    assert apireq.format is None
+
+    req = mock_request(HTTP_ACCEPT='image/*')
+    apireq = APIRequest(req, api_.locales)
+    assert apireq.is_valid()
+    assert apireq.format == 'png'
+
+    req = mock_request(HTTP_ACCEPT='not a valid media type;')
+    apireq = APIRequest(req, api_.locales, strict_content_negotiation=True)
+    assert not apireq.is_valid()
+
     # Test complex format string
     hh = 'text/html,application/xhtml+xml,application/xml;q=0.9,'
     req = mock_request(HTTP_ACCEPT=hh)
@@ -267,6 +296,7 @@ def test_apirules_active(config_with_rules, rules_api):
     assert rules_api.config == config_with_rules
     rules = get_api_rules(config_with_rules)
     base_url = get_base_url(config_with_rules)
+    assert rules.strict_content_negotiation
 
     # Test Flask
     flask_prefix = rules.get_url_prefix('flask')
@@ -286,6 +316,12 @@ def test_apirules_active(config_with_rules, rules_api):
         # Test strict slashes
         response = flask_client.get(f'{flask_prefix}/conformance/')
         assert response.status_code == 404
+
+        # Test strict content negotiation
+        response = flask_client.get(
+            f'{flask_prefix}/conformance',
+            headers={'Accept': 'application/vnd.lala'})
+        assert response.status_code == HTTPStatus.NOT_ACCEPTABLE
         # For the landing page ONLY, trailing slashes are actually preferred.
         # See https://docs.opengeospatial.org/is/17-069r4/17-069r4.html#_api_landing_page  # noqa
         # Omitting the trailing slash should lead to a redirect.
@@ -320,6 +356,12 @@ def test_apirules_active(config_with_rules, rules_api):
         response = starlette_client.get('/static/img/pygeoapi.png')
         assert response.status_code == 200
 
+        # Test strict content negotiation
+        response = starlette_client.get(
+            f'{starlette_prefix}/conformance',
+            headers={'Accept': 'application/vnd.lala'})
+        assert response.status_code == HTTPStatus.NOT_ACCEPTABLE
+
         # Test strict slashes
         response = starlette_client.get(f'{starlette_prefix}/conformance/')
         assert response.status_code == 404
@@ -344,6 +386,7 @@ def test_apirules_active(config_with_rules, rules_api):
 def test_apirules_inactive(config, api_):
     assert api_.config == config
     rules = get_api_rules(config)
+    assert not rules.strict_content_negotiation
 
     # Test Flask
     flask_prefix = rules.get_url_prefix('flask')

tests/pygeoapi-test-config-apirules.yml

@@ -56,6 +56,7 @@ server:
         output_dir: /tmp
     api_rules:
         strict_slashes: true
+        strict_content_negotiation: true
         url_prefix: 'v{api_major}'
         version_header: 'X-API-Version'