ELI-578 consumer_id error validation

Author: Karthikeyannhs

src/eligibility_signposting_api/common/api_error_response.py

@@ -135,3 +135,11 @@ def log_and_generate_response(
     fhir_error_code=FHIRSpineErrorCode.ACCESS_DENIED,
     fhir_display_message="Access has been denied to process this request.",
 )
+
+CONSUMER_ID_NOT_PROVIDED_ERROR = APIErrorResponse(
+    status_code=HTTPStatus.FORBIDDEN,
+    fhir_issue_code=FHIRIssueCode.FORBIDDEN,
+    fhir_issue_severity=FHIRIssueSeverity.ERROR,
+    fhir_error_code=FHIRSpineErrorCode.ACCESS_DENIED,
+    fhir_display_message="Access has been denied to process this request.",
+)

src/eligibility_signposting_api/common/request_validator.py

@@ -7,12 +7,13 @@
 from flask.typing import ResponseReturnValue
 
 from eligibility_signposting_api.common.api_error_response import (
+    CONSUMER_ID_NOT_PROVIDED_ERROR,
     INVALID_CATEGORY_ERROR,
     INVALID_CONDITION_FORMAT_ERROR,
     INVALID_INCLUDE_ACTIONS_ERROR,
     NHS_NUMBER_MISMATCH_ERROR,
 )
-from eligibility_signposting_api.config.constants import NHS_NUMBER_HEADER
+from eligibility_signposting_api.config.constants import CONSUMER_ID, NHS_NUMBER_HEADER
 
 logger = logging.getLogger(__name__)
 
@@ -56,6 +57,14 @@ def validate_request_params() -> Callable:
     def decorator(func: Callable) -> Callable:
         @wraps(func)
         def wrapper(*args, **kwargs) -> ResponseReturnValue:  # noqa:ANN002,ANN003
+            consumer_id = str(request.headers.get(CONSUMER_ID))
+
+            if not consumer_id:
+                message = "You are not authorised to request"
+                return CONSUMER_ID_NOT_PROVIDED_ERROR.log_and_generate_response(
+                    log_message=message, diagnostics=message
+                )
+
             path_nhs_number = str(kwargs.get("nhs_number"))
             header_nhs_no = str(request.headers.get(NHS_NUMBER_HEADER))
 

src/eligibility_signposting_api/config/constants.py

@@ -3,4 +3,5 @@
 URL_PREFIX = "patient-check"
 RULE_STOP_DEFAULT = False
 NHS_NUMBER_HEADER = "nhs-login-nhs-number"
+CONSUMER_ID = "consumer-id"
 ALLOWED_CONDITIONS = Literal["COVID", "FLU", "MMR", "RSV"]

src/eligibility_signposting_api/views/eligibility.py

@@ -13,7 +13,8 @@
 from eligibility_signposting_api.audit.audit_service import AuditService
 from eligibility_signposting_api.common.api_error_response import NHS_NUMBER_NOT_FOUND_ERROR
 from eligibility_signposting_api.common.request_validator import validate_request_params
-from eligibility_signposting_api.config.constants import URL_PREFIX
+from eligibility_signposting_api.config.constants import CONSUMER_ID, URL_PREFIX
+from eligibility_signposting_api.model.consumer_mapping import ConsumerId
 from eligibility_signposting_api.model.eligibility_status import Condition, EligibilityStatus, NHSNumber, Status
 from eligibility_signposting_api.services import EligibilityService, UnknownPersonError
 from eligibility_signposting_api.views.response_model import eligibility_response
@@ -48,13 +49,14 @@ def check_eligibility(
 ) -> ResponseReturnValue:
     logger.info("checking nhs_number %r in %r", nhs_number, eligibility_service, extra={"nhs_number": nhs_number})
     try:
-        query_params = get_or_default_query_params()
+        query_params = _get_or_default_query_params()
+        consumer_id = _get_consumer_id_from_headers()
         eligibility_status = eligibility_service.get_eligibility_status(
             nhs_number,
             query_params["includeActions"],
             query_params["conditions"],
             query_params["category"],
-            request.headers.get("Consumer-ID"),
+            consumer_id,
         )
     except UnknownPersonError:
         return handle_unknown_person_error(nhs_number)
@@ -64,7 +66,14 @@ def check_eligibility(
         return make_response(response.model_dump(by_alias=True, mode="json", exclude_none=True), HTTPStatus.OK)
 
 
-def get_or_default_query_params() -> dict[str, Any]:
+def _get_consumer_id_from_headers() -> ConsumerId:
+    """
+    @validate_request_params() ensures the consumer ID is never null at this stage.
+    """
+    return ConsumerId(request.headers.get(CONSUMER_ID, ""))
+
+
+def _get_or_default_query_params() -> dict[str, Any]:
     default_query_params = {"category": "ALL", "conditions": ["ALL"], "includeActions": "Y"}
 
     if not request.args:

tests/integration/in_process/test_eligibility_endpoint.py

@@ -13,6 +13,7 @@
     has_key,
 )
 
+from eligibility_signposting_api.config.constants import CONSUMER_ID
 from eligibility_signposting_api.model.campaign_config import CampaignConfig
 from eligibility_signposting_api.model.consumer_mapping import ConsumerMapping
 from eligibility_signposting_api.model.eligibility_status import (
@@ -30,7 +31,7 @@ def test_nhs_number_given(
         secretsmanager_client: BaseClient,  # noqa: ARG002
     ):
         # Given
-        headers = {"nhs-login-nhs-number": str(persisted_person), "Consumer-ID": "23-mic7heal-jor6don"}
+        headers = {"nhs-login-nhs-number": str(persisted_person), CONSUMER_ID: "23-mic7heal-jor6don"}
 
         # When
         response = client.get(f"/patient-check/{persisted_person}", headers=headers)

tests/unit/views/test_eligibility.py

@@ -31,8 +31,7 @@
 from eligibility_signposting_api.views.eligibility import (
     build_actions,
     build_eligibility_cohorts,
-    build_suitability_results,
-    get_or_default_query_params,
+    build_suitability_results, _get_or_default_query_params,
 )
 from eligibility_signposting_api.views.response_model import eligibility_response
 from tests.fixtures.builders.model.eligibility import (
@@ -509,7 +508,7 @@ def test_build_response_include_values_that_are_not_null(client: FlaskClient):
 
 def test_get_or_default_query_params_with_no_args(app: Flask):
     with app.test_request_context("/patient-check"):
-        result = get_or_default_query_params()
+        result = _get_or_default_query_params()
 
         expected = {"category": "ALL", "conditions": ["ALL"], "includeActions": "Y"}
 
@@ -518,7 +517,7 @@ def test_get_or_default_query_params_with_no_args(app: Flask):
 
 def test_get_or_default_query_params_with_all_args(app: Flask):
     with app.test_request_context("/patient-check?includeActions=Y&category=VACCINATIONS&conditions=FLU"):
-        result = get_or_default_query_params()
+        result = _get_or_default_query_params()
 
         expected = {"includeActions": "Y", "category": "VACCINATIONS", "conditions": ["FLU"]}
 
@@ -527,7 +526,7 @@ def test_get_or_default_query_params_with_all_args(app: Flask):
 
 def test_get_or_default_query_params_with_partial_args(app: Flask):
     with app.test_request_context("/patient-check?includeActions=N"):
-        result = get_or_default_query_params()
+        result = _get_or_default_query_params()
 
         expected = {"includeActions": "N", "category": "ALL", "conditions": ["ALL"]}
 
@@ -536,13 +535,13 @@ def test_get_or_default_query_params_with_partial_args(app: Flask):
 
 def test_get_or_default_query_params_with_lowercase_y(app: Flask):
     with app.test_request_context("/patient-check?includeActions=y"):
-        result = get_or_default_query_params()
+        result = _get_or_default_query_params()
         assert_that(result["includeActions"], is_("Y"))
 
 
 def test_get_or_default_query_params_missing_include_actions(app: Flask):
     with app.test_request_context("/patient-check?category=SCREENING&conditions=COVID19,FLU"):
-        result = get_or_default_query_params()
+        result = _get_or_default_query_params()
 
         expected = {"includeActions": "Y", "category": "SCREENING", "conditions": ["COVID19", "FLU"]}