feat: ssl support

aiostomp/aiostomp.py

@@ -19,6 +19,7 @@
 class AioStomp:
 
     def __init__(self, host, port,
+                 ssl_context=None,
                  reconnect_max_attempts=-1, reconnect_timeout=1000,
                  heartbeat=True, heartbeat_interval_cx=1000, heartbeat_interval_cy=1000,
                  error_handler=None):
@@ -29,7 +30,9 @@ def __init__(self, host, port,
             'cy': heartbeat_interval_cy
         }
 
-        self._protocol = StompProtocol(self, host, port, heartbeat=self._heartbeat)
+        self._protocol = StompProtocol(
+            self, host, port, heartbeat=self._heartbeat,
+            ssl_context=ssl_context)
         self._last_subscribe_id = 0
         self._subscriptions = {}
 
@@ -286,20 +289,24 @@ async def start(self):
 
 class StompProtocol(object):
 
-    def __init__(self, handler, host, port, loop=None, heartbeat={}):
+    def __init__(self, handler, host, port,
+                 loop=None, heartbeat={}, ssl_context=None):
 
         self.host = host
         self.port = port
+        self.ssl_context = ssl_context
 
         if loop is None:
             loop = asyncio.get_event_loop()
 
         self._loop = loop
-        self._factory = functools.partial(StompReader, handler, loop=loop, heartbeat=heartbeat)
+        self._factory = functools.partial(
+            StompReader, handler, loop=loop, heartbeat=heartbeat)
 
     async def connect(self):
         trans, proto = await self._loop.create_connection(
-            self._factory, host=self.host, port=self.port)
+            self._factory, host=self.host, port=self.port,
+            ssl=self.ssl_context)
 
         self._transport = trans
         self._protocol = proto

example/activemq_keystore.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+
+openssl req -newkey rsa:2048 -nodes -keyout key.pem -x509 -days 365 -out certificate.pem
+openssl pkcs12 -inkey key.pem -in certificate.pem -export -out certificate.p12
+
+cat key.pem certificate.pem > temp.pem
+openssl pkcs12 -export -in temp.pem -out server.p12 -name localhost
+
+keytool -importkeystore  -destkeystore keystore.jks -srckeystore server.p12 -srcstoretype PKCS12 -alias localhost
+
+rm temp.pem
+rm server.p12

example/certificate.pem

@@ -0,0 +1,19 @@
+-----BEGIN CERTIFICATE-----
+MIIDBjCCAe4CCQDOJtxfR2JVSzANBgkqhkiG9w0BAQsFADBFMQswCQYDVQQGEwJC
+UjELMAkGA1UECAwCUlMxFTATBgNVBAcMDFBvcnRvIEFsZWdyZTESMBAGA1UEAwwJ
+bG9jYWxob3N0MB4XDTE4MDMyNDE5MzQ0MFoXDTE5MDMyNDE5MzQ0MFowRTELMAkG
+A1UEBhMCQlIxCzAJBgNVBAgMAlJTMRUwEwYDVQQHDAxQb3J0byBBbGVncmUxEjAQ
+BgNVBAMMCWxvY2FsaG9zdDCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEB
+AKyDrrJNIvfvZxOyKmzrQfImoGHOeb1RQHAHLnCwcu5QblSxHS59HSkZ5kDqc4EA
+sMNR1vYUm+ikXKpKBXT/BgikHpDA+ZXP/3A/VDif9L8J77DvZeV4/TI9yU2Wq/rm
+31cByxOWRpaLIhdpE+IdfU7xbwD39o7iQlFp2E9z+ObtQ3MXFPLWCIiMoqgruDJ6
+K1z+a2Eav5TwixeXRhpHLB1SFx0YQB1PeRMyEaHybGLJgetTOFjC15mv2R1J7M3C
+bK2fdAfK2FMvY8FF0fElSMt6N7PACe4DTDDCvNxVZlz2wyhLUND2tTvlC5UbzuMq
+orZ2hR3B8SwhO+/ghm4TvD0CAwEAATANBgkqhkiG9w0BAQsFAAOCAQEAXIS2cNaF
+Nqm7+bSw2r/pH3aR87uFUDZsRno2/S/Xr75LtrRquJcWH7olTrWqTk8Bt1sUb+9I
+6aTRPeTT+dMPypIYWeT5OwHrUiJ+OqJhuBhbOdSOfeMVTa51LgsSUximdBGGpz9B
+6plYvMis8VFd0JEenOOU9eHlR86eSVh4suxorsJWIjN1IX54GU6QT8e3aWdIBRmE
+f5Hsbw2ZrVPhO1nlx7sdOR73GRbg8rQbi14a5tTUaxwgkk7owdyhcSkCeu30rzJE
+3dfVpqkRHXChbrmI2DftwEyxZoMccgd++3GmjMbSmuEkmv49wLqJe9OG44jqooc2
+UCI+cBVJZ4jvmQ==
+-----END CERTIFICATE-----

example/key.pem

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQCsg66yTSL372cT
+sips60HyJqBhznm9UUBwBy5wsHLuUG5UsR0ufR0pGeZA6nOBALDDUdb2FJvopFyq
+SgV0/wYIpB6QwPmVz/9wP1Q4n/S/Ce+w72XleP0yPclNlqv65t9XAcsTlkaWiyIX
+aRPiHX1O8W8A9/aO4kJRadhPc/jm7UNzFxTy1giIjKKoK7gyeitc/mthGr+U8IsX
+l0YaRywdUhcdGEAdT3kTMhGh8mxiyYHrUzhYwteZr9kdSezNwmytn3QHythTL2PB
+RdHxJUjLejezwAnuA0wwwrzcVWZc9sMoS1DQ9rU75QuVG87jKqK2doUdwfEsITvv
+4IZuE7w9AgMBAAECggEAbDA5AsC1SV/t+MAnMx9mwAHRWwUt+7hU/OzEBw/hQ+tV
+Be/9KQZ1ny0K5XMi9KgGMrfEY5cRH5jUNEcvnOvCUPD76jPbUkSbIT3WB5zo1FXa
+5896y6pQv41YDTyMplDbh6ksPiBkLE0dGClhutBboYJv5t+a1Ii4pm4Gpe1JDyh8
+WfnT4D0eRf4wQcvhNcAPT2XtL+Ej+OFGvhZfXh7DwuK6v5bl/lX2pVCKiQW9yn2V
+MRrUUC/5OLQMtoTNNaP0/GZMxzRaRbfHEipNIvgi7jRnpCJGoJfaAcWRIkTyOMwv
+s+tS1Ky5x+hpUPjFhndot0G4G2nnDZcJxf2zpzLEUQKBgQDjUAWjhoI1MQPdH7ZR
+HFUkIIYFb2xvq0t5/+nNORdYNWWB9Ml3sY99D5gareE0bm50PJkvv+wtQ+rY9/4O
+6HnWL4WzVk5jkJIq7ctFcVNfY2BjpZHfXnNzfhuKoFCS2idyPqffhA9LBpesQ0Qa
+q78C2FIiqarvEKSpCX0fLiO4dwKBgQDCST+/ixDwP6m4TaiiP8T5lhTPWPadrylx
+EYRHZUwFBxod5owWpVQSd4KlMrOSMdMR+d6mSgf2i5pDRnLV1Tsq1VT3y6okBfI5
+/Cf14ZPvy5CpH67709zqs4C8Chn50bM74kUMS6+XIC+KqbRI0Deh3oaNkvJkmVGG
+YuQaaCKR6wKBgAqCaqBS2bZH29vfWXiKjM0sy8V4j1TiYIz1bgPU/eVQIq9Nd/sW
+tIGdtPSW0XCxIl4xZwu0HeFB4JvDa3WY/fkPHWlbNiiYSIm/kMRXJvmkCJLS1tmy
+7jeKvw87tj3z5Br+FpjY7DQVy/Fk8sMg6aXiBnofyg1fgZ1nGOfWJESlAoGBAIiJ
+g7Rv3fqJMxQj93P2zs38TCmfwsj0NXKRtd2a8b7Cy/bhc+fiTmAf/v9rz0jZnRUP
+/gntEwWP/Fx1IPTnwooKnIj8Z1ijmSRo8ghZfGbUL5i7fHTb4nBKsBj2HGU1mvll
+vdKzSfYyF1cihmF5eu9V/Ed4UsKiAOopi1aHgmv7AoGBAI8DvTZBnHI2kKckh2ih
+KW45ig0oj6C0vlNZQ11WJsjbLx3tBCFEu6XSpNM96rbvUvrwbZZxoZmH+zEypcpv
++yKRRwVlleTjFtkcbYVeniyXZBYQV46C+0lHgh5PN6axjZhpAMmTQDTsgM/0YIac
+j81cZoXmdlXZndIW1ZV76ntm
+-----END PRIVATE KEY-----

example/subscribe_ssl.py

@@ -0,0 +1,71 @@
+# Self-signed key:
+#
+#  openssl req -x509 -newkey rsa:2048 -keyout selfsigned.key -nodes\
+#   -out selfsigned.cert -sha256 -days 1000
+#
+# Use 'localhost' as Common Name
+
+
+import sys
+import logging
+import asyncio
+import ssl
+try:
+    import tornado
+except ImportError:
+    pass
+
+from aiostomp import AioStomp
+
+
+logging.basicConfig(
+    format="%(asctime)s - %(filename)s:%(lineno)d - "
+    "%(levelname)s - %(message)s",
+    level='DEBUG')
+
+
+async def on_message(frame, message):
+    print(message)
+    return True
+
+
+async def report_error(error):
+    print('report_error', error)
+
+
+async def run():
+    ssl_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH,
+                                             cafile='certificate.pem')
+
+    client = AioStomp('localhost', 61612,
+                      error_handler=report_error,
+                      ssl_context=ssl_context)
+
+    client.subscribe('/queue/test', handler=on_message)
+    await client.connect()
+
+    await asyncio.sleep(10)
+    client.subscribe('/queue/test', handler=on_message)
+
+    client.send('/queue/test', body=u'Pedro Kiefer', headers={})
+
+    await asyncio.sleep(10)
+
+
+def main(args):
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(run())
+    loop.run_forever()
+
+
+def tornado_main(args):
+    from tornado.platform.asyncio import AsyncIOMainLoop
+    AsyncIOMainLoop().install()
+
+    loop = tornado.ioloop.IOLoop.instance()
+    loop.add_callback(lambda: run())
+    loop.start()
+
+
+if __name__ == '__main__':
+    main(sys.argv)

tests/test_main.py

@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 import asyncio
+import ssl
 
 from aiostomp.test_utils import AsyncTestCase, unittest_run_loop
 
@@ -325,6 +326,19 @@ class TestAioStomp(AsyncTestCase):
     async def setUpAsync(self):
         self.stomp = AioStomp('127.0.0.1', 61613)
 
+    @patch('aiostomp.aiostomp.StompProtocol')
+    @unittest_run_loop
+    async def test_aiostomp_supports_ssl(self, stom_protocol_mock):
+        ssl_context = ssl.create_default_context()
+        stomp = AioStomp('127.0.0.1', 61613, ssl_context=ssl_context)
+
+        args, kwargs = stom_protocol_mock.call_args
+
+        self.assertTrue('127.0.0.1' in args)
+        self.assertTrue(61613 in args)
+        self.assertTrue(stomp in args)
+        self.assertTrue(kwargs['ssl_context'] == ssl_context)
+
     @unittest_run_loop
     async def test_can_connect_to_server(self):
         self.stomp._protocol.connect = CoroutineMock()
@@ -519,7 +533,19 @@ async def test_can_create_a_connection(self):
         await self.protocol.connect()
 
         self._loop.create_connection.assert_called_with(
-            self.protocol._factory, host='127.0.0.1', port=61613)
+            self.protocol._factory, host='127.0.0.1', port=61613,
+            ssl=None)
+
+    @unittest_run_loop
+    async def test_can_create_a_connection_with_ssl_context(self):
+        ssl_context = ssl.create_default_context()
+        self.protocol.ssl_context = ssl_context
+
+        await self.protocol.connect()
+
+        self._loop.create_connection.assert_called_with(
+            self.protocol._factory, host='127.0.0.1', port=61613,
+            ssl=ssl_context)
 
     @unittest_run_loop
     async def test_can_subscribe(self):