testing mutual authentication

Author: DanielePalaia

.ci/conf/rabbitmq.conf

@@ -5,3 +5,5 @@ ssl_options.certfile = /etc/rabbitmq/certs/server_certificate.pem
 ssl_options.keyfile = /etc/rabbitmq/certs/server_key.pem
 listeners.ssl.default = 5671
 stream.listeners.ssl.default = 5551
+ssl_options.verify               = verify_peer
+ssl_options.fail_if_no_peer_cert = false

rabbitmq_amqp_python_client/__init__.py

@@ -22,7 +22,10 @@
     QuorumQueueSpecification,
     StreamSpecification,
 )
-from .ssl_configuration import SSlConfigurationContext
+from .ssl_configuration import (
+    ClientCert,
+    SSlConfigurationContext,
+)
 
 try:
     __version__ = metadata.version(__package__)
@@ -54,4 +57,5 @@
     "AMQPMessagingHandler",
     "ArgumentOutOfRangeException",
     "SSlConfigurationContext",
+    "ClientCert",
 ]

tests/conftest.py

@@ -3,6 +3,7 @@
 from rabbitmq_amqp_python_client import (
     AddressHelper,
     AMQPMessagingHandler,
+    ClientCert,
     Connection,
     Event,
     SSlConfigurationContext,
@@ -23,9 +24,15 @@ def connection(pytestconfig):
 
 @pytest.fixture()
 def connection_ssl(pytestconfig):
+    ca_cert_file = ".ci/certs/ca_certificate.pem"
+    client_cert = ".ci/certs/client_certificate.pem"
+    client_key = ".ci/certs/client_key.pem"
     connection = Connection(
         "amqps://guest:guest@localhost:5671/",
-        ssl_context=SSlConfigurationContext(ca_cert=".ci/certs/ca_certificate.pem"),
+        ssl_context=SSlConfigurationContext(
+            ca_cert=ca_cert_file,
+            client_cert=ClientCert(client_cert=client_cert, client_key=client_key),
+        ),
     )
     connection.dial()
     try:

tests/test_connection.py

@@ -1,7 +1,25 @@
-from rabbitmq_amqp_python_client import Connection
+from rabbitmq_amqp_python_client import (
+    ClientCert,
+    Connection,
+    SSlConfigurationContext,
+)
 
 
 def test_connection() -> None:
     connection = Connection("amqp://guest:guest@localhost:5672/")
     connection.dial()
     connection.close()
+
+
+def test_connection_ssl() -> None:
+    ca_cert_file = ".ci/certs/ca_certificate.pem"
+    client_cert = ".ci/certs/client_certificate.pem"
+    client_key = ".ci/certs/client_key.pem"
+    connection = Connection(
+        "amqps://guest:guest@localhost:5671/",
+        ssl_context=SSlConfigurationContext(
+            ca_cert=ca_cert_file,
+            client_cert=ClientCert(client_cert=client_cert, client_key=client_key),
+        ),
+    )
+    connection.dial()