PYTHON-1075 Support running the entire test suite with TLS
This commit is contained in:
Shane Harvey
parent
dff826b7e3
commit
b057cd47e8
@ -35,8 +35,12 @@ import pymongo.errors
|
||||
|
||||
from bson.py3compat import _unicode
|
||||
from pymongo import common
|
||||
from pymongo.ssl_support import HAVE_SSL
|
||||
from test.version import Version
|
||||
|
||||
if HAVE_SSL:
|
||||
import ssl
|
||||
|
||||
# hostnames retrieved from isMaster will be of unicode type in Python 2,
|
||||
# so ensure these hostnames are unicodes, too. It makes tests like
|
||||
# `test_repr` predictable.
|
||||
@ -53,6 +57,24 @@ port3 = int(os.environ.get("DB_PORT3", 27019))
|
||||
db_user = _unicode(os.environ.get("DB_USER", "user"))
|
||||
db_pwd = _unicode(os.environ.get("DB_PASSWORD", "password"))
|
||||
|
||||
CERT_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)),
|
||||
'certificates')
|
||||
CLIENT_PEM = os.path.join(CERT_PATH, 'client.pem')
|
||||
|
||||
|
||||
def is_server_resolvable():
|
||||
"""Returns True if 'server' is resolvable."""
|
||||
socket_timeout = socket.getdefaulttimeout()
|
||||
socket.setdefaulttimeout(1)
|
||||
try:
|
||||
try:
|
||||
socket.gethostbyname('server')
|
||||
return True
|
||||
except socket.error:
|
||||
return False
|
||||
finally:
|
||||
socket.setdefaulttimeout(socket_timeout)
|
||||
|
||||
|
||||
class client_knobs(object):
|
||||
def __init__(
|
||||
@ -119,18 +141,38 @@ class ClientContext(object):
|
||||
self.is_mongos = False
|
||||
self.is_rs = False
|
||||
self.has_ipv6 = False
|
||||
self.ssl_cert_none = False
|
||||
self.ssl_certfile = False
|
||||
self.server_is_resolvable = is_server_resolvable()
|
||||
|
||||
try:
|
||||
client = pymongo.MongoClient(host, port,
|
||||
serverSelectionTimeoutMS=100)
|
||||
client.admin.command('ismaster') # Can we connect?
|
||||
self.client = self.rs_or_standalone_client = None
|
||||
|
||||
# If so, then reset client to defaults.
|
||||
self.client = pymongo.MongoClient(host, port)
|
||||
def connect(**kwargs):
|
||||
try:
|
||||
client = pymongo.MongoClient(
|
||||
host, port, serverSelectionTimeoutMS=100, **kwargs)
|
||||
client.admin.command('ismaster') # Can we connect?
|
||||
# If connected, then return client with default timeout
|
||||
return pymongo.MongoClient(host, port, **kwargs)
|
||||
except pymongo.errors.ConnectionFailure:
|
||||
return None
|
||||
|
||||
except pymongo.errors.ConnectionFailure:
|
||||
self.client = self.rs_or_standalone_client = None
|
||||
else:
|
||||
self.client = connect()
|
||||
|
||||
if HAVE_SSL and not self.client:
|
||||
# Is MongoDB configured for SSL?
|
||||
self.client = connect(ssl=True, ssl_cert_reqs=ssl.CERT_NONE)
|
||||
if self.client:
|
||||
self.ssl_cert_none = True
|
||||
|
||||
# Can client connect with certfile?
|
||||
client = connect(ssl=True, ssl_cert_reqs=ssl.CERT_NONE,
|
||||
ssl_certfile=CLIENT_PEM,)
|
||||
if client:
|
||||
self.ssl_certfile = True
|
||||
self.client = client
|
||||
|
||||
if self.client:
|
||||
self.connected = True
|
||||
self.ismaster = self.client.admin.command('ismaster')
|
||||
self.w = len(self.ismaster.get("hosts", [])) or 1
|
||||
@ -338,6 +380,36 @@ class ClientContext(object):
|
||||
"Test commands must be enabled",
|
||||
func=func)
|
||||
|
||||
def require_ssl(self, func):
|
||||
"""Run a test only if the client can connect over SSL."""
|
||||
return self._require(self.ssl_cert_none or self.ssl_certfile,
|
||||
"Must be able to connect via SSL",
|
||||
func=func)
|
||||
|
||||
def require_no_ssl(self, func):
|
||||
"""Run a test only if the client can connect over SSL."""
|
||||
return self._require(not (self.ssl_cert_none or self.ssl_certfile),
|
||||
"Must be able to connect without SSL",
|
||||
func=func)
|
||||
|
||||
def require_ssl_cert_none(self, func):
|
||||
"""Run a test only if the client can connect with ssl.CERT_NONE."""
|
||||
return self._require(self.ssl_cert_none,
|
||||
"Must be able to connect with ssl.CERT_NONE",
|
||||
func=func)
|
||||
|
||||
def require_ssl_certfile(self, func):
|
||||
"""Run a test only if the client can connect with ssl_certfile."""
|
||||
return self._require(self.ssl_certfile,
|
||||
"Must be able to connect with ssl_certfile",
|
||||
func=func)
|
||||
|
||||
def require_server_resolvable(self, func):
|
||||
"""Run a test only if the hostname 'server' is resolvable."""
|
||||
return self._require(self.server_is_resolvable,
|
||||
"No hosts entry for 'server'. Cannot validate "
|
||||
"hostname in the certificate",
|
||||
func=func)
|
||||
|
||||
# Reusable client context
|
||||
client_context = ClientContext()
|
||||
|
||||
273
test/test_ssl.py
273
test/test_ssl.py
@ -31,14 +31,20 @@ from pymongo.errors import (ConfigurationError,
|
||||
ConnectionFailure,
|
||||
OperationFailure)
|
||||
from pymongo.ssl_support import HAVE_SSL, get_ssl_context, validate_cert_reqs
|
||||
from test import (host,
|
||||
from pymongo.write_concern import WriteConcern
|
||||
from test import (IntegrationTest,
|
||||
client_context,
|
||||
db_pwd,
|
||||
db_user,
|
||||
host,
|
||||
pair,
|
||||
port,
|
||||
SkipTest,
|
||||
unittest)
|
||||
from test.utils import server_started_with_auth, remove_all_users, connected
|
||||
from test.version import Version
|
||||
from test.utils import remove_all_users, connected
|
||||
|
||||
if HAVE_SSL:
|
||||
import ssl
|
||||
|
||||
CERT_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)),
|
||||
'certificates')
|
||||
@ -46,9 +52,6 @@ CLIENT_PEM = os.path.join(CERT_PATH, 'client.pem')
|
||||
CLIENT_ENCRYPTED_PEM = os.path.join(CERT_PATH, 'client_encrypted.pem')
|
||||
CA_PEM = os.path.join(CERT_PATH, 'ca.pem')
|
||||
CRL_PEM = os.path.join(CERT_PATH, 'crl.pem')
|
||||
SIMPLE_SSL = False
|
||||
CERT_SSL = False
|
||||
SERVER_IS_RESOLVABLE = False
|
||||
MONGODB_X509_USERNAME = (
|
||||
"C=US,ST=California,L=Palo Alto,O=,OU=Drivers,CN=client")
|
||||
|
||||
@ -62,65 +65,12 @@ MONGODB_X509_USERNAME = (
|
||||
# Note: For all replica set tests to pass, the replica set configuration must
|
||||
# use 'server' for the hostname of all hosts.
|
||||
|
||||
def is_server_resolvable():
|
||||
"""Returns True if 'server' is resolvable."""
|
||||
socket_timeout = socket.getdefaulttimeout()
|
||||
socket.setdefaulttimeout(1)
|
||||
try:
|
||||
try:
|
||||
socket.gethostbyname('server')
|
||||
return True
|
||||
except socket.error:
|
||||
return False
|
||||
finally:
|
||||
socket.setdefaulttimeout(socket_timeout)
|
||||
|
||||
|
||||
# Shared ssl-enabled client for the tests
|
||||
ssl_client = None
|
||||
|
||||
if HAVE_SSL:
|
||||
import ssl
|
||||
|
||||
# Check this all once instead of before every test method below.
|
||||
|
||||
# Is MongoDB configured for SSL?
|
||||
try:
|
||||
connected(MongoClient(host, port, ssl=True,
|
||||
ssl_cert_reqs=ssl.CERT_NONE,
|
||||
serverSelectionTimeoutMS=100))
|
||||
|
||||
SIMPLE_SSL = True
|
||||
except ConnectionFailure:
|
||||
pass
|
||||
|
||||
# Is MongoDB configured with server.pem, ca.pem, and crl.pem from
|
||||
# mongodb jstests/lib?
|
||||
try:
|
||||
ssl_client = connected(MongoClient(
|
||||
host, port, ssl=True, ssl_certfile=CLIENT_PEM,
|
||||
ssl_cert_reqs=ssl.CERT_NONE,
|
||||
serverSelectionTimeoutMS=100))
|
||||
|
||||
CERT_SSL = True
|
||||
except ConnectionFailure:
|
||||
pass
|
||||
|
||||
if CERT_SSL:
|
||||
SERVER_IS_RESOLVABLE = is_server_resolvable()
|
||||
|
||||
|
||||
class TestClientSSL(unittest.TestCase):
|
||||
|
||||
@unittest.skipIf(HAVE_SSL, "The ssl module is available, can't test what "
|
||||
"happens without it.")
|
||||
def test_no_ssl_module(self):
|
||||
# Test that ConfigurationError is raised if the ssl
|
||||
# module isn't available.
|
||||
if HAVE_SSL:
|
||||
raise SkipTest(
|
||||
"The ssl module is available, can't test what happens "
|
||||
"without it."
|
||||
)
|
||||
|
||||
# Explicit
|
||||
self.assertRaises(ConfigurationError,
|
||||
MongoClient, ssl=True)
|
||||
@ -129,6 +79,7 @@ class TestClientSSL(unittest.TestCase):
|
||||
self.assertRaises(ConfigurationError,
|
||||
MongoClient, ssl_certfile=CLIENT_PEM)
|
||||
|
||||
@unittest.skipUnless(HAVE_SSL, "The ssl module is not available.")
|
||||
def test_config_ssl(self):
|
||||
# Tests various ssl configurations
|
||||
self.assertRaises(ValueError, MongoClient, ssl='foo')
|
||||
@ -193,42 +144,33 @@ class TestClientSSL(unittest.TestCase):
|
||||
ssl.CERT_REQUIRED)
|
||||
|
||||
|
||||
class TestSSL(unittest.TestCase):
|
||||
class TestSSL(IntegrationTest):
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
if not HAVE_SSL:
|
||||
raise SkipTest("The ssl module is not available.")
|
||||
def assertClientWorks(self, client):
|
||||
coll = client.pymongo_test.ssl_test.with_options(
|
||||
write_concern=WriteConcern(w=client_context.w))
|
||||
coll.drop()
|
||||
coll.insert_one({'ssl': True})
|
||||
self.assertTrue(coll.find_one()['ssl'])
|
||||
coll.drop()
|
||||
|
||||
@unittest.skipUnless(HAVE_SSL, "The ssl module is not available.")
|
||||
def setUp(self):
|
||||
super(TestSSL, self).setUp()
|
||||
|
||||
@client_context.require_ssl
|
||||
def test_simple_ssl(self):
|
||||
# Expects the server to be running with ssl and with
|
||||
# no --sslPEMKeyFile or with --sslWeakCertificateValidation
|
||||
if not SIMPLE_SSL:
|
||||
raise SkipTest("No simple mongod available over SSL")
|
||||
|
||||
client = MongoClient(host, port, ssl=True, ssl_cert_reqs=ssl.CERT_NONE)
|
||||
response = client.admin.command('ismaster')
|
||||
if 'setName' in response:
|
||||
client = MongoClient(pair,
|
||||
replicaSet=response['setName'],
|
||||
w=len(response['hosts']),
|
||||
ssl=True,
|
||||
ssl_cert_reqs=ssl.CERT_NONE)
|
||||
|
||||
db = client.pymongo_ssl_test
|
||||
db.test.drop()
|
||||
db.test.insert_one({'ssl': True})
|
||||
self.assertTrue(db.test.find_one()['ssl'])
|
||||
client.drop_database('pymongo_ssl_test')
|
||||
self.assertClientWorks(self.client)
|
||||
|
||||
@client_context.require_ssl_certfile
|
||||
@client_context.require_server_resolvable
|
||||
def test_ssl_pem_passphrase(self):
|
||||
# Expects the server to be running with server.pem and ca.pem
|
||||
#
|
||||
# --sslPEMKeyFile=/path/to/pymongo/test/certificates/server.pem
|
||||
# --sslCAFile=/path/to/pymongo/test/certificates/ca.pem
|
||||
if not CERT_SSL:
|
||||
raise SkipTest("No mongod available over SSL with certs")
|
||||
|
||||
vi = sys.version_info
|
||||
if vi[0] == 2 and vi < (2, 7, 9) or vi[0] == 3 and vi < (3, 3):
|
||||
self.assertRaises(
|
||||
@ -253,46 +195,20 @@ class TestSSL(unittest.TestCase):
|
||||
"&ssl_ca_certs=%s&serverSelectionTimeoutMS=100")
|
||||
connected(MongoClient(uri_fmt % (CLIENT_ENCRYPTED_PEM, CA_PEM)))
|
||||
|
||||
def test_cert_ssl(self):
|
||||
# Expects the server to be running with server.pem and ca.pem.
|
||||
#
|
||||
# --sslPEMKeyFile=/path/to/pymongo/test/certificates/server.pem
|
||||
# --sslCAFile=/path/to/pymongo/test/certificates/ca.pem
|
||||
#
|
||||
# Also requires an /etc/hosts entry where "server" is resolvable
|
||||
if not CERT_SSL:
|
||||
raise SkipTest("No mongod available over SSL with certs")
|
||||
|
||||
client = ssl_client
|
||||
response = ssl_client.admin.command('ismaster')
|
||||
if 'setName' in response:
|
||||
client = MongoClient(pair,
|
||||
replicaSet=response['setName'],
|
||||
w=len(response['hosts']),
|
||||
ssl=True,
|
||||
ssl_cert_reqs=ssl.CERT_NONE,
|
||||
ssl_certfile=CLIENT_PEM)
|
||||
|
||||
db = client.pymongo_ssl_test
|
||||
db.test.drop()
|
||||
db.test.insert_one({'ssl': True})
|
||||
self.assertTrue(db.test.find_one()['ssl'])
|
||||
client.drop_database('pymongo_ssl_test')
|
||||
|
||||
@client_context.require_ssl_certfile
|
||||
@client_context.require_no_auth
|
||||
def test_cert_ssl_implicitly_set(self):
|
||||
# Expects the server to be running with server.pem and ca.pem
|
||||
#
|
||||
# --sslPEMKeyFile=/path/to/pymongo/test/certificates/server.pem
|
||||
# --sslCAFile=/path/to/pymongo/test/certificates/ca.pem
|
||||
#
|
||||
# Also requires an /etc/hosts entry where "server" is resolvable
|
||||
if not CERT_SSL:
|
||||
raise SkipTest("No mongod available over SSL with certs")
|
||||
|
||||
# test that setting ssl_certfile causes ssl to be set to True
|
||||
client = MongoClient(host, port,
|
||||
ssl_cert_reqs=ssl.CERT_NONE,
|
||||
ssl_certfile=CLIENT_PEM)
|
||||
response = ssl_client.admin.command('ismaster')
|
||||
response = client.admin.command('ismaster')
|
||||
if 'setName' in response:
|
||||
client = MongoClient(pair,
|
||||
replicaSet=response['setName'],
|
||||
@ -300,26 +216,17 @@ class TestSSL(unittest.TestCase):
|
||||
ssl_cert_reqs=ssl.CERT_NONE,
|
||||
ssl_certfile=CLIENT_PEM)
|
||||
|
||||
db = client.pymongo_ssl_test
|
||||
db.test.drop()
|
||||
db.test.insert_one({'ssl': True})
|
||||
self.assertTrue(db.test.find_one()['ssl'])
|
||||
client.drop_database('pymongo_ssl_test')
|
||||
self.assertClientWorks(client)
|
||||
|
||||
@client_context.require_ssl_certfile
|
||||
@client_context.require_server_resolvable
|
||||
@client_context.require_no_auth
|
||||
def test_cert_ssl_validation(self):
|
||||
# Expects the server to be running with server.pem and ca.pem
|
||||
#
|
||||
# --sslPEMKeyFile=/path/to/pymongo/test/certificates/server.pem
|
||||
# --sslCAFile=/path/to/pymongo/test/certificates/ca.pem
|
||||
#
|
||||
# Also requires an /etc/hosts entry where "server" is resolvable
|
||||
if not CERT_SSL:
|
||||
raise SkipTest("No mongod available over SSL with certs")
|
||||
|
||||
if not SERVER_IS_RESOLVABLE:
|
||||
raise SkipTest("No hosts entry for 'server'. Cannot validate "
|
||||
"hostname in the certificate")
|
||||
|
||||
client = MongoClient('server',
|
||||
ssl=True,
|
||||
ssl_certfile=CLIENT_PEM,
|
||||
@ -339,50 +246,31 @@ class TestSSL(unittest.TestCase):
|
||||
ssl_cert_reqs=ssl.CERT_REQUIRED,
|
||||
ssl_ca_certs=CA_PEM)
|
||||
|
||||
db = client.pymongo_ssl_test
|
||||
db.test.drop()
|
||||
db.test.insert_one({'ssl': True})
|
||||
self.assertTrue(db.test.find_one()['ssl'])
|
||||
client.drop_database('pymongo_ssl_test')
|
||||
self.assertClientWorks(client)
|
||||
|
||||
@client_context.require_ssl_certfile
|
||||
@client_context.require_server_resolvable
|
||||
@client_context.require_no_auth
|
||||
def test_cert_ssl_uri_support(self):
|
||||
# Expects the server to be running with server.pem and ca.pem.
|
||||
# Expects the server to be running with server.pem and ca.pem
|
||||
#
|
||||
# --sslPEMKeyFile=/path/to/pymongo/test/certificates/server.pem
|
||||
# --sslCAFile=/path/to/pymongo/test/certificates/ca.pem
|
||||
#
|
||||
# Also requires an /etc/hosts entry where "server" is resolvable
|
||||
if not CERT_SSL:
|
||||
raise SkipTest("No mongod available over SSL with certs")
|
||||
|
||||
if not SERVER_IS_RESOLVABLE:
|
||||
raise SkipTest("No hosts entry for 'server'. Cannot validate "
|
||||
"hostname in the certificate")
|
||||
|
||||
uri_fmt = ("mongodb://server/?ssl=true&ssl_certfile=%s&ssl_cert_reqs"
|
||||
"=%s&ssl_ca_certs=%s&ssl_match_hostname=true")
|
||||
client = MongoClient(uri_fmt % (CLIENT_PEM, 'CERT_REQUIRED', CA_PEM))
|
||||
self.assertClientWorks(client)
|
||||
|
||||
db = client.pymongo_ssl_test
|
||||
db.test.drop()
|
||||
db.test.insert_one({'ssl': True})
|
||||
self.assertTrue(db.test.find_one()['ssl'])
|
||||
client.drop_database('pymongo_ssl_test')
|
||||
|
||||
@client_context.require_ssl_certfile
|
||||
@client_context.require_server_resolvable
|
||||
@client_context.require_no_auth
|
||||
def test_cert_ssl_validation_optional(self):
|
||||
# Expects the server to be running with server.pem and ca.pem.
|
||||
# Expects the server to be running with server.pem and ca.pem
|
||||
#
|
||||
# --sslPEMKeyFile=/path/to/pymongo/test/certificates/server.pem
|
||||
# --sslCAFile=/path/to/pymongo/test/certificates/ca.pem
|
||||
#
|
||||
# Also requires an /etc/hosts entry where "server" is resolvable
|
||||
if not CERT_SSL:
|
||||
raise SkipTest("No mongod available over SSL with certs")
|
||||
|
||||
if not SERVER_IS_RESOLVABLE:
|
||||
raise SkipTest("No hosts entry for 'server'. Cannot validate "
|
||||
"hostname in the certificate")
|
||||
|
||||
client = MongoClient('server',
|
||||
ssl=True,
|
||||
ssl_certfile=CLIENT_PEM,
|
||||
@ -403,21 +291,16 @@ class TestSSL(unittest.TestCase):
|
||||
ssl_cert_reqs=ssl.CERT_OPTIONAL,
|
||||
ssl_ca_certs=CA_PEM)
|
||||
|
||||
db = client.pymongo_ssl_test
|
||||
db.test.drop()
|
||||
db.test.insert_one({'ssl': True})
|
||||
self.assertTrue(db.test.find_one()['ssl'])
|
||||
client.drop_database('pymongo_ssl_test')
|
||||
self.assertClientWorks(client)
|
||||
|
||||
@client_context.require_ssl_certfile
|
||||
def test_cert_ssl_validation_hostname_matching(self):
|
||||
# Expects the server to be running with server.pem and ca.pem
|
||||
#
|
||||
# --sslPEMKeyFile=/path/to/pymongo/test/certificates/server.pem
|
||||
# --sslCAFile=/path/to/pymongo/test/certificates/ca.pem
|
||||
if not CERT_SSL:
|
||||
raise SkipTest("No mongod available over SSL with certs")
|
||||
|
||||
response = ssl_client.admin.command('ismaster')
|
||||
#
|
||||
response = self.client.admin.command('ismaster')
|
||||
|
||||
with self.assertRaises(ConnectionFailure):
|
||||
connected(MongoClient(pair,
|
||||
@ -435,7 +318,6 @@ class TestSSL(unittest.TestCase):
|
||||
ssl_match_hostname=False,
|
||||
serverSelectionTimeoutMS=100))
|
||||
|
||||
|
||||
if 'setName' in response:
|
||||
with self.assertRaises(ConnectionFailure):
|
||||
connected(MongoClient(pair,
|
||||
@ -455,14 +337,9 @@ class TestSSL(unittest.TestCase):
|
||||
ssl_match_hostname=False,
|
||||
serverSelectionTimeoutMS=100))
|
||||
|
||||
@client_context.require_ssl_certfile
|
||||
@client_context.require_server_resolvable
|
||||
def test_ssl_crlfile_support(self):
|
||||
# Expects the server to be running with server.pem and ca.pem
|
||||
#
|
||||
# --sslPEMKeyFile=/path/to/pymongo/test/certificates/server.pem
|
||||
# --sslCAFile=/path/to/pymongo/test/certificates/ca.pem
|
||||
if not CERT_SSL:
|
||||
raise SkipTest("No mongod available over SSL with certs")
|
||||
|
||||
if not hasattr(ssl, 'VERIFY_CRL_CHECK_LEAF'):
|
||||
self.assertRaises(
|
||||
ConfigurationError,
|
||||
@ -494,6 +371,8 @@ class TestSSL(unittest.TestCase):
|
||||
with self.assertRaises(ConnectionFailure):
|
||||
connected(MongoClient(uri_fmt % (CRL_PEM, CA_PEM)))
|
||||
|
||||
@client_context.require_ssl_certfile
|
||||
@client_context.require_server_resolvable
|
||||
def test_validation_with_system_ca_certs(self):
|
||||
# Expects the server to be running with server.pem and ca.pem.
|
||||
#
|
||||
@ -501,21 +380,12 @@ class TestSSL(unittest.TestCase):
|
||||
# --sslCAFile=/path/to/pymongo/test/certificates/ca.pem
|
||||
# --sslWeakCertificateValidation
|
||||
#
|
||||
# Also requires an /etc/hosts entry where "server" is resolvable
|
||||
if not CERT_SSL:
|
||||
raise SkipTest("No mongod available over SSL with certs")
|
||||
|
||||
if not SERVER_IS_RESOLVABLE:
|
||||
raise SkipTest("No hosts entry for 'server'. Cannot validate "
|
||||
"hostname in the certificate")
|
||||
|
||||
if sys.platform == "win32":
|
||||
raise SkipTest("Can't test system ca certs on Windows.")
|
||||
|
||||
if sys.version_info < (2, 7, 9):
|
||||
raise SkipTest("Can't load system CA certificates.")
|
||||
|
||||
|
||||
# Tell OpenSSL where CA certificates live.
|
||||
os.environ['SSL_CERT_FILE'] = CA_PEM
|
||||
try:
|
||||
@ -605,29 +475,17 @@ class TestSSL(unittest.TestCase):
|
||||
ssl_sock = ctx.wrap_socket(socket.socket())
|
||||
self.assertEqual(ssl_sock.ca_certs, ssl_support._WINCERTS.name)
|
||||
|
||||
@client_context.require_version_min(2, 5, 3, -1)
|
||||
@client_context.require_auth
|
||||
@client_context.require_ssl_certfile
|
||||
def test_mongodb_x509_auth(self):
|
||||
# Expects the server to be running with the server.pem and ca.pem
|
||||
# as well as --auth
|
||||
#
|
||||
# --sslPEMKeyFile=/path/to/pymongo/test/certificates/server.pem
|
||||
# --sslCAFile=/path/to/pymongo/test/certificates/ca.pem
|
||||
# --auth
|
||||
if not CERT_SSL:
|
||||
raise SkipTest("No mongod available over SSL with certs")
|
||||
|
||||
if not Version.from_client(ssl_client).at_least(2, 5, 3, -1):
|
||||
raise SkipTest("MONGODB-X509 tests require MongoDB 2.5.3 or newer")
|
||||
if not server_started_with_auth(ssl_client):
|
||||
raise SkipTest('Authentication is not enabled on server')
|
||||
|
||||
self.addCleanup(ssl_client['$external'].logout)
|
||||
ssl_client = MongoClient(pair, ssl=True, ssl_cert_reqs=ssl.CERT_NONE,
|
||||
ssl_certfile=CLIENT_PEM)
|
||||
self.addCleanup(remove_all_users, ssl_client['$external'])
|
||||
self.addCleanup(remove_all_users, ssl_client.admin)
|
||||
|
||||
ssl_client.admin.add_user('admin', 'pass')
|
||||
ssl_client.admin.authenticate('admin', 'pass')
|
||||
ssl_client.admin.authenticate(db_user, db_pwd)
|
||||
|
||||
# Give admin all necessary privileges.
|
||||
# Give x509 user all necessary privileges.
|
||||
ssl_client['$external'].add_user(MONGODB_X509_USERNAME, roles=[
|
||||
{'role': 'readWriteAnyDatabase', 'db': 'admin'},
|
||||
{'role': 'userAdminAnyDatabase', 'db': 'admin'}])
|
||||
@ -643,8 +501,9 @@ class TestSSL(unittest.TestCase):
|
||||
'MONGODB-X509' % (
|
||||
quote_plus(MONGODB_X509_USERNAME), host, port))
|
||||
# SSL options aren't supported in the URI...
|
||||
self.assertTrue(MongoClient(uri,
|
||||
ssl=True, ssl_certfile=CLIENT_PEM))
|
||||
self.assertTrue(MongoClient(uri, ssl=True,
|
||||
ssl_cert_reqs=ssl.CERT_NONE,
|
||||
ssl_certfile=CLIENT_PEM))
|
||||
|
||||
# Should require a username
|
||||
uri = ('mongodb://%s:%d/?authMechanism=MONGODB-X509' % (host,
|
||||
@ -675,10 +534,10 @@ class TestSSL(unittest.TestCase):
|
||||
try:
|
||||
connected(MongoClient(uri,
|
||||
ssl=True,
|
||||
ssl_cert_reqs="CERT_NONE",
|
||||
ssl_cert_reqs=ssl.CERT_NONE,
|
||||
ssl_certfile=CA_PEM,
|
||||
serverSelectionTimeoutMS=100))
|
||||
except OperationFailure:
|
||||
except (ConnectionFailure, ssl.SSLError):
|
||||
pass
|
||||
else:
|
||||
self.fail("Invalid certificate accepted.")
|
||||
|
||||
Loading…
Reference in New Issue
Block a user