Add support for mTLS

Author: xrmx

opamp/opentelemetry-opamp-client/src/opentelemetry/_opamp/client.py

@@ -63,6 +63,10 @@ def __init__(
         agent_identifying_attributes: Mapping[str, AnyValue],
         agent_non_identifying_attributes: Mapping[str, AnyValue] | None = None,
         transport: HttpTransport | None = None,
+        # this matches requests but can be mapped to other http libraries APIs
+        tls_certificate: str | bool = True,
+        tls_client_certificate: str | None = None,
+        tls_client_key: str | None = None,
     ):
         self._timeout_millis = timeout_millis
         self._transport = (
@@ -72,6 +76,9 @@ def __init__(
         self._endpoint = endpoint
         headers = headers or {}
         self._headers = {**_OPAMP_HTTP_HEADERS, **headers}
+        self._tls_certificate = tls_certificate
+        self._tls_client_certificate = tls_client_certificate
+        self._tls_client_key = tls_client_key
 
         self._agent_description = messages.build_agent_description(
             identifying_attributes=agent_identifying_attributes,
@@ -159,6 +166,9 @@ def send(self, data: bytes):
                 headers=self._headers,
                 data=data,
                 timeout_millis=self._timeout_millis,
+                tls_certificate=self._tls_certificate,
+                tls_client_certificate=self._tls_client_certificate,
+                tls_client_key=self._tls_client_key,
             )
             return response
         finally:

opamp/opentelemetry-opamp-client/src/opentelemetry/_opamp/transport/base.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from __future__ import annotations
+
 import abc
 from typing import Mapping
 
@@ -26,9 +28,13 @@ class HttpTransport(abc.ABC):
     @abc.abstractmethod
     def send(
         self,
+        *,
         url: str,
         headers: Mapping[str, str],
         data: bytes,
         timeout_millis: int,
+        tls_certificate: str | bool,
+        tls_client_certificate: str | None = None,
+        tls_client_key: str | None = None,
     ) -> opamp_pb2.ServerToAgent:
         pass

opamp/opentelemetry-opamp-client/src/opentelemetry/_opamp/transport/requests.py

@@ -32,16 +32,30 @@ def __init__(self, session: requests.Session | None = None):
 
     def send(
         self,
+        *,
         url: str,
         headers: Mapping[str, str],
         data: bytes,
         timeout_millis: int,
+        tls_certificate: str | bool,
+        tls_client_certificate: str | None = None,
+        tls_client_key: str | None = None,
     ):
         headers = {**base_headers, **headers}
         timeout: float = timeout_millis / 1e3
+        client_cert = (
+            (tls_client_certificate, tls_client_key)
+            if tls_client_certificate and tls_client_key
+            else tls_client_certificate
+        )
         try:
             response = self.session.post(
-                url, headers=headers, data=data, timeout=timeout
+                url,
+                headers=headers,
+                data=data,
+                timeout=timeout,
+                verify=tls_certificate,
+                cert=client_cert,
             )
             response.raise_for_status()
         except Exception as exc:

opamp/opentelemetry-opamp-client/tests/opamp/test_client.py

@@ -53,6 +53,9 @@ def test_can_instantiate_opamp_client_with_defaults():
         "Content-Type": "application/x-protobuf",
         "User-Agent": "OTel-OpAMP-Python/" + __version__,
     }
+    assert client._tls_certificate is True
+    assert client._tls_client_certificate is None
+    assert client._tls_client_key is None
     assert client._timeout_millis == 1_000
     assert client._sequence_num == 0
     assert isinstance(client._instance_uid, bytes)
@@ -72,6 +75,9 @@ def test_can_instantiate_opamp_client_all_params():
         agent_identifying_attributes={"foo": "bar"},
         agent_non_identifying_attributes={"bar": "baz"},
         transport=transport,
+        tls_certificate="ca.pem",
+        tls_client_certificate="client.pem",
+        tls_client_key="client-key.pem",
     )
 
     assert client
@@ -80,6 +86,9 @@ def test_can_instantiate_opamp_client_all_params():
         "User-Agent": "OTel-OpAMP-Python/" + __version__,
         "an": "header",
     }
+    assert client._tls_certificate == "ca.pem"
+    assert client._tls_client_certificate == "client.pem"
+    assert client._tls_client_key == "client-key.pem"
     assert client._timeout_millis == 2_000
     assert client._sequence_num == 0
     assert isinstance(client._instance_uid, bytes)
@@ -111,6 +120,9 @@ def test_client_headers_override_defaults():
         },
         data=b"",
         timeout_millis=1000,
+        tls_certificate=True,
+        tls_client_certificate=None,
+        tls_client_key=None,
     )
 
 
@@ -331,6 +343,9 @@ def test_send(client):
         },
         data=b"foo",
         timeout_millis=1000,
+        tls_certificate=True,
+        tls_client_certificate=None,
+        tls_client_key=None,
     )
 
 

opamp/opentelemetry-opamp-client/tests/opamp/transport/test_requests.py

@@ -47,17 +47,104 @@ def test_can_send():
     with mock.patch.object(transport, "session") as session_mock:
         session_mock.post.return_value = response_mock
         response = transport.send(
-            "http://127.0.0.1/v1/opamp",
+            url="http://127.0.0.1/v1/opamp",
             headers=headers,
             data=data,
             timeout_millis=1_000,
+            tls_certificate=True,
         )
 
         session_mock.post.assert_called_once_with(
             "http://127.0.0.1/v1/opamp",
             headers=expected_headers,
             data=data,
             timeout=1,
+            verify=True,
+            cert=None,
+        )
+
+    assert isinstance(response, opamp_pb2.ServerToAgent)
+
+
+def test_send_tls_certificate_mapped_to_verify():
+    transport = RequestsTransport()
+    serialized_message = opamp_pb2.ServerToAgent().SerializeToString()
+    response_mock = mock.Mock(content=serialized_message)
+    data = b""
+    with mock.patch.object(transport, "session") as session_mock:
+        session_mock.post.return_value = response_mock
+        response = transport.send(
+            url="https://127.0.0.1/v1/opamp",
+            headers={},
+            data=data,
+            timeout_millis=1_000,
+            tls_certificate=False,
+        )
+
+        session_mock.post.assert_called_once_with(
+            "https://127.0.0.1/v1/opamp",
+            headers=base_headers,
+            data=data,
+            timeout=1,
+            verify=False,
+            cert=None,
+        )
+
+    assert isinstance(response, opamp_pb2.ServerToAgent)
+
+
+def test_send_mtls():
+    transport = RequestsTransport()
+    serialized_message = opamp_pb2.ServerToAgent().SerializeToString()
+    response_mock = mock.Mock(content=serialized_message)
+    data = b""
+    with mock.patch.object(transport, "session") as session_mock:
+        session_mock.post.return_value = response_mock
+        response = transport.send(
+            url="https://127.0.0.1/v1/opamp",
+            headers={},
+            data=data,
+            timeout_millis=1_000,
+            tls_certificate="server.pem",
+            tls_client_certificate="client.pem",
+            tls_client_key="client.key",
+        )
+
+        session_mock.post.assert_called_once_with(
+            "https://127.0.0.1/v1/opamp",
+            headers=base_headers,
+            data=data,
+            timeout=1,
+            verify="server.pem",
+            cert=("client.pem", "client.key"),
+        )
+
+    assert isinstance(response, opamp_pb2.ServerToAgent)
+
+
+def test_send_mtls_no_client_key():
+    transport = RequestsTransport()
+    serialized_message = opamp_pb2.ServerToAgent().SerializeToString()
+    response_mock = mock.Mock(content=serialized_message)
+    data = b""
+    with mock.patch.object(transport, "session") as session_mock:
+        session_mock.post.return_value = response_mock
+        response = transport.send(
+            url="https://127.0.0.1/v1/opamp",
+            headers={},
+            data=data,
+            timeout_millis=1_000,
+            tls_certificate="server.pem",
+            tls_client_certificate="client.pem",
+        )
+
+        session_mock.post.assert_called_once_with(
+            "https://127.0.0.1/v1/opamp",
+            headers=base_headers,
+            data=data,
+            timeout=1,
+            verify="server.pem",
+            cert="client.pem",
         )
 
     assert isinstance(response, opamp_pb2.ServerToAgent)
@@ -74,15 +161,18 @@ def test_send_exceptions_raises_opamp_exception():
         response_mock.raise_for_status.side_effect = Exception
         with pytest.raises(OpAMPException):
             transport.send(
-                "http://127.0.0.1/v1/opamp",
+                url="http://127.0.0.1/v1/opamp",
                 headers=headers,
                 data=data,
                 timeout_millis=1_000,
+                tls_certificate=True,
             )
 
         session_mock.post.assert_called_once_with(
             "http://127.0.0.1/v1/opamp",
             headers=expected_headers,
             data=data,
             timeout=1,
+            verify=True,
+            cert=None,
         )