You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gm...@apache.org on 2021/06/25 13:43:06 UTC
[qpid-dispatch] branch main updated: DISPATCH-2179: Set
SSLDomain.VERIFY_PEER or SSLDomain.VERIFY_PEER_NAME based on the
ssl_disable_peer_name_verify flag. Removed --ssl-disable-peer-name-verify
from run_qdstat in system_tests_qdstat. This closes #1270.
This is an automated email from the ASF dual-hosted git repository.
gmurthy pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-dispatch.git
The following commit(s) were added to refs/heads/main by this push:
new e3f2733 DISPATCH-2179: Set SSLDomain.VERIFY_PEER or SSLDomain.VERIFY_PEER_NAME based on the ssl_disable_peer_name_verify flag. Removed --ssl-disable-peer-name-verify from run_qdstat in system_tests_qdstat. This closes #1270.
e3f2733 is described below
commit e3f27339a7fba6c8f43c4715392d20dafad33bfa
Author: Ganesh Murthy <gm...@apache.org>
AuthorDate: Sun Jun 20 22:40:24 2021 -0400
DISPATCH-2179: Set SSLDomain.VERIFY_PEER or SSLDomain.VERIFY_PEER_NAME based on the ssl_disable_peer_name_verify flag. Removed --ssl-disable-peer-name-verify from run_qdstat in system_tests_qdstat. This closes #1270.
---
python/qpid_dispatch_internal/tools/command.py | 19 +++---
tests/authservice.py | 15 ++++-
tests/system_tests_authz_service_plugin.py | 21 ++++---
tests/system_tests_qdstat.py | 82 +++++++++++++++++++++-----
tests/system_tests_ssl.py | 4 ++
tests/system_tests_user_id.py | 10 ++--
6 files changed, 112 insertions(+), 39 deletions(-)
diff --git a/python/qpid_dispatch_internal/tools/command.py b/python/qpid_dispatch_internal/tools/command.py
index 39137dd..470713a 100644
--- a/python/qpid_dispatch_internal/tools/command.py
+++ b/python/qpid_dispatch_internal/tools/command.py
@@ -261,9 +261,8 @@ def opts_url(opts):
url = Url(opts.bus)
# If the options indicate SSL, make sure we use the amqps scheme.
- if opts.ssl_certificate or opts.ssl_trustfile:
+ if opts.ssl_certificate or opts.ssl_trustfile or opts.bus.startswith("amqps:"):
url.scheme = "amqps"
-
return url
@@ -282,6 +281,10 @@ def opts_ssl_domain(opts, mode=SSLDomain.MODE_CLIENT):
@param opts: Parsed optoins including connection_options()
"""
+ url = opts_url(opts)
+ if not url.scheme == "amqps":
+ return None
+
certificate, key, trustfile, password, password_file, ssl_disable_peer_name_verify = opts.ssl_certificate,\
opts.ssl_key,\
opts.ssl_trustfile,\
@@ -289,9 +292,6 @@ def opts_ssl_domain(opts, mode=SSLDomain.MODE_CLIENT):
opts.ssl_password_file, \
opts.ssl_disable_peer_name_verify
- if not (certificate or trustfile):
- return None
-
if password_file:
password = get_password(password_file)
@@ -299,10 +299,11 @@ def opts_ssl_domain(opts, mode=SSLDomain.MODE_CLIENT):
if trustfile:
domain.set_trusted_ca_db(str(trustfile))
- if ssl_disable_peer_name_verify:
- domain.set_peer_authentication(SSLDomain.VERIFY_PEER, str(trustfile))
- else:
- domain.set_peer_authentication(SSLDomain.VERIFY_PEER_NAME, str(trustfile))
+
+ if ssl_disable_peer_name_verify:
+ domain.set_peer_authentication(SSLDomain.VERIFY_PEER)
+ else:
+ domain.set_peer_authentication(SSLDomain.VERIFY_PEER_NAME)
if certificate:
domain.set_credentials(str(certificate), str(key), str(password))
diff --git a/tests/authservice.py b/tests/authservice.py
index 66d5d0a..1d417a4 100755
--- a/tests/authservice.py
+++ b/tests/authservice.py
@@ -21,11 +21,15 @@
import optparse
import signal
import sys
+import os
from cproton import pn_sasl_config_path
from proton import Array, Data, symbol, UNDESCRIBED
from proton.handlers import MessagingHandler
from proton.reactor import Container
+from proton import SSLDomain
+
+import system_test
class AuthService(MessagingHandler):
@@ -38,6 +42,15 @@ class AuthService(MessagingHandler):
self.listener = None
self.tmo = 0.1 # seconds
self.stop_req = False
+ self.acceptor = None
+ self.ssl_domain = SSLDomain(SSLDomain.MODE_SERVER)
+ self.ssl_domain.set_credentials(self.ssl_file('server-certificate.pem'),
+ self.ssl_file('server-private-key.pem'),
+ password="server-password")
+ self.ssl_domain.set_trusted_ca_db(self.ssl_file('ca-certificate.pem'))
+
+ def ssl_file(self, name):
+ return os.path.join(system_test.DIR, 'ssl_certs', name)
def allow(self, user, address, permissions):
if not self.permissions.get(user):
@@ -45,7 +58,7 @@ class AuthService(MessagingHandler):
self.permissions[user][address] = Array(UNDESCRIBED, Data.STRING, *permissions)
def on_start(self, event):
- self.listener = event.container.listen(self.address)
+ self.listener = event.container.listen(self.address, ssl_domain=self.ssl_domain)
event.container.schedule(self.tmo, self)
def stop(self):
diff --git a/tests/system_tests_authz_service_plugin.py b/tests/system_tests_authz_service_plugin.py
index 4fd893f..a741afd 100644
--- a/tests/system_tests_authz_service_plugin.py
+++ b/tests/system_tests_authz_service_plugin.py
@@ -25,6 +25,7 @@ from proton import Message, SASL
from proton.handlers import MessagingHandler
from proton.reactor import Container
+import system_test
from system_test import TestCase, Qdrouterd, main_module, DIR, Process, SkipIfNeeded
from system_test import unittest
@@ -55,6 +56,10 @@ sasldb_path: users.sasldb
mech_list: SCRAM-SHA-1 PLAIN
""")
+ @staticmethod
+ def ssl_file(name):
+ return os.path.join(system_test.DIR, 'ssl_certs', name)
+
@classmethod
def setUpClass(cls):
"""
@@ -69,18 +74,18 @@ mech_list: SCRAM-SHA-1 PLAIN
cls.auth_service_port = cls.tester.get_port()
cls.tester.popen([sys.executable, os.path.join(os.path.dirname(os.path.abspath(__file__)), 'authservice.py'),
- '-a', 'amqps://127.0.0.1:%d' % cls.auth_service_port, '-c', os.getcwd()], expect=Process.RUNNING)
+ '-a', 'amqps://localhost:%d' % cls.auth_service_port, '-c', os.getcwd()], expect=Process.RUNNING)
policy_config_path = os.path.join(DIR, 'policy-authz')
cls.router_port = cls.tester.get_port()
cls.tester.qdrouterd('router', Qdrouterd.Config([
- ('sslProfile', {'name': 'myssl'}),
+ ('sslProfile', {'name': 'myssl', 'caCertFile': cls.ssl_file('ca-certificate.pem')}),
('policy', {'maxConnections': 2, 'policyDir': policy_config_path, 'enableVhostPolicy': 'true'}),
# authService attribute has been deprecated. We are using it here to make sure that we are
# still backward compatible.
- ('authServicePlugin', {'name': 'myauth', 'sslProfile': 'myssl', 'port': cls.auth_service_port, 'host': '127.0.0.1'}),
- ('listener', {'host': '0.0.0.0', 'port': cls.router_port, 'role': 'normal', 'saslPlugin': 'myauth', 'saslMechanisms': 'SCRAM-SHA-1 PLAIN'}),
+ ('authServicePlugin', {'name': 'myauth', 'sslProfile': 'myssl', 'port': cls.auth_service_port, 'host': 'localhost'}),
+ ('listener', {'host': 'localhost', 'port': cls.router_port, 'role': 'normal', 'saslPlugin': 'myauth', 'saslMechanisms': 'SCRAM-SHA-1 PLAIN'}),
('router', {'mode': 'standalone', 'id': 'router',
'saslConfigName': 'tests-mech-SCRAM',
'saslConfigPath': os.getcwd()})
@@ -154,15 +159,15 @@ class AuthServicePluginAuthzDeprecatedTest(AuthServicePluginAuthzTest):
cls.auth_service_port = cls.tester.get_port()
cls.tester.popen([sys.executable, os.path.join(os.path.dirname(os.path.abspath(__file__)), 'authservice.py'),
- '-a', 'amqps://127.0.0.1:%d' % cls.auth_service_port, '-c', os.getcwd()], expect=Process.RUNNING)
+ '-a', 'amqps://localhost:%d' % cls.auth_service_port, '-c', os.getcwd()], expect=Process.RUNNING)
cls.router_port = cls.tester.get_port()
cls.tester.qdrouterd('router', Qdrouterd.Config([
- ('sslProfile', {'name': 'myssl'}),
+ ('sslProfile', {'name': 'myssl', 'caCertFile': cls.ssl_file('ca-certificate.pem')}),
# authService and authSslProfile attributea have been deprecated.
# We are using it here to make sure that we are backward compatible.
- ('authServicePlugin', {'name': 'myauth', 'authSslProfile': 'myssl', 'authService': '127.0.0.1:%d' % cls.auth_service_port}),
- ('listener', {'host': '0.0.0.0', 'port': cls.router_port, 'role': 'normal', 'saslPlugin': 'myauth', 'saslMechanisms': 'SCRAM-SHA-1 PLAIN'}),
+ ('authServicePlugin', {'name': 'myauth', 'authSslProfile': 'myssl', 'authService': 'localhost:%d' % cls.auth_service_port}),
+ ('listener', {'host': 'localhost', 'port': cls.router_port, 'role': 'normal', 'saslPlugin': 'myauth', 'saslMechanisms': 'SCRAM-SHA-1 PLAIN'}),
('router', {'mode': 'standalone', 'id': 'router',
'saslConfigName': 'tests-mech-SCRAM',
'saslConfigPath': os.getcwd()})
diff --git a/tests/system_tests_qdstat.py b/tests/system_tests_qdstat.py
index d46e163..896fd35 100644
--- a/tests/system_tests_qdstat.py
+++ b/tests/system_tests_qdstat.py
@@ -648,7 +648,6 @@ class QdstatLinkPriorityTest(system_test.TestCase):
def _test_links_all_routers(self, command):
out = self.run_qdstat(command)
- print("_test_links_all_routers out=", out)
self.assertTrue(out.count('UTC') == 1)
self.assertTrue(out.count('Router Links') == 2)
self.assertTrue(out.count('inter-router') == 40)
@@ -728,17 +727,20 @@ try:
'certFile': cls.ssl_file('server-certificate.pem'),
'privateKeyFile': cls.ssl_file('server-private-key.pem'),
'password': 'server-password'}),
- ('listener', {'port': cls.tester.get_port()}),
- ('listener', {'port': cls.tester.get_port(), 'sslProfile': 'server-ssl', 'authenticatePeer': 'no', 'requireSsl': 'yes'}),
- ('listener', {'port': cls.tester.get_port(), 'sslProfile': 'server-ssl', 'authenticatePeer': 'no', 'requireSsl': 'no'}),
- ('listener', {'port': cls.tester.get_port(), 'sslProfile': 'server-ssl', 'authenticatePeer': 'yes', 'requireSsl': 'yes',
+ ('listener', {'host': 'localhost', 'port': cls.tester.get_port()}),
+ ('listener', {'host': 'localhost', 'port': cls.tester.get_port(), 'sslProfile': 'server-ssl',
+ 'authenticatePeer': 'no', 'requireSsl': 'yes'}),
+ ('listener', {'host': 'localhost', 'port': cls.tester.get_port(), 'sslProfile': 'server-ssl',
+ 'authenticatePeer': 'no', 'requireSsl': 'no'}),
+ ('listener', {'host': 'localhost', 'port': cls.tester.get_port(), 'sslProfile': 'server-ssl',
+ 'authenticatePeer': 'yes', 'requireSsl': 'yes',
'saslMechanisms': 'EXTERNAL'})
])
cls.router = cls.tester.qdrouterd('test-router', config)
def run_qdstat(self, args, regexp=None, address=None):
p = self.popen(
- ['qdstat', '--bus', str(address or self.router.addresses[0]), '--ssl-disable-peer-name-verify',
+ ['qdstat', '--bus', str(address or self.router.addresses[0]),
'--timeout', str(system_test.TIMEOUT)] + args,
name='qdstat-' + self.id(), stdout=PIPE, expect=None,
universal_newlines=True)
@@ -778,75 +780,123 @@ try:
def ssl_test_bad(self, url_name, arg_names):
self.assertRaises(AssertionError, self.ssl_test, url_name, arg_names)
- # Non-SSL enabled listener should fail SSL connections.
+ # qdstat -b amqp://localhost:<port> --general and makes sure
+ # the router sends back a valid response.
def test_ssl_none(self):
self.ssl_test('none', [])
+ # qdstat -b amqps://localhost:<port> --general
+ # Make sure that the command fails.
def test_ssl_scheme_to_none(self):
self.ssl_test_bad('none_s', [])
+ # qdstat -b amqp://localhost:<port> --general --ssl-certificate /path/to/client-certificate.pem
+ # Makes sure the command fails.
def test_ssl_cert_to_none(self):
self.ssl_test_bad('none', ['client_cert'])
- # Strict SSL listener, SSL only
+ # Tries to run the following command on a listener that requires SSL (requireSsl:yes)
+ # qdstat -b amqp://localhost:<port> --general
+ # Makes sure the command fails.
def test_ssl_none_to_strict(self):
self.ssl_test_bad('strict', [])
+ # qdstat -b amqps://localhost:<port> --general
def test_ssl_schema_to_strict(self):
- self.ssl_test('strict_s', [])
+ self.ssl_test_bad('strict_s', [])
+ # qdstat -b amqps://localhost:<port> --general --ssl-certificate /path/to/client-certificate.pem
+ # --ssl-key /path/to/client-private-key.pem --ssl-password client-password'
def test_ssl_cert_to_strict(self):
- self.ssl_test('strict_s', ['client_cert_all'])
+ self.ssl_test_bad('strict_s', ['client_cert_all'])
+ # qdstat -b amqps://localhost:<port> --general --ssl-trustfile /path/to/ca-certificate.pem
def test_ssl_trustfile_to_strict(self):
self.ssl_test('strict_s', ['trustfile'])
+ # qdstat -b amqps://localhost:<port> --general --ssl-trustfile
+ # /path/to/ca-certificate.pem --ssl-certificate /path/to/client-certificate.pem
+ # --ssl-key /path/to/client-private-key.pem --ssl-password client-password
def test_ssl_trustfile_cert_to_strict(self):
self.ssl_test('strict_s', ['trustfile', 'client_cert_all'])
+ # qdstat -b amqps://localhost:<port> --general --ssl-trustfile /path/to/bad-ca-certificate.pem
+ # Send in a bad ca cert and make sure the test fails.
def test_ssl_bad_trustfile_to_strict(self):
self.ssl_test_bad('strict_s', ['bad_trustfile'])
# Require-auth SSL listener
-
+ # qdstat -b amqp://localhost:<port> --general
+ # Send in no certs to a 'authenticatePeer': 'yes', 'requireSsl': 'yes' listener and make sure it fails.
+ # Also protocol is amqp not amqps
def test_ssl_none_to_auth(self):
self.ssl_test_bad('auth', [])
+ # qdstat -b amqps://localhost:28491 --general
+ # Send in no certs to a 'authenticatePeer': 'yes', 'requireSsl': 'yes' listener and make sure it fails.
def test_ssl_schema_to_auth(self):
self.ssl_test_bad('auth_s', [])
+ # qdstat -b amqps://localhost:<port> --general --ssl-trustfile /path/to/ca-certificate.pem'
+ # Send in just a trustfile to an 'authenticatePeer': 'yes', 'requireSsl': 'yes' listener and make sure it fails.
def test_ssl_trustfile_to_auth(self):
self.ssl_test_bad('auth_s', ['trustfile'])
+ # qdstat -b amqps://localhost:<port> --general --ssl-certificate /path/to/client-certificate.pem
+ # --ssl-key /path/to/client-private-key.pem --ssl-password client-password
+ # Without a trustfile, this test fails
def test_ssl_cert_to_auth(self):
- self.ssl_test('auth_s', ['client_cert_all'])
+ self.ssl_test_bad('auth_s', ['client_cert_all'])
+ # qdstat -b amqps://localhost:<port> --general --ssl-trustfile /path/to/ca-certificate.pem
+ # --ssl-certificate /path/to/client-certificate.pem
+ # --ssl-key /path/to/client-private-key.pem --ssl-password client-password
+ # This has everything, the test should pass.
def test_ssl_trustfile_cert_to_auth(self):
self.ssl_test('auth_s', ['trustfile', 'client_cert_all'])
+ # qdstat -b amqps://localhost:<port> --general --ssl-trustfile /path/to/bad-ca-certificate.pem
+ # --ssl-certificate /path/to/client-certificate.pem --ssl-key /path/to/client-private-key.pem
+ # --ssl-password client-password
+ # Bad trustfile should be rejected.
def test_ssl_bad_trustfile_to_auth(self):
self.ssl_test_bad('auth_s', ['bad_trustfile', 'client_cert_all'])
+ # qdstat -b amqps://localhost:<port> --general --sasl-mechanisms EXTERNAL
+ # --ssl-certificate /path/to/client-certificate.pem --ssl-key /path/to/client-private-key.pem
+ # --ssl-password client-password --ssl-trustfile /path/to/ca-certificate.pem'
def test_ssl_cert_explicit_external_to_auth(self):
- self.ssl_test('auth_s', ['sasl_external', 'client_cert_all'])
+ self.ssl_test('auth_s', ['sasl_external', 'client_cert_all', 'trustfile'])
# Unsecured SSL listener, allows non-SSL
-
+ # qdstat -b amqp://localhost:<port> --general
def test_ssl_none_to_unsecured(self):
self.ssl_test('unsecured', [])
+ # qdstat -b amqps://localhost:<port> --general
def test_ssl_schema_to_unsecured(self):
- self.ssl_test('unsecured_s', [])
+ self.ssl_test_bad('unsecured_s', [])
+ # qdstat -b amqps://localhost:<port> --general --ssl-certificate /path/to/client-certificate.pem --ssl-key
+ # /path/to/client-private-key.pem --ssl-password client-password
+ # A trustfile is required, test will fail
def test_ssl_cert_to_unsecured(self):
- self.ssl_test('unsecured_s', ['client_cert_all'])
+ self.ssl_test_bad('unsecured_s', ['client_cert_all'])
+ # qdstat -b amqps://localhost:<port> --general --ssl-trustfile /path/to/ca-certificate.pem'
+ # Just send in the trustfile, should be all good.
def test_ssl_trustfile_to_unsecured(self):
self.ssl_test('unsecured_s', ['trustfile'])
+ # qdstat -b amqps://localhost:<port> --general --ssl-trustfile /path/to/ca-certificate.pem
+ # --ssl-certificate /path/to/client-certificate.pem --ssl-key /path/to/client-private-key.pem
+ # --ssl-password client-password
+ # We have everything, this should work.
def test_ssl_trustfile_cert_to_unsecured(self):
self.ssl_test('unsecured_s', ['trustfile', 'client_cert_all'])
+ # qdstat -b amqps://localhost:<port> --general --ssl-trustfile /path/to/bad-ca-certificate.pem']
+ # Bad trustfile, test will fail.
def test_ssl_bad_trustfile_to_unsecured(self):
self.ssl_test_bad('unsecured_s', ['bad_trustfile'])
diff --git a/tests/system_tests_ssl.py b/tests/system_tests_ssl.py
index b101882..ed563d4 100644
--- a/tests/system_tests_ssl.py
+++ b/tests/system_tests_ssl.py
@@ -382,6 +382,8 @@ class RouterTestSslClient(RouterTestSslBase):
url = Url("amqps://0.0.0.0:%d/$management" % listener_port)
# Preparing SSLDomain (client cert) and SASL authentication info
domain = SSLDomain(SSLDomain.MODE_CLIENT)
+ domain.set_trusted_ca_db(self.ssl_file('ca-certificate.pem'))
+ domain.set_peer_authentication(SSLDomain.VERIFY_PEER)
# Enforcing given TLS protocol
cproton.pn_ssl_domain_set_protocols(domain._domain, tls_protocol)
@@ -415,6 +417,8 @@ class RouterTestSslClient(RouterTestSslBase):
domain.set_credentials(self.ssl_file('client-certificate.pem'),
self.ssl_file('client-private-key.pem'),
'client-password')
+ domain.set_trusted_ca_db(self.ssl_file('ca-certificate.pem'))
+ domain.set_peer_authentication(SSLDomain.VERIFY_PEER)
# Enforcing given TLS protocol
cproton.pn_ssl_domain_set_protocols(domain._domain, tls_protocol)
diff --git a/tests/system_tests_user_id.py b/tests/system_tests_user_id.py
index 7fe6bba..76289b2 100644
--- a/tests/system_tests_user_id.py
+++ b/tests/system_tests_user_id.py
@@ -208,7 +208,7 @@ class QdSSLUseridTest(TestCase):
('listener', {'port': cls.tester.get_port(), 'sslProfile': 'server-ssl11', 'authenticatePeer': 'yes',
'requireSsl': 'yes', 'saslMechanisms': 'EXTERNAL'}),
- # peer is not being authenticated here. the user must "anonymous" which is what pn_transport_get_user
+ # peer is not being authenticated here. the user must be "anonymous" which is what pn_transport_get_user
# returns
('listener', {'port': cls.tester.get_port(), 'sslProfile': 'server-ssl12', 'authenticatePeer': 'no',
'requireSsl': 'yes', 'saslMechanisms': 'ANONYMOUS'}),
@@ -242,7 +242,7 @@ class QdSSLUseridTest(TestCase):
domain = SSLDomain(mode)
if trustfile:
domain.set_trusted_ca_db(str(trustfile))
- domain.set_peer_authentication(SSLDomain.VERIFY_PEER, str(trustfile))
+ domain.set_peer_authentication(SSLDomain.VERIFY_PEER, None)
if certificate:
domain.set_credentials(str(certificate), str(key), str(password))
@@ -314,11 +314,11 @@ class QdSSLUseridTest(TestCase):
user = node.query(type='org.apache.qpid.dispatch.connection', attribute_names=[u'user']).results[10][0]
self.assertEqual("C=US,ST=NC,L=Raleigh,OU=Dev,O=Client,CN=127.0.0.1", str(user))
- # authenticatePeer is set to 'no' in this listener, there should be no user on the connection.
+ # authenticatePeer is set to 'no' in this listener, the user should anonymous on the connection.
addr = self.address(11).replace("amqp", "amqps")
- node = Node.connect(addr)
+ node = Node.connect(addr, ssl_domain=domain)
user = node.query(type='org.apache.qpid.dispatch.connection', attribute_names=[u'user']).results[11][0]
- self.assertEqual(None, user)
+ self.assertEqual("anonymous", user)
addr = self.address(12).replace("amqp", "amqps")
node = Node.connect(addr, ssl_domain=domain)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org