[NDR-284] Add E2E test for invalid common name (#876)

Author: megan-bower4

.gitignore

@@ -27,6 +27,7 @@ share/python-wheels/
 MANIFEST
 .pip_cache/
 package/
+lambdas/mtls_env_certs/
 
 # PyInstaller
 #  Usually these files are written by a python script from a template

lambdas/tests/e2e/api/fhir/conftest.py

@@ -14,9 +14,9 @@
 PDM_SNOMED = 717391000000106
 PDM_METADATA_TABLE = os.environ.get("PDM_METADATA_TABLE")
 PDM_S3_BUCKET = os.environ.get("PDM_S3_BUCKET") or ""
-MTLS_ENDPOINT = os.environ["MTLS_ENDPOINT"]
-CLIENT_CERT_PATH = os.environ["CLIENT_CERT_PATH"]
-CLIENT_KEY_PATH = os.environ["CLIENT_KEY_PATH"]
+MTLS_ENDPOINT = os.environ.get("MTLS_ENDPOINT")
+CLIENT_CERT_PATH = os.environ.get("CLIENT_CERT_PATH")
+CLIENT_KEY_PATH = os.environ.get("CLIENT_KEY_PATH")
 
 
 @pytest.fixture
@@ -47,9 +47,11 @@ def fetch_with_retry_mtls(
     raise Exception("Condition not met within retry limit")
 
 
-def create_mtls_session():
+def create_mtls_session(
+    client_cert_path=CLIENT_CERT_PATH, client_key_path=CLIENT_KEY_PATH
+):
     session = requests.Session()
-    session.cert = (CLIENT_CERT_PATH, CLIENT_KEY_PATH)
+    session.cert = (client_cert_path, client_key_path)
     return session
 
 
@@ -83,3 +85,36 @@ def temp_cert_and_key():
         yield cert_path, key_path
     finally:
         shutil.rmtree(temp_dir)
+
+
+def get_pdm_document_reference(record_id, client_cert_path=None, client_key_path=None):
+    url = f"https://{MTLS_ENDPOINT}/DocumentReference/{PDM_SNOMED}~{record_id}"
+    headers = {
+        "X-Correlation-Id": "1234",
+    }
+
+    # Call with invalid or unauthorised certs
+    if client_cert_path and client_key_path:
+        session = create_mtls_session(client_cert_path, client_key_path)
+    else:
+        # Call with default valid certs
+        session = create_mtls_session()
+
+    response = session.get(url, headers=headers)
+    return response
+
+
+def create_and_store_pdm_record(
+    test_data,
+    nhs_number: str = "9912003071",
+    doc_status: str | None = None,
+    size: int | None = None,
+):
+    """Helper to create metadata and resource for a record."""
+    record = pdm_data_helper.build_record(
+        nhs_number=nhs_number, doc_status=doc_status, size=size
+    )
+    test_data.append(record)
+    pdm_data_helper.create_metadata(record)
+    pdm_data_helper.create_resource(record)
+    return record

lambdas/tests/e2e/api/fhir/test_mtls_routing.py

@@ -0,0 +1,19 @@
+import os
+import uuid
+
+from lambdas.tests.e2e.api.fhir.conftest import get_pdm_document_reference
+
+UNAUTHORISED_CLIENT_CERT_PATH = os.environ.get("UNAUTHORISED_CLIENT_CERT_PATH")
+UNAUTHORISED_CLIENT_KEY_PATH = os.environ.get("UNAUTHORISED_CLIENT_KEY_PATH")
+
+
+def test_mtls_invalid_common_name():
+    record_id = str(uuid.uuid4())
+    response = get_pdm_document_reference(
+        record_id, UNAUTHORISED_CLIENT_CERT_PATH, UNAUTHORISED_CLIENT_KEY_PATH
+    )
+    assert response.status_code == 400
+
+    data = response.json()
+    diagnostics = data.get("issue", [{}])[0].get("diagnostics", "")
+    assert diagnostics == "Invalid document type requested"

lambdas/tests/e2e/api/fhir/test_retrieve_document_fhir_api.py

@@ -1,43 +1,25 @@
 import base64
-import io
+import uuid
 
 import pytest
-import requests
 from tests.e2e.helpers.data_helper import PdmDataHelper
 
 from lambdas.tests.e2e.api.fhir.conftest import (
-    MTLS_ENDPOINT,
     PDM_S3_BUCKET,
-    create_mtls_session,
+    create_and_store_pdm_record,
+    get_pdm_document_reference,
 )
-from lambdas.tests.e2e.conftest import PDM_SNOMED
 
 pdm_data_helper = PdmDataHelper()
 
 
-def get_document_reference(record_id):
-    """Helper to perform GET request for DocumentReference."""
-    url = f"https://{MTLS_ENDPOINT}/DocumentReference/{PDM_SNOMED}~{record_id}"
-    headers = {
-        "X-Correlation-Id": "1234",
-    }
-    session = create_mtls_session()
-    return session.get(url, headers=headers)
-
-
 @pytest.mark.parametrize("file_size", [None, 10 * 1024 * 1024])
 def test_file_retrieval(test_data, file_size):
     """Test retrieval for small and large files."""
-    pdm_record = pdm_data_helper.build_record(
-        data=io.BytesIO(b"A" * file_size) if file_size else None,
-        size=file_size * 1024 if file_size else None,
+    pdm_record = create_and_store_pdm_record(
+        test_data, size=file_size if file_size else None
     )
-    test_data.append(pdm_record)
-
-    pdm_data_helper.create_metadata(pdm_record)
-    pdm_data_helper.create_resource(pdm_record)
-
-    response = get_document_reference(pdm_record["id"])
+    response = get_pdm_document_reference(pdm_record["id"])
     assert response.status_code == 200
 
     json = response.json()
@@ -56,15 +38,15 @@ def test_file_retrieval(test_data, file_size):
 
 
 @pytest.mark.parametrize(
-    "nhs_id,expected_status,expected_code,expected_diagnostics",
+    "record_id,expected_status,expected_code,expected_diagnostics",
     [
-        ("9912003071", 404, "RESOURCE_NOT_FOUND", "Document reference not found"),
+        (str(uuid.uuid4()), 404, "RESOURCE_NOT_FOUND", "Document reference not found"),
     ],
 )
 def test_retrieve_edge_cases(
-    nhs_id, expected_status, expected_code, expected_diagnostics
+    record_id, expected_status, expected_code, expected_diagnostics
 ):
-    response = get_document_reference(nhs_id)
+    response = get_pdm_document_reference(record_id)
     assert response.status_code == expected_status
 
     body = response.json()
@@ -78,30 +60,25 @@ def test_retrieve_edge_cases(
 
 
 def test_preliminary_file(test_data):
-    pdm_record = pdm_data_helper.build_record(doc_status="preliminary")
-    test_data.append(pdm_record)
-    pdm_data_helper.create_metadata(pdm_record)
-    pdm_data_helper.create_resource(pdm_record)
+    pdm_record = create_and_store_pdm_record(test_data, doc_status="preliminary")
 
-    response = get_document_reference(pdm_record["id"])
+    response = get_pdm_document_reference(pdm_record["id"])
     assert response.status_code == 200
 
     response_json = response.json()
     assert response_json.get("docStatus") == "preliminary"
 
 
 def test_forbidden_with_invalid_cert(test_data, temp_cert_and_key):
-    pdm_record = pdm_data_helper.build_record()
-    test_data.append(pdm_record)
-    pdm_data_helper.create_metadata(pdm_record)
-    pdm_data_helper.create_resource(pdm_record)
+    pdm_record = create_and_store_pdm_record(test_data)
 
     # Use an invalid cert that is trusted by TLS but fails truststore validation
     cert_path, key_path = temp_cert_and_key
-    url = f"https://{MTLS_ENDPOINT}/DocumentReference/{PDM_SNOMED}~{pdm_record['id']}"
-    headers = {"Authorization": "Bearer 123", "X-Correlation-Id": "1234"}
 
-    response = requests.get(url, headers=headers, cert=(cert_path, key_path))
+    response = get_pdm_document_reference(
+        pdm_record["id"], client_cert_path=cert_path, client_key_path=key_path
+    )
+
     body = response.json()
     assert response.status_code == 403
     assert body["message"] == "Forbidden"

lambdas/tests/e2e/api/fhir/test_search_patient_fhir_api.py

@@ -1,40 +1,37 @@
 import pytest
-import requests
 
-from lambdas.tests.e2e.api.fhir.conftest import MTLS_ENDPOINT, create_mtls_session
+from lambdas.tests.e2e.api.fhir.conftest import (
+    MTLS_ENDPOINT,
+    create_and_store_pdm_record,
+    create_mtls_session,
+)
 from lambdas.tests.e2e.conftest import APIM_ENDPOINT, PDM_SNOMED
 from lambdas.tests.e2e.helpers.data_helper import PdmDataHelper
 
 pdm_data_helper = PdmDataHelper()
 
 
-def search_document_reference(nhs_number):
-    """Helper to perform search by NHS number."""
+def search_document_reference(nhs_number, client_cert_path=None, client_key_path=None):
+    """Helper to perform search by NHS number with optional mTLS certs."""
     url = (
         f"https://{MTLS_ENDPOINT}/DocumentReference?"
         f"subject:identifier=https://fhir.nhs.uk/Id/nhs-number|{nhs_number}"
     )
     headers = {
-        "Authorization": "Bearer 123",
         "X-Correlation-Id": "1234",
     }
-    session = create_mtls_session()
-    return session.get(url, headers=headers)
 
+    # Use provided certs if available, else defaults
+    if client_cert_path and client_key_path:
+        session = create_mtls_session(client_cert_path, client_key_path)
+    else:
+        session = create_mtls_session()
 
-def create_and_store_record(
-    test_data, nhs_number="9912003071", doc_status: str | None = None
-):
-    """Helper to create metadata and resource for a record."""
-    record = pdm_data_helper.build_record(nhs_number=nhs_number, doc_status=doc_status)
-    test_data.append(record)
-    pdm_data_helper.create_metadata(record)
-    pdm_data_helper.create_resource(record)
-    return record
+    return session.get(url, headers=headers)
 
 
 def test_search_patient_details(test_data):
-    create_and_store_record(test_data)
+    create_and_store_pdm_record(test_data)
 
     response = search_document_reference("9912003071")
     assert response.status_code == 200
@@ -50,8 +47,8 @@ def test_search_patient_details(test_data):
 
 
 def test_multiple_cancelled_search_patient_details(test_data):
-    create_and_store_record(test_data, doc_status="cancelled")
-    create_and_store_record(test_data, doc_status="cancelled")
+    create_and_store_pdm_record(test_data, doc_status="cancelled")
+    create_and_store_pdm_record(test_data, doc_status="cancelled")
 
     response = search_document_reference("9912003071")
     assert response.status_code == 200
@@ -90,20 +87,15 @@ def test_search_edge_cases(
 
 def test_search_patient_unauthorized_mtls(test_data, temp_cert_and_key):
     """Search should return 403 when mTLS certificate is invalid or missing."""
-    create_and_store_record(test_data)
-    url = (
-        f"https://{MTLS_ENDPOINT}/DocumentReference?"
-        f"subject:identifier=https://fhir.nhs.uk/Id/nhs-number|9912003071"
-    )
-    headers = {
-        "Authorization": "Bearer 123",
-        "X-Correlation-Id": "unauthorized-test",
-    }
+    create_and_store_pdm_record(test_data)
 
     # Use an invalid cert that is trusted by TLS but fails truststore validation
     cert_path, key_path = temp_cert_and_key
 
-    response = requests.get(url, headers=headers, cert=(cert_path, key_path))
+    response = search_document_reference(
+        "9912003071", client_cert_path=cert_path, client_key_path=key_path
+    )
+
     body = response.json()
     assert response.status_code == 403
     assert body["message"] == "Forbidden"