nhs number validation fix

src/eligibility_signposting_api/common/api_error_response.py

@@ -128,7 +128,7 @@ def log_and_generate_response(
     fhir_display_message="An unexpected internal server error occurred.",
 )
 
-NHS_NUMBER_MISMATCH_ERROR = APIErrorResponse(
+NHS_NUMBER_ERROR = APIErrorResponse(
     status_code=HTTPStatus.FORBIDDEN,
     fhir_issue_code=FHIRIssueCode.FORBIDDEN,
     fhir_issue_severity=FHIRIssueSeverity.ERROR,

src/eligibility_signposting_api/common/request_validator.py

@@ -10,7 +10,7 @@
     INVALID_CATEGORY_ERROR,
     INVALID_CONDITION_FORMAT_ERROR,
     INVALID_INCLUDE_ACTIONS_ERROR,
-    NHS_NUMBER_MISMATCH_ERROR,
+    NHS_NUMBER_ERROR,
 )
 from eligibility_signposting_api.config.constants import NHS_NUMBER_HEADER
 
@@ -39,17 +39,7 @@ def validate_query_params(query_params: dict[str, str]) -> tuple[bool, ResponseR
     return True, None
 
 
-def validate_nhs_number(path_nhs: str | None, header_nhs: str | None) -> bool:
-    logger.info("NHS numbers from the request", extra={"header_nhs": header_nhs, "path_nhs": path_nhs})
-
-    if not path_nhs:
-        logger.error("NHS number is not present in path", extra={"path_nhs": path_nhs})
-        return False
-
-    if not header_nhs:  # Not a validation error
-        logger.info("NHS number is not present in header", extra={"header_nhs": header_nhs, "path_nhs": path_nhs})
-        return True
-
+def validate_nhs_number_in_header(path_nhs: str | None, header_nhs: str | None) -> bool:
     if header_nhs != path_nhs:
         logger.error("NHS number mismatch", extra={"header_nhs": header_nhs, "path_nhs": path_nhs})
         return False
@@ -60,12 +50,22 @@ def validate_request_params() -> Callable:
     def decorator(func: Callable) -> Callable:
         @wraps(func)
         def wrapper(*args, **kwargs) -> ResponseReturnValue:  # noqa:ANN002,ANN003
-            path_nhs_number = str(kwargs.get("nhs_number"))
-            header_nhs_no = str(request.headers.get(NHS_NUMBER_HEADER))
+            path_nhs_number = str(kwargs.get("nhs_number")) if kwargs.get("nhs_number") else None
 
-            if not validate_nhs_number(path_nhs_number, header_nhs_no):
+            if not path_nhs_number:
                 message = "You are not authorised to request information for the supplied NHS Number"
-                return NHS_NUMBER_MISMATCH_ERROR.log_and_generate_response(log_message=message, diagnostics=message)
+                return NHS_NUMBER_ERROR.log_and_generate_response(log_message=message, diagnostics=message)
+
+            if NHS_NUMBER_HEADER in request.headers:
+                header_nhs_no = request.headers.get(NHS_NUMBER_HEADER)
+                logger.info(
+                    "NHS numbers from the request", extra={"header_nhs": header_nhs_no, "path_nhs": path_nhs_number}
+                )
+                if not validate_nhs_number_in_header(path_nhs_number, header_nhs_no):
+                    message = "You are not authorised to request information for the supplied NHS Number"
+                    return NHS_NUMBER_ERROR.log_and_generate_response(log_message=message, diagnostics=message)
+            else:
+                logger.info("NHS numbers from the request", extra={"header_nhs": None, "path_nhs": path_nhs_number})
 
             query_params = request.args
             if query_params:

tests/integration/in_process/test_eligibility_endpoint.py

@@ -1,5 +1,6 @@
 from http import HTTPStatus
 
+import pytest
 from botocore.client import BaseClient
 from brunns.matchers.data import json_matching as is_json_that
 from brunns.matchers.werkzeug import is_werkzeug_response as is_response
@@ -39,6 +40,77 @@ def test_nhs_number_given(
             is_response().with_status_code(HTTPStatus.OK).and_text(is_json_that(has_key("processedSuggestions"))),
         )
 
+    @pytest.mark.parametrize(
+        "headers",
+        [
+            {},  # header missing entirely, valid
+        ],
+    )
+    def test_nhs_number_given_in_path_but_no_nhs_number_header_present(
+        self,
+        client: FlaskClient,
+        persisted_person: NHSNumber,
+        campaign_config: CampaignConfig,  # noqa: ARG002
+        secretsmanager_client: BaseClient,  # noqa: ARG002
+        headers: dict,
+    ):
+        # Given
+        # When
+        response = client.get(f"/patient-check/{persisted_person}", headers=headers)
+
+        # Then
+        assert_that(
+            response,
+            is_response().with_status_code(HTTPStatus.OK).and_text(is_json_that(has_key("processedSuggestions"))),
+        )
+
+    @pytest.mark.parametrize(
+        "headers",
+        [
+            {"nhs-login-nhs-number": None},  # header present but empty, invalid
+            {"nhs-login-nhs-number": ""},  # header present but blank, invalid
+        ],
+    )
+    def test_nhs_number_in_path_and_header_present_but_empty_or_none(
+        self,
+        headers: dict,
+        client: FlaskClient,
+        persisted_person: NHSNumber,
+        campaign_config: CampaignConfig,  # noqa: ARG002
+        secretsmanager_client: BaseClient,  # noqa: ARG002
+    ):
+        # When
+        response = client.get(f"/patient-check/{persisted_person}", headers=headers)
+
+        # Then
+        assert_that(
+            response,
+            is_response()
+            .with_status_code(HTTPStatus.FORBIDDEN)
+            .and_text(is_json_that(has_entries(resourceType="OperationOutcome"))),
+        )
+
+    def test_nhs_number_given_but_header_nhs_number_doesnt_match(
+        self,
+        client: FlaskClient,
+        persisted_person: NHSNumber,
+        campaign_config: CampaignConfig,  # noqa: ARG002
+        secretsmanager_client: BaseClient,  # noqa: ARG002
+    ):
+        # Given
+        headers = {"nhs-login-nhs-number": f"123{persisted_person!s}"}
+
+        # When
+        response = client.get(f"/patient-check/{persisted_person}", headers=headers)
+
+        # Then
+        assert_that(
+            response,
+            is_response()
+            .with_status_code(HTTPStatus.FORBIDDEN)
+            .and_text(is_json_that(has_entries(resourceType="OperationOutcome"))),
+        )
+
     def test_no_nhs_number_given(self, client: FlaskClient):
         # Given
 

tests/integration/lambda/test_app_running_as_lambda.py

@@ -321,7 +321,7 @@ def test_given_nhs_number_in_path_does_not_match_with_nhs_number_in_headers_resu
     )
 
 
-def test_given_nhs_number_not_present_in_headers_results_in_error_response(
+def test_given_nhs_number_not_present_in_headers_results_in_valid_for_application_restricted_users(
     lambda_client: BaseClient,  # noqa:ARG001
     persisted_person: NHSNumber,
     campaign_config: CampaignConfig,  # noqa:ARG001
@@ -335,12 +335,32 @@ def test_given_nhs_number_not_present_in_headers_results_in_error_response(
         timeout=10,
     )
 
+    assert_that(
+        response,
+        is_response().with_status_code(HTTPStatus.OK).and_body(is_json_that(has_key("processedSuggestions"))),
+    )
+
+
+def test_given_nhs_number_key_present_in_headers_have_no_value_results_in_error_response(
+    lambda_client: BaseClient,  # noqa:ARG001
+    persisted_person: NHSNumber,
+    campaign_config: CampaignConfig,  # noqa:ARG001
+    api_gateway_endpoint: URL,
+):
+    # Given
+    # When
+    invoke_url = f"{api_gateway_endpoint}/patient-check/{persisted_person}"
+    response = httpx.get(
+        invoke_url,
+        headers={"nhs-login-nhs-number": ""},
+        timeout=10,
+    )
+
     # Then
     assert_that(
         response,
         is_response()
         .with_status_code(HTTPStatus.FORBIDDEN)
-        .with_headers(has_entries({"Content-Type": "application/fhir+json"}))
         .and_body(
             is_json_that(
                 has_entries(

tests/unit/common/test_request_validator.py

@@ -20,16 +20,15 @@ class TestValidateNHSNumber:
     @pytest.mark.parametrize(
         ("path_nhs", "header_nhs", "expected_result", "expected_log_msg"),
         [
-            (None, None, False, "NHS number is not present in path"),
-            ("1234567890", None, True, None),
-            (None, "1234567890", False, "NHS number is not present in path"),
+            ("1234567890", None, False, "NHS number mismatch"),
+            ("1234567890", "", False, "NHS number mismatch"),
             ("1234567890", "0987654321", False, "NHS number mismatch"),
             ("1234567890", "1234567890", True, None),
         ],
     )
-    def test_validate_nhs_number(self, path_nhs, header_nhs, expected_result, expected_log_msg, caplog):
+    def test_validate_nhs_number_in_header(self, path_nhs, header_nhs, expected_result, expected_log_msg, caplog):
         with caplog.at_level(logging.ERROR):
-            result = request_validator.validate_nhs_number(path_nhs, header_nhs)
+            result = request_validator.validate_nhs_number_in_header(path_nhs, header_nhs)
 
         assert result == expected_result
 
@@ -40,15 +39,22 @@ def test_validate_nhs_number(self, path_nhs, header_nhs, expected_result, expect
 
 
 class TestValidateRequestParams:
-    def test_validate_request_params_success(self, app, caplog):
+    @pytest.mark.parametrize(
+        "headers",
+        [
+            {},  # header missing entirely - request from application restricted consumer
+            {"nhs-login-nhs-number": "1234567890"},  # valid request from consumer
+        ],
+    )
+    def test_validate_request_params_success(self, headers, app, caplog):
         mock_api = MagicMock(return_value="success")
 
         decorator = request_validator.validate_request_params()
         dummy_route = decorator(mock_api)
 
         with app.test_request_context(
             "/dummy?id=1234567890",
-            headers={"nhs-login-nhs-number": "1234567890"},
+            headers=headers,
             method="GET",
         ):
             with caplog.at_level(logging.INFO):
@@ -58,15 +64,23 @@ def test_validate_request_params_success(self, app, caplog):
             assert any("NHS numbers from the request" in record.message for record in caplog.records)
             assert not any(record.levelname == "ERROR" for record in caplog.records)
 
-    def test_validate_request_params_nhs_mismatch(self, app, caplog):
+    @pytest.mark.parametrize(
+        "headers",
+        [
+            {"nhs-login-nhs-number": None},  # not valid
+            {"nhs-login-nhs-number": ""},  # not valid
+            {"nhs-login-nhs-number": "9834567890"},  # not valid, due to mismatch
+        ],
+    )
+    def test_validate_request_params_nhs_mismatch(self, headers, app, caplog):
         mock_api = MagicMock()
 
         decorator = request_validator.validate_request_params()
         dummy_route = decorator(mock_api)
 
         with app.test_request_context(
             "/dummy?id=1234567890",
-            headers={"nhs-login-nhs-number": "0987654321"},
+            headers=headers,
             method="GET",
         ):
             with caplog.at_level(logging.INFO):