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)