Add SSL support for other sipyco protocols

sipyco/broadcast.py

@@ -1,6 +1,7 @@
 import asyncio
 
 from sipyco import keepalive, pyon
+from sipyco.ssl_tools import create_ssl_context
 from sipyco.asyncio_tools import AsyncioServer
 
 
@@ -15,9 +16,10 @@ def __init__(self, name, notify_cb, disconnect_cb=None):
         self.notify_cbs = notify_cb
         self.disconnect_cb = disconnect_cb
 
-    async def connect(self, host, port):
+    async def connect(self, host, port, local_cert=None, local_key=None, peer_cert=None):
+        ssl_context = create_ssl_context(local_cert, local_key, peer_cert)
         self.reader, self.writer = \
-            await keepalive.async_open_connection(host, port, limit=100 * 1024 * 1024)
+            await keepalive.async_open_connection(host, port, ssl=ssl_context, limit=100 * 1024 * 1024)
         try:
             self.writer.write(_init_string)
             self.writer.write((self.name + "\n").encode())

sipyco/logging_tools.py

@@ -3,6 +3,7 @@
 import re
 
 from sipyco import keepalive
+from sipyco.ssl_tools import create_ssl_context
 from sipyco.asyncio_tools import TaskObject, AsyncioServer
 
 
@@ -178,11 +179,14 @@ def filter(self, record):
 
 
 class LogForwarder(logging.Handler, TaskObject):
-    def __init__(self, host, port, reconnect_timer=5.0, queue_size=1000,
-                 **kwargs):
+    def __init__(self, host, port, local_cert=None, local_key=None, peer_cert=None,
+                 reconnect_timer=5.0, queue_size=1000, **kwargs):
         logging.Handler.__init__(self, **kwargs)
         self.host = host
         self.port = port
+        self.local_cert = local_cert
+        self.local_key = local_key
+        self.peer_cert = peer_cert
         self.setFormatter(MultilineFormatter())
         self._queue = asyncio.Queue(queue_size)
         self.reconnect_timer = reconnect_timer
@@ -192,10 +196,12 @@ def emit(self, record):
 
     async def _do(self):
         reader = writer = None
+        ssl_context = create_ssl_context(self.local_cert, self.local_key, self.peer_cert)
         while True:
             try:
                 reader, writer = await keepalive.async_open_connection(self.host,
-                                                                       self.port)
+                                                                       self.port,
+                                                                       ssl=ssl_context)
                 writer.write(_init_string)
                 while True:
                     message = await self._queue.get() + "\n"

sipyco/sync_struct.py

@@ -18,6 +18,7 @@
 import logging
 
 from sipyco import keepalive, pyon
+from sipyco.ssl_tools import create_ssl_context
 from sipyco.asyncio_tools import AsyncioServer
 
 
@@ -102,6 +103,9 @@ class Subscriber:
         A list of functions may also be used, and they will be called in turn.
     :param disconnect_cb: An optional function called when disconnection
         happens from external causes (i.e. not when ``close`` is called).
+    :param local_cert: Client's certificate file. Providing this enables SSL.
+    :param local_key: Client's private key file. Required when local_cert is provided.
+    :param peer_cert: Server's SSL certificate file to trust. Required when local_cert is provided.
     """
     def __init__(self, notifier_name, target_builder, notify_cb=None,
                  disconnect_cb=None):
@@ -114,9 +118,12 @@ def __init__(self, notifier_name, target_builder, notify_cb=None,
         self.notify_cbs = notify_cb
         self.disconnect_cb = disconnect_cb
 
-    async def connect(self, host, port, before_receive_cb=None):
+    async def connect(self, host, port, local_cert=None, local_key=None, peer_cert=None,
+                      before_receive_cb=None):
+        ssl_context = create_ssl_context(local_cert, local_key, peer_cert)
         self.reader, self.writer = \
-            await keepalive.async_open_connection(host, port, limit=100 * 1024 * 1024)
+            await keepalive.async_open_connection(host, port, ssl=ssl_context,
+                                                  limit=100 * 1024 * 1024)
         try:
             if before_receive_cb is not None:
                 before_receive_cb()