Merge pull request #464 from NHSDigital/remove-shared-key-from-mesh-client

Use predefined endpoints for mailbox connectivity

Author: steventux

manage_breast_screening/notifications/services/mesh_inbox.py

@@ -1,19 +1,17 @@
 import os
 from tempfile import NamedTemporaryFile
 
-from mesh_client import MeshClient, Message
+from mesh_client import INT_ENDPOINT, LIVE_ENDPOINT, Endpoint, MeshClient, Message
 
 
 class MeshInbox:
     def __init__(self):
-        cert_file, private_key_file, ca_cert_file = self.ssl_credentials()
+        cert_file, private_key_file = self.ssl_credentials()
         self.client = MeshClient(
-            url=os.getenv("NBSS_MESH_HOST"),
+            self.endpoint_for_env(),
             mailbox=os.getenv("NBSS_MESH_INBOX_NAME"),
             password=os.getenv("NBSS_MESH_PASSWORD"),
-            shared_key=os.getenv("NBSS_MESH_SHARED_KEY"),
             cert=(cert_file, private_key_file),
-            verify=ca_cert_file,
         )
         self.client.handshake()
 
@@ -26,6 +24,15 @@ def fetch_message(self, message_id: str) -> Message:
     def acknowledge(self, message_id: str):
         self.client.acknowledge_message(message_id)
 
+    def endpoint_for_env(self) -> Endpoint:
+        current_environment = os.getenv("DJANGO_ENV", "dev")
+        if current_environment == "production":
+            return LIVE_ENDPOINT
+        elif current_environment == "test":
+            return Endpoint(os.getenv("NBSS_MESH_HOST"), None, None, False, False)
+        else:
+            return INT_ENDPOINT
+
     def __enter__(self):
         return self
 
@@ -36,12 +43,10 @@ def __exit__(self, type_, value, tb):
     def ssl_credentials(cls) -> tuple[NamedTemporaryFile]:
         cert = os.getenv("NBSS_MESH_CERT")
         private_key = os.getenv("NBSS_MESH_PRIVATE_KEY")
-        ca_cert = os.getenv("NBSS_MESH_CA_CERT")
 
         return (
             cls.to_file(cert).name,
             cls.to_file(private_key).name,
-            cls.to_file(ca_cert).name,
         )
 
     @staticmethod

manage_breast_screening/notifications/tests/end_to_end/test_get_from_mesh_and_send_message_batch.py

@@ -24,6 +24,7 @@
 class TestEndToEnd:
     @pytest.fixture(autouse=True)
     def setup(self, monkeypatch):
+        monkeypatch.setenv("DJANGO_ENV", "test")
         monkeypatch.setenv("NBSS_MESH_HOST", "http://localhost:8700")
         monkeypatch.setenv("NBSS_MESH_PASSWORD", "password")
         monkeypatch.setenv("NBSS_MESH_SHARED_KEY", "TestKey")

manage_breast_screening/notifications/tests/integration/test_store_mesh_messages.py

@@ -15,9 +15,9 @@
 class TestStoreMeshMessages:
     @pytest.fixture(autouse=True)
     def setup(self, monkeypatch):
+        monkeypatch.setenv("DJANGO_ENV", "test")
         monkeypatch.setenv("NBSS_MESH_HOST", "http://localhost:8700")
         monkeypatch.setenv("NBSS_MESH_PASSWORD", "password")
-        monkeypatch.setenv("NBSS_MESH_SHARED_KEY", "TestKey")
         monkeypatch.setenv("NBSS_MESH_INBOX_NAME", "X26ABC1")
         monkeypatch.setenv("NBSS_MESH_CERT", "mesh-cert")
         monkeypatch.setenv("NBSS_MESH_PRIVATE_KEY", "mesh-private-key")
@@ -45,7 +45,6 @@ def test_retrieve_file(self, helpers):
             url=os.getenv("NBSS_MESH_HOST"),
             mailbox=os.getenv("NBSS_MESH_INBOX_NAME"),
             password=os.getenv("NBSS_MESH_PASSWORD"),
-            shared_key=os.getenv("NBSS_MESH_SHARED_KEY"),
         ) as client:
             assert len(client.list_messages()) == 2
 

manage_breast_screening/notifications/tests/services/test_mesh_inbox.py

@@ -1,6 +1,7 @@
 from unittest.mock import ANY, MagicMock, patch
 
 import pytest
+from mesh_client import INT_ENDPOINT, LIVE_ENDPOINT
 
 from manage_breast_screening.notifications.services.mesh_inbox import MeshInbox
 
@@ -9,42 +10,52 @@
 class TestMeshInbox:
     @pytest.fixture(autouse=True)
     def setup(self, monkeypatch):
-        monkeypatch.setenv("NBSS_MESH_HOST", "https://mesh.test")
+        monkeypatch.setenv("DJANGO_ENV", "dev")
         monkeypatch.setenv("NBSS_MESH_INBOX_NAME", "mesh-inbox-name")
         monkeypatch.setenv("NBSS_MESH_PASSWORD", "mesh-password")
-        monkeypatch.setenv("NBSS_MESH_SHARED_KEY", "mesh-shared-key")
         monkeypatch.setenv("NBSS_MESH_CERT", "mesh-cert")
         monkeypatch.setenv("NBSS_MESH_PRIVATE_KEY", "mesh-private-key")
-        monkeypatch.setenv("NBSS_MESH_CA_CERT", "mesh-ca-cert")
 
     def test_client_initialises(self, mock_mesh_client):
         with patch.object(
             MeshInbox,
             "ssl_credentials",
-            return_value=("cert", "private-key", "ca-cert"),
+            return_value=("cert", "private-key"),
         ):
             MeshInbox()
 
         mock_mesh_client.assert_called_once_with(
-            url="https://mesh.test",
+            INT_ENDPOINT,
+            mailbox="mesh-inbox-name",
+            password="mesh-password",
+            cert=("cert", "private-key"),
+        )
+
+    def test_client_initialises_with_prod_endpoint(self, mock_mesh_client, monkeypatch):
+        monkeypatch.setenv("DJANGO_ENV", "production")
+        with patch.object(
+            MeshInbox,
+            "ssl_credentials",
+            return_value=("cert", "private-key"),
+        ):
+            MeshInbox()
+
+        mock_mesh_client.assert_called_once_with(
+            LIVE_ENDPOINT,
             mailbox="mesh-inbox-name",
             password="mesh-password",
-            shared_key="mesh-shared-key",
             cert=("cert", "private-key"),
-            verify="ca-cert",
         )
 
     def test_constructor_as_contextmanager(self, mock_mesh_client):
         with MeshInbox() as inbox:
             inbox.fetch_message_ids()
 
         mock_mesh_client.assert_called_once_with(
-            url="https://mesh.test",
+            INT_ENDPOINT,
             mailbox="mesh-inbox-name",
             password="mesh-password",
-            shared_key="mesh-shared-key",
             cert=(ANY, ANY),
-            verify=ANY,
         )
         mock_mesh_client.return_value.list_messages.assert_called_once()
         mock_mesh_client.return_value.close.assert_called_once()
@@ -54,21 +65,16 @@ def test_ssl_credentials(self, _unused_mock):
         mock_cert_file.name = "cert"
         mock_private_key_file = MagicMock()
         mock_private_key_file.name = "private-key"
-        mock_ca_cert_file = MagicMock()
-        mock_ca_cert_file.name = "ca-cert"
 
         with patch.object(
             MeshInbox,
             "to_file",
-            side_effect=[mock_cert_file, mock_private_key_file, mock_ca_cert_file],
+            side_effect=[mock_cert_file, mock_private_key_file],
         ):
-            cert_file_name, private_key_file_name, ca_cert_file_name = (
-                MeshInbox.ssl_credentials()
-            )
+            cert_file_name, private_key_file_name = MeshInbox.ssl_credentials()
 
         assert cert_file_name == "cert"
         assert private_key_file_name == "private-key"
-        assert ca_cert_file_name == "ca-cert"
 
     def test_fetch_message_ids(self, mock_mesh_client):
         MeshInbox().fetch_message_ids()