NRL-512 Initial implementation for rejecting HEAD requests

Author: axelkrastek1-nhs

terraform/infrastructure/modules/api_gateway/api_gateway.tf

@@ -60,7 +60,8 @@ resource "aws_api_gateway_deployment" "api_gateway_deployment" {
   }
 
   depends_on = [
-    aws_api_gateway_rest_api.api_gateway_rest_api
+    aws_api_gateway_rest_api.api_gateway_rest_api,
+    aws_api_gateway_integration_response.head_integration_response
   ]
 }
 

terraform/infrastructure/modules/api_gateway/method_responses.tf

@@ -0,0 +1,76 @@
+# Define a reusable map of allowed methods for each path
+locals {
+  endpoint_allowed_methods = {
+    "/DocumentReference" = "GET,POST,PUT"
+    # "/DocumentReference/_search" = "POST"
+    # "/DocumentReference/{id}" = "GET"
+  }
+}
+
+data "aws_api_gateway_resource" "resource" {
+  for_each    = local.endpoint_allowed_methods
+  rest_api_id = aws_api_gateway_rest_api.api_gateway_rest_api.id
+  path        = each.key
+
+  depends_on = [aws_api_gateway_rest_api.api_gateway_rest_api]
+}
+
+# Add HEAD method to each resource with 405 response
+resource "aws_api_gateway_method" "head_method" {
+  for_each = local.endpoint_allowed_methods
+
+  rest_api_id   = aws_api_gateway_rest_api.api_gateway_rest_api.id
+  resource_id   = data.aws_api_gateway_resource.resource[each.key].id
+  http_method   = "HEAD"
+  authorization = "NONE"
+}
+
+resource "aws_api_gateway_integration" "head_integration" {
+  for_each = local.endpoint_allowed_methods
+
+  rest_api_id = aws_api_gateway_rest_api.api_gateway_rest_api.id
+  resource_id = data.aws_api_gateway_resource.resource[each.key].id
+  http_method = aws_api_gateway_method.head_method[each.key].http_method
+  type        = "MOCK"
+  # passthrough_behavior = "WHEN_NO_TEMPLATES"
+  passthrough_behavior = "WHEN_NO_MATCH"
+
+  request_templates = {
+    "application/json"      = <<-EOF
+      { "statusCode": 405 }
+    EOF
+    "application/json+fhir" = <<-EOF
+      { "statusCode": 405 }
+    EOF
+    "application/fhir+json" = <<-EOF
+      { "statusCode": 405 }
+    EOF
+  }
+}
+
+resource "aws_api_gateway_method_response" "head_method_response" {
+  for_each = local.endpoint_allowed_methods
+
+  rest_api_id = aws_api_gateway_rest_api.api_gateway_rest_api.id
+  resource_id = data.aws_api_gateway_resource.resource[each.key].id
+  http_method = aws_api_gateway_method.head_method[each.key].http_method
+  status_code = "405"
+
+  response_parameters = {
+    "method.response.header.Allow" = true
+  }
+}
+
+resource "aws_api_gateway_integration_response" "head_integration_response" {
+  for_each = local.endpoint_allowed_methods
+
+  rest_api_id       = aws_api_gateway_rest_api.api_gateway_rest_api.id
+  resource_id       = data.aws_api_gateway_resource.resource[each.key].id
+  http_method       = aws_api_gateway_method.head_method[each.key].http_method
+  status_code       = aws_api_gateway_method_response.head_method_response[each.key].status_code
+  selection_pattern = "" # default catch-all
+
+  response_parameters = {
+    "method.response.header.Allow" = "'${each.value}'"
+  }
+}

tests/features/consumer/headOnAnyEndpoint.feature

@@ -0,0 +1,9 @@
+Feature: Consumer - HEAD Requests
+
+  @custom_tag
+  Scenario: DocumentReference with HEAD fails
+    Given the application 'DataShare' (ID 'z00z-y11y-x22x') is registered to access the API
+    When consumer 'RX898' sends HEAD request to DocumentReference endpoint
+    Then the response status code is 405
+    And the response has an empty body
+    And the Allow header is 'GET'

tests/features/steps/2_request.py

@@ -76,6 +76,27 @@ def consumer_read_document_reference_step(
     context.response = client.read(doc_ref_id)
 
 
+@when("consumer '{ods_code}' sends HEAD request to {endpoint} endpoint")
+def consumer_head_request_step(context: Context, ods_code: str, endpoint: str):
+    client = consumer_client_from_context(context, ods_code)
+    context.response = client.head(endpoint)
+
+
+@when(
+    "consumer '{ods_code}' sends HEAD request to '{endpoint}' endpoint with parameters"
+)
+def consumer_head_request_step_with_parameters(
+    context: Context, ods_code: str, endpoint: str
+):
+    if not context.table:
+        raise ValueError("No search query table provided")
+
+    items = {row["parameter"]: row["value"] for row in context.table}
+
+    client = consumer_client_from_context(context, ods_code)
+    context.response = client.head(endpoint, items)
+
+
 @when("producer '{ods_code}' creates a DocumentReference with values")
 def create_post_document_reference_step(context: Context, ods_code: str):
     client = producer_client_from_context(context, ods_code)

tests/features/steps/3_assert.py

@@ -51,6 +51,20 @@ def assert_bundle_step(context: Context, bundle_type: str):
     context.bundle = Bundle.model_validate(body)
 
 
+@then("the response has an empty body")
+def assert_empty_body_step(context: Context):
+    """
+    Asserts that the response body is empty.
+    """
+    assert context.response.text == "", format_error(
+        "Response body is not empty",
+        "empty",
+        context.response.text,
+        context.response.text,
+    )
+    context.bundle = None
+
+
 @then("the Bundle has a total of {total}")
 def assert_bundle_total_step(context: Context, total: str):
     assert (

tests/utilities/api_clients.py

@@ -192,6 +192,22 @@ def search_post(
             cert=self.config.client_cert,
         )
 
+    @retry_if([502])
+    def head(
+        self,
+        endpoint: str,
+        extra_params: dict[str, str] | None = None,
+    ) -> Response:
+        params = {**(extra_params or {})}
+        url = f"{self.api_url}/{endpoint}"
+        headers = {**self.request_headers, "Content-Type": "application/json"}
+        return requests.head(
+            url,
+            params=params,
+            headers=headers,
+            cert=self.config.client_cert,
+        )
+
     @retry_if([502])
     def read_capability_statement(self) -> Response:
         return requests.get(