Merge 6e64e9b into 3f13a38

python/tests/integration/certificates/client.json

@@ -0,0 +1,17 @@
+{
+  "CN": "client",
+  "key": {
+    "algo": "rsa",
+    "size": 2048
+  },
+  "names": [
+    {
+      "C": "GB",
+      "L": "London",
+      "O": "Custom Widgets",
+      "OU": "Custom Widgets Hosts",
+      "ST": "England"
+    }
+  ],
+  "hosts": []
+}

python/tests/integration/certificates/client_ca1-key.pem

@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEogIBAAKCAQEApemlaHEsetE0ANYgY7ZJQzjK4lt732SGvMZaGzNNCrEEkMj9
+7D/rZgDZ8L9ZJ+R7I9oDSLfUDEw5k3g9NpG0mlY3McJLwLa2TiwpwWGoiiGbArsu
+3qztbNFxQIrj/4UCHEIqLd521LuA66S8r9P1uhvatRZbjw/Nwh9J3sE4A1LojZmj
+4DR6Aw2L0dk4AD0ixhb512r8bS7T+vWIxmIB+Y9w/cvPbLnu7ufAow0ntUI3Je4x
+t1McaoUsBQ/RK00PT/f3J17BKyfB0PvLYMMpRscLTEwHyqsEK6B4+WM3GeUH6/o3
+j0AKpvtmKGSYfXvyLMgDSl9gZvD0tHj80EqXUwIDAQABAoIBAAZCt8wmISCNTmIN
+snEwyrjvprA99YGrgG4VKgdGu0yA+4QfIX3Nt6tEsvSjs9COjZr/ugn/bc/8/Fs1
+OVIa027TfAezpjoiauSuQ/EZJ0v3Eqtatt0ON3NYv+ZIl2vn7/lzAbZzY5aJcMbz
+k28rF2WrcWhN7KyMUx5VIet27Q8q/SexGHhMWbIQdtJ/8X1f0FCJfwZq18789ZRg
+ij0M+O22EV3/vu9bQ07mewcSgaSJsnoylPkYki6+JDo55sdJdjkUkLnmhGcFGKMa
+NSpChO1XmMsY5njRxkdPoZ2dtnocvIt5X0gpKywyUc+l5lJqQBpCwYXXjrl3tlda
+tKNCHbkCgYEA26Lxcx6Udfx/KwFMYCXnniPRzPUVrzzjcBIoa2tHaC+0V3DehWb+
+AOSt4A9uIegD0fU9PCP8zmiktNQVl31i2I4QUwczd0kYGbCcdqyqAZNyW8idfpBG
+bnUDJjPVRSC+65bVd6tupYoeU7BTW430rIIaPIm1a5PuBp/y0qLsZNUCgYEAwWGt
+PwBFpSkC2QC11l7owL3TWdHjM4iej1okHA1xjtktSPRu5ofwcrutg6ntAE+Jprpd
+MhA6HnXIjldeiKTiSzKoaWH7OZM7evjYNlMH86Uyt7lV+B/4ZfuRHaQwC3TvLZkp
+2OLW808E0n+M8fS/osetsfyyLdwt6OW6pjINP4cCgYBWJkGir/n3lYSj76xvgi6p
+fs9KH0/UHoPvb4/fIoMtwJhyO9lsZgt3ejshSawfLIxjDFhqgIsmwzDnpNCbTRk6
+a+7HQmnTfh4v5XBZtDwyxgzzJ2tvO1QE+2yyzV338XIxokY03E9YKybeGKl5neK8
+z4NO/4zjl3CjtJVgPXuPyQKBgE9MOHiPKf/x80L88ZO4U4VF0fcRBDPLoAl0kz4V
+nS1QjStPYHKT59uEbkCBW7g25WFDJpgy40I+VkFYPmGWC11+pmSgUx5m64sfo7mT
+Dr2wTj3ceA5JPdjD8dvPygvIpZNzLR/M1QvsqTOQLkHBdRvQ+b70ujPoB8NrAMDJ
+4XjdAoGATLef9CPLUfFVbcWqh46zAZz3Q6o5TBuOCCFYBVRrN192vRtjFYmgceCn
+dMTsayPOKYuIiJe+WN3hQr8z/u0pNkKu8v/3qWZCaXWF0U90+fqLP5GOhUH0Y/wD
+INO3uhBrFCvc/zKCBcw00+V2eDT0HvzDE17gEVFwIYIAuy8C+fI=
+-----END RSA PRIVATE KEY-----

python/tests/integration/certificates/client_ca1.pem

@@ -0,0 +1,24 @@
+-----BEGIN CERTIFICATE-----
+MIIEEjCCAvqgAwIBAgIUVKiYrie266hGq81cOj07J8V4Iv8wDQYJKoZIhvcNAQEL
+BQAwgYsxCzAJBgNVBAYTAkdCMRAwDgYDVQQIEwdFbmdsYW5kMQ8wDQYDVQQHEwZM
+b25kb24xFzAVBgNVBAoTDkN1c3RvbSBXaWRnZXRzMR8wHQYDVQQLExZDdXN0b20g
+V2lkZ2V0cyBSb290IENBMR8wHQYDVQQDExZDdXN0b20gV2lkZ2V0cyBSb290IENB
+MB4XDTIwMDYyODA4NDQwMFoXDTIxMDYyODA4NDQwMFoweTELMAkGA1UEBhMCR0Ix
+EDAOBgNVBAgTB0VuZ2xhbmQxDzANBgNVBAcTBkxvbmRvbjEXMBUGA1UEChMOQ3Vz
+dG9tIFdpZGdldHMxHTAbBgNVBAsTFEN1c3RvbSBXaWRnZXRzIEhvc3RzMQ8wDQYD
+VQQDEwZjbGllbnQwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQCl6aVo
+cSx60TQA1iBjtklDOMriW3vfZIa8xlobM00KsQSQyP3sP+tmANnwv1kn5Hsj2gNI
+t9QMTDmTeD02kbSaVjcxwkvAtrZOLCnBYaiKIZsCuy7erO1s0XFAiuP/hQIcQiot
+3nbUu4DrpLyv0/W6G9q1FluPD83CH0newTgDUuiNmaPgNHoDDYvR2TgAPSLGFvnX
+avxtLtP69YjGYgH5j3D9y89sue7u58CjDSe1Qjcl7jG3UxxqhSwFD9ErTQ9P9/cn
+XsErJ8HQ+8tgwylGxwtMTAfKqwQroHj5YzcZ5Qfr+jePQAqm+2YoZJh9e/IsyANK
+X2Bm8PS0ePzQSpdTAgMBAAGjfzB9MA4GA1UdDwEB/wQEAwIFoDAdBgNVHSUEFjAU
+BggrBgEFBQcDAQYIKwYBBQUHAwIwDAYDVR0TAQH/BAIwADAdBgNVHQ4EFgQU6tWe
+O4UB6qYzOf2jBAQ8lPXU8REwHwYDVR0jBBgwFoAUZf+ASAG3seauXqVAe0g4IVTC
+ymcwDQYJKoZIhvcNAQELBQADggEBAIJ1FOdp7CJTrsHeS5BZ+HwOcF4nx2GeCkwP
+HCvKs2TLyqTiMeQufloweiWT+Eh8Y3cpyof3HSZ1mBiOZvmeraeFLmrTzWNHW5ef
+Tie4bbdSWuAwHOFB5dCjr3rkkG9O3T4hF/FLhRLPetM6dfQiMkn+TLFzHFT35Wb7
+ry9CVjNIKVFrnlGhGODuhkMlhoONxUgnoWE2A0IfYpf/Fll2QyqihpAXXQ/vTtax
+HWpSiSkSQBvxGk4AocqUy9AKUV00tnvRKIWbYuwwZSIjDCVEOp6PXVg7AGQzHaCb
+AVJeufJZpy/P8n6r4iHFwkUppQpqKlu3nw6kVo6wt7aV2X029ds=
+-----END CERTIFICATE-----

python/tests/integration/certificates/mkcerts.sh

@@ -25,6 +25,7 @@ set -Eeuxo pipefail
 
 cfssl gencert -initca ca.json | cfssljson -bare ca1
 cfssl gencert -ca ca1.pem -ca-key ca1-key.pem -config=ca-config.json localhost.json | cfssljson -bare localhost_ca1
+cfssl gencert -ca ca1.pem -ca-key ca1-key.pem -profile client client.json | cfssljson -bare client_ca1
 
 cfssl gencert -initca ca.json | cfssljson -bare ca2
 cfssl gencert -ca ca2.pem -ca-key ca2-key.pem -config=ca-config.json localhost.json | cfssljson -bare localhost_ca2

python/tests/integration/test_PROTON_1870_ssl_error_logging.py

@@ -0,0 +1,274 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License
+#
+
+"""
+PROTON-2111 python: memory leak on Container, SSL, and SSLDomain objects
+"""
+
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import contextlib
+import logging
+
+import os
+import sys
+
+try:
+    import queue
+except ImportError:
+    import Queue as queue  # for Python 2.6, 2.7 compatibility
+import socket
+import ssl
+import threading
+
+import cproton
+
+import proton.handlers
+import proton.utils
+import proton.reactor
+
+from test_unittest import unittest
+
+
+class Broker(proton.handlers.MessagingHandler):
+    """Mock broker with TLS support and error capture."""
+    def __init__(self, acceptor_url, ssl_domain=None):
+        # type: (str, proton.SSLDomain) -> None
+        super(Broker, self).__init__()
+        self.acceptor_url = acceptor_url
+        self.ssl_domain = ssl_domain
+
+        self.acceptor = None
+        self._acceptor_opened_event = threading.Event()
+
+        self.on_message_ = threading.Event()
+        self.errors = []
+
+    def get_acceptor_sockname(self):
+        # type: () -> (str, int)
+        self._acceptor_opened_event.wait()
+        if hasattr(self.acceptor, '_selectable'):  # proton 0.30.0+
+            sockname = self.acceptor._selectable._delegate.getsockname()
+        else:  # works in proton 0.27.0
+            selectable = cproton.pn_cast_pn_selectable(self.acceptor._impl)
+            fd = cproton.pn_selectable_get_fd(selectable)
+            s = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM)
+            sockname = s.getsockname()
+        return sockname[:2]
+
+    def on_start(self, event):
+        self.acceptor = event.container.listen(self.acceptor_url, ssl_domain=self.ssl_domain)
+        self._acceptor_opened_event.set()
+
+    def on_link_opening(self, event):
+        link = event.link  # type: proton.Link
+        if link.is_sender:
+            assert not link.remote_source.dynamic, "This cannot happen"
+            link.source.address = link.remote_source.address
+        elif link.remote_target.address:
+            link.target.address = link.remote_target.address
+
+    def on_message(self, event):
+        self.on_message_.set()
+
+    def on_transport_error(self, event):
+        super(Broker, self).on_transport_error(event)
+        self.errors.append(event.transport.condition)
+
+
+@contextlib.contextmanager
+def test_broker(ssl_domain=None):
+    # type: (proton.SSLDomain) -> Broker
+    broker = Broker('localhost:0', ssl_domain=ssl_domain)
+    container = proton.reactor.Container(broker)
+    t = threading.Thread(target=container.run)
+    t.start()
+
+    yield broker
+
+    container.stop()
+    if broker.acceptor:
+        broker.acceptor.close()
+    t.join()
+
+
+class SampleSender(proton.handlers.MessagingHandler):
+    """Client with TLS support which sends one message and then ends."""
+    def __init__(self, msg_id, urls, ssl_domain=None, *args, **kwargs):
+        # type: (str, str, proton.SSLDomain, *object, **object) -> None
+        super(SampleSender, self).__init__(*args, **kwargs)
+        self.urls = urls
+        self.msg_id = msg_id
+        self.ssl_domain = ssl_domain
+
+        self.errors = []
+
+    def on_start(self, event):
+        # type: (proton.Event) -> None
+        conn = event.container.connect(url=self.urls, reconnect=False, ssl_domain=self.ssl_domain)
+        event.container.create_sender(conn, target='someTarget')
+
+    def on_sendable(self, event):
+        msg = proton.Message(body={'msg-id': self.msg_id, 'name': 'python'})
+        event.sender.send(msg)
+        event.sender.close()
+        event.connection.close()
+
+    def on_transport_error(self, event):
+        super(SampleSender, self).on_transport_error(event)
+        self.errors.append(event.transport.condition)
+
+
+class Proton1870Test(unittest.TestCase):
+    """Starts a broker with ssl configuration (or without it, in some cases) and connects
+    to it with a client to check if helpful error messages are logged."""
+    cwd = os.path.dirname(__file__)
+
+    def test_broker_cert_success(self):
+        """Basic TLS scenario without any error."""
+        certificate_db = os.path.join(self.cwd, 'certificates', 'ca1.pem')
+
+        cert_file = os.path.join(self.cwd, 'certificates', 'localhost_ca1.pem')
+        key_file = os.path.join(self.cwd, 'certificates', 'localhost_ca1-key.pem')
+
+        broker_ssl_domain = proton.SSLDomain(proton.SSLDomain.MODE_SERVER)
+        broker_ssl_domain.set_credentials(cert_file, key_file, password=None)
+
+        client_ssl_domain = proton.SSLDomain(proton.SSLDomain.MODE_CLIENT)
+        client_ssl_domain.set_trusted_ca_db(certificate_db)
+        client_ssl_domain.set_peer_authentication(proton.SSLDomain.VERIFY_PEER)
+
+        with test_broker(ssl_domain=broker_ssl_domain) as broker:
+            urls = "amqps://localhost:{0}".format(broker.get_acceptor_sockname()[1])
+            container = proton.reactor.Container(SampleSender('msg_id', urls, client_ssl_domain))
+            container.run()
+
+    @unittest.skipIf(sys.platform.startswith("win32"), "TODO: Gets stuck on Windows")
+    def test_broker_cert_shutdown_connection_sslsock(self):
+        """When a remote peer drops TCP connection (with established
+         SSL+AMQP connection.session on it) and it drops the TCP
+         connection by sending FIN+ACK packet, descriptive error is generated."""
+        port_q = queue.Queue()
+
+        def server():
+            """Mock TLS server without any AMQP support which kills incoming connections."""
+            cert_file = os.path.join(self.cwd, 'certificates', 'localhost_ca1.pem')
+            key_file = os.path.join(self.cwd, 'certificates', 'localhost_ca1-key.pem')
+
+            context = ssl.SSLContext(ssl.PROTOCOL_TLS)
+            context.load_cert_chain(cert_file, key_file)
+
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) as sock:
+                sock.bind(('localhost', 0))
+                sock.listen(5)
+                with context.wrap_socket(sock, server_side=True) as ssock:
+                    port_q.put(ssock.getsockname()[1])
+                    conn, _ = ssock.accept()
+                    conn.shutdown(socket.SHUT_RDWR)
+                    conn.close()
+
+        t = threading.Thread(target=server)
+        t.start()
+
+        try:
+            url = "amqps://localhost:{0}".format(port_q.get())
+            bc = proton.utils.BlockingConnection(url)
+            s = bc.create_sender("address")
+            s.send(proton.Message())
+            self.fail("Expected a ConnectionException")
+        except proton.ConnectionException as e:
+            # TODO XXX "SSL Failure: Unknown error" string is misleading, as TLS was closed cleanly
+            error_message = str(e)
+            self.assertIn("amqp:connection:framing-error", error_message)
+            self.assertIn("SSL Failure: Unknown error", error_message)
+
+        t.join()
+
+    def test_broker_cert_file_does_not_exist(self):
+        """When the certificate files we specified do not exist
+        on disk, we get an error."""
+        cert_file = os.path.join(self.cwd, 'certificates', 'no_such_file.pem')
+        key_file = os.path.join(self.cwd, 'certificates', 'localhost_ca1-key.pem')
+
+        broker_ssl_domain = proton.SSLDomain(proton.SSLDomain.MODE_SERVER)
+        try:
+            broker_ssl_domain.set_credentials(cert_file, key_file, password=None)
+        except proton.SSLException as e:
+            # TODO XXX "SSL failure" is too generic, it should talk about missing files instead
+            error_message = str(e)
+            self.assertIn("SSL failure", error_message)
+
+    def test_brokers_ca_not_trusted_by_client(self):
+        cert_file = os.path.join(self.cwd, 'certificates', 'localhost_ca1.pem')
+        key_file = os.path.join(self.cwd, 'certificates', 'localhost_ca1-key.pem')
+
+        broker_ssl_domain = proton.SSLDomain(proton.SSLDomain.MODE_SERVER)
+        broker_ssl_domain.set_credentials(cert_file, key_file, password=None)
+
+        # intentionally not setting trusted_ca_db here
+        client_ssl_domain = proton.SSLDomain(proton.SSLDomain.MODE_CLIENT)
+        client_ssl_domain.set_peer_authentication(proton.SSLDomain.VERIFY_PEER)
+
+        with test_broker(ssl_domain=broker_ssl_domain) as broker:
+            urls = "amqps://localhost:{0}".format(broker.get_acceptor_sockname()[1])
+            sender = SampleSender('msg_id', urls, client_ssl_domain)
+            container = proton.reactor.Container(sender)
+            container.run()
+            # TODO XXX "certificate verify failed" is too generic,
+            #  it should say exactly what is wrong with the certificate, e.g. wrong hostname, expired certificate, ...
+            self.assertEqual(1, len(sender.errors))
+            client_error = str(sender.errors[0])
+            self.assertIn("amqp:connection:framing-error", client_error)
+            self.assertIn("certificate verify failed", client_error)
+        # TODO XXX "Unknown error" is unhelpful in diagnosing the problem
+        self.assertEqual(1, len(broker.errors))
+        broker_error = str(broker.errors[0])
+        self.assertIn("amqp:connection:framing-error", broker_error)
+        self.assertIn("SSL Failure: Unknown error", broker_error)
+
+    def test_broker_certificate_fails_peer_name_check(self):
+        cert_file = os.path.join(self.cwd, 'certificates', 'localhost_ca1.pem')
+        key_file = os.path.join(self.cwd, 'certificates', 'localhost_ca1-key.pem')
+        certificate_db = os.path.join(self.cwd, 'certificates', 'ca1.pem')
+
+        broker_ssl_domain = proton.SSLDomain(proton.SSLDomain.MODE_SERVER)
+        broker_ssl_domain.set_credentials(cert_file, key_file, password=None)
+
+        client_ssl_domain = proton.SSLDomain(proton.SSLDomain.MODE_CLIENT)
+        client_ssl_domain.set_trusted_ca_db(certificate_db)
+        client_ssl_domain.set_peer_authentication(proton.SSLDomain.VERIFY_PEER_NAME)
+
+        with test_broker(ssl_domain=broker_ssl_domain) as broker:
+            urls = "amqps://127.0.0.1:{0}".format(broker.get_acceptor_sockname()[1])
+            sender = SampleSender('msg_id', urls, client_ssl_domain)
+            container = proton.reactor.Container(sender)
+            container.run()
+            # TODO XXX "certificate verify failed" is too generic,
+            #  it should say exactly what is wrong with the certificate, e.g. wrong hostname, expired certificate, ...
+            self.assertEqual(1, len(sender.errors))
+            client_error = str(sender.errors[0])
+            self.assertIn("amqp:connection:framing-error", client_error)
+            self.assertIn("certificate verify failed", client_error)
+        # TODO XXX "Unknown error" is unhelpful in diagnosing the problem
+        self.assertEqual(1, len(broker.errors))
+        broker_error = str(broker.errors[0])
+        self.assertIn("amqp:connection:framing-error", broker_error)
+        self.assertIn("SSL Failure: Unknown error", broker_error)