mongodb/mongo-python-driver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

PYTHON-1464 - Implement SCRAM-SHA-256

Browse Source
This commit is contained in:
Bernie Hackett 2018-02-20 21:55:22 -08:00
parent 00968e5bb1
commit 5b9257644f
8 changed files with 261 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

94
pymongo/auth.py
View File

@ -14,6 +14,7 @@
"""Authentication helpers."""
import functools
import hmac
import socket
@ -36,17 +37,24 @@ except ImportError:
from base64 import standard_b64decode, standard_b64encode
from collections import namedtuple
from hashlib import md5, sha1
from hashlib import md5, sha1, sha256
from random import SystemRandom
from bson.binary import Binary
from bson.py3compat import b, string_type, _unicode, PY3
from bson.son import SON
from pymongo.errors import ConfigurationError, OperationFailure
from pymongo.saslprep import saslprep
MECHANISMS = frozenset(
['GSSAPI', 'MONGODB-CR', 'MONGODB-X509', 'PLAIN', 'SCRAM-SHA-1', 'DEFAULT'])
['GSSAPI',
'MONGODB-CR',
'MONGODB-X509',
'PLAIN',
'SCRAM-SHA-1',
'SCRAM-SHA-256',
'DEFAULT'])
"""The authentication mechanisms supported by PyMongo."""
@ -66,8 +74,6 @@ GSSAPIProperties = namedtuple('GSSAPIProperties',
def _build_credentials_tuple(mech, source, user, passwd, extra):
"""Build and return a mechanism specific credentials tuple.
"""
user = _unicode(user) if user is not None else None
password = passwd if passwd is None else _unicode(passwd)
if mech == 'GSSAPI':
properties = extra.get('authmechanismproperties', {})
service_name = properties.get('SERVICE_NAME', 'mongodb')
@ -77,14 +83,14 @@ def _build_credentials_tuple(mech, source, user, passwd, extra):
canonicalize_host_name=canonicalize,
service_realm=service_realm)
# Source is always $external.
return MongoCredential(mech, '$external', user, password, props)
return MongoCredential(mech, '$external', user, passwd, props)
elif mech == 'MONGODB-X509':
# user can be None.
return MongoCredential(mech, '$external', user, None, None)
else:
if passwd is None:
raise ConfigurationError("A password is required.")
return MongoCredential(mech, source, user, password, None)
return MongoCredential(mech, source, user, passwd, None)
if PY3:
@ -110,31 +116,32 @@ else:
return int(_hexlify(value), 16)
def _to_bytes(value, dummy0, dummy1, _unhexlify=_unhexlify):
def _to_bytes(value, length, dummy, _unhexlify=_unhexlify):
"""An implementation of int.to_bytes for python 2.x."""
return _unhexlify('%040x' % value)
fmt = '%%0%dx' % (2 * length,)
return _unhexlify(fmt % value)
try:
# The fastest option, if it's been compiled to use OpenSSL's HMAC.
from backports.pbkdf2 import pbkdf2_hmac
def _hi(data, salt, iterations):
return pbkdf2_hmac('sha1', data, salt, iterations)
def _hi(data, salt, iterations, digestmod):
return pbkdf2_hmac(digestmod().name, data, salt, iterations)
except ImportError:
try:
# Python 2.7.8+, or Python 3.4+.
from hashlib import pbkdf2_hmac
def _hi(data, salt, iterations):
return pbkdf2_hmac('sha1', data, salt, iterations)
def _hi(data, salt, iterations, digestmod):
return pbkdf2_hmac(digestmod().name, data, salt, iterations)
except ImportError:
def _hi(data, salt, iterations):
"""A simple implementation of PBKDF2."""
mac = hmac.HMAC(data, None, sha1)
def _hi(data, salt, iterations, digestmod):
"""A simple implementation of PBKDF2-HMAC."""
mac = hmac.HMAC(data, None, digestmod)
def _digest(msg, mac=mac):
"""Get a digest for msg."""
@ -150,7 +157,7 @@ except ImportError:
for _ in range(iterations - 1):
_u1 = _digest(_u1)
_ui ^= from_bytes(_u1, 'big')
return to_bytes(_ui, 20, 'big')
return to_bytes(_ui, mac.digest_size, 'big')
try:
from hmac import compare_digest
@ -187,15 +194,21 @@ def _parse_scram_response(response):
return dict(item.split(b"=", 1) for item in response.split(b","))
def _authenticate_scram_sha1(credentials, sock_info):
"""Authenticate using SCRAM-SHA-1."""
username = credentials.username
password = credentials.password
def _authenticate_scram(credentials, sock_info, mechanism):
"""Authenticate using SCRAM."""
if mechanism == 'SCRAM-SHA-256':
digestmod = sha256
username = saslprep(credentials.username)
data = saslprep(credentials.password).encode("utf-8")
else:
digestmod = sha1
username = credentials.username
data = _password_digest(username, credentials.password).encode("utf-8")
source = credentials.source
# Make local
_hmac = hmac.HMAC
_sha1 = sha1
user = username.encode("utf-8").replace(b"=", b"=3D").replace(b",", b"=2C")
nonce = standard_b64encode(
@ -203,7 +216,7 @@ def _authenticate_scram_sha1(credentials, sock_info):
first_bare = b"n=" + user + b",r=" + nonce
cmd = SON([('saslStart', 1),
('mechanism', 'SCRAM-SHA-1'),
('mechanism', mechanism),
('payload', Binary(b"n,," + first_bare)),
('autoAuthorize', 1)])
res = sock_info.command(source, cmd)
@ -211,25 +224,26 @@ def _authenticate_scram_sha1(credentials, sock_info):
server_first = res['payload']
parsed = _parse_scram_response(server_first)
iterations = int(parsed[b'i'])
if iterations < 4096:
raise OperationFailure("Server returned an invalid iteration count.")
salt = parsed[b's']
rnonce = parsed[b'r']
if not rnonce.startswith(nonce):
raise OperationFailure("Server returned an invalid nonce.")
without_proof = b"c=biws,r=" + rnonce
salted_pass = _hi(_password_digest(username, password).encode("utf-8"),
standard_b64decode(salt),
iterations)
client_key = _hmac(salted_pass, b"Client Key", _sha1).digest()
stored_key = _sha1(client_key).digest()
salted_pass = _hi(
data, standard_b64decode(salt), iterations, digestmod)
client_key = _hmac(salted_pass, b"Client Key", digestmod).digest()
stored_key = digestmod(client_key).digest()
auth_msg = b",".join((first_bare, server_first, without_proof))
client_sig = _hmac(stored_key, auth_msg, _sha1).digest()
client_sig = _hmac(stored_key, auth_msg, digestmod).digest()
client_proof = b"p=" + standard_b64encode(_xor(client_key, client_sig))
client_final = b",".join((without_proof, client_proof))
server_key = _hmac(salted_pass, b"Server Key", _sha1).digest()
server_key = _hmac(salted_pass, b"Server Key", digestmod).digest()
server_sig = standard_b64encode(
_hmac(server_key, auth_msg, _sha1).digest())
_hmac(server_key, auth_msg, digestmod).digest())
cmd = SON([('saslContinue', 1),
('conversationId', res['conversationId']),
@ -462,8 +476,19 @@ def _authenticate_mongo_cr(credentials, sock_info):
def _authenticate_default(credentials, sock_info):
if sock_info.max_wire_version >= 3:
return _authenticate_scram_sha1(credentials, sock_info)
if sock_info.max_wire_version >= 7:
source = credentials.source
cmd = SON([
('ismaster', 1),
('saslSupportedMechs', source + '.' + credentials.username)])
mechs = sock_info.command(
source, cmd, publish_events=False).get('saslSupportedMechs', [])
if 'SCRAM-SHA-256' in mechs:
return _authenticate_scram(credentials, sock_info, 'SCRAM-SHA-256')
else:
return _authenticate_scram(credentials, sock_info, 'SCRAM-SHA-1')
elif sock_info.max_wire_version >= 3:
return _authenticate_scram(credentials, sock_info, 'SCRAM-SHA-1')
else:
return _authenticate_mongo_cr(credentials, sock_info)
@ -474,7 +499,10 @@ _AUTH_MAP = {
'MONGODB-CR': _authenticate_mongo_cr,
'MONGODB-X509': _authenticate_x509,
'PLAIN': _authenticate_plain,
'SCRAM-SHA-1': _authenticate_scram_sha1,
'SCRAM-SHA-1': functools.partial(
_authenticate_scram, mechanism='SCRAM-SHA-1'),
'SCRAM-SHA-256': functools.partial(
_authenticate_scram, mechanism='SCRAM-SHA-256'),
'DEFAULT': _authenticate_default,
}

13
pymongo/database.py
View File

@ -939,14 +939,12 @@ class Database(common.BaseObject):
">= 2.6, use 'roles' instead", DeprecationWarning)
if password is not None:
# We always salt and hash client side.
if "digestPassword" in kwargs:
raise ConfigurationError("The digestPassword option is not "
"supported via add_user. Please use "
"db.command('createUser', ...) "
"instead for this option.")
opts["pwd"] = auth._password_digest(name, password)
opts["digestPassword"] = False
opts["pwd"] = password
# Don't send {} as writeConcern.
if self.write_concern.acknowledged and self.write_concern.document:
@ -994,6 +992,9 @@ class Database(common.BaseObject):
.. _updateUser: https://docs.mongodb.com/manual/reference/command/updateUser/
.. _dropUser: https://docs.mongodb.com/manual/reference/command/createUser/
.. warning:: Never create or modify users over an insecure network without
the use of TLS. See :doc:`/examples/tls` for more information.
:Parameters:
- `name`: the name of the user to create
- `password` (optional): the password of the user to create. Can not
@ -1006,6 +1007,9 @@ class Database(common.BaseObject):
- `session` (optional): a
:class:`~pymongo.client_session.ClientSession`.
.. versionchanged:: 3.7
Added support for SCRAM-SHA-256 users with MongoDB 4.0 and later.
.. versionchanged:: 3.6
Added ``session`` parameter. Deprecated add_user.
@ -1122,6 +1126,9 @@ class Database(common.BaseObject):
name for GSSAPI authentication pass
authMechanismProperties='SERVICE_NAME:<service name>'
.. versionchanged:: 3.7
Added support for SCRAM-SHA-256 with MongoDB 4.0 and later.
.. versionchanged:: 3.5
Deprecated. Authenticating multiple users conflicts with support for
logical sessions in MongoDB 3.6. To authenticate as multiple users,

2
pymongo/saslprep.py
View File

@ -19,6 +19,7 @@ from bson.py3compat import text_type as _text_type
try:
import stringprep
except ImportError:
HAVE_STRINGPREP = False
def saslprep(data):
"""SASLprep dummy"""
if isinstance(data, _text_type):
@ -27,6 +28,7 @@ except ImportError:
"passwords must be ASCII strings.")
return data
else:
HAVE_STRINGPREP = True
import unicodedata
# RFC4013 section 2.3 prohibited output.
_PROHIBITED = (

14
test/__init__.py
View File

@ -50,14 +50,12 @@ if HAVE_SSL:
import ssl
# The host and port of a single mongod or mongos, or the seed host
# for a replica set. 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.
host = _unicode(os.environ.get("DB_IP", 'localhost'))
# for a replica set.
host = os.environ.get("DB_IP", 'localhost')
port = int(os.environ.get("DB_PORT", 27017))
db_user = _unicode(os.environ.get("DB_USER", "user"))
db_pwd = _unicode(os.environ.get("DB_PASSWORD", "password"))
db_user = os.environ.get("DB_USER", "user")
db_pwd = os.environ.get("DB_PASSWORD", "password")
CERT_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)),
'certificates')
@ -234,7 +232,7 @@ class ClientContext(object):
self.sessions_enabled = 'logicalSessionTimeoutMinutes' in ismaster
if 'setName' in ismaster:
self.replica_set_name = ismaster['setName']
self.replica_set_name = str(ismaster['setName'])
self.is_rs = True
if self.auth_enabled:
# It doesn't matter which member we use as the seed here.
@ -314,7 +312,7 @@ class ClientContext(object):
def host(self):
if self.is_rs:
primary = self.client.primary
return primary[0] if primary is not None else host
return str(primary[0]) if primary is not None else host
return host
@property

152
test/test_auth.py
View File

@ -26,15 +26,17 @@ except ImportError:
sys.path[0:0] = [""]
from pymongo import MongoClient
from pymongo import MongoClient, monitoring
from pymongo.auth import HAVE_KERBEROS, _build_credentials_tuple
from pymongo.errors import OperationFailure
from pymongo.read_preferences import ReadPreference
from pymongo.saslprep import saslprep, HAVE_STRINGPREP
from test import client_context, SkipTest, unittest, Version
from test.utils import (delay,
ignore_deprecations,
rs_or_single_client_noauth,
single_client_noauth)
single_client_noauth,
WhiteListEventListener)
# YOU MUST RUN KINIT BEFORE RUNNING GSSAPI TESTS ON UNIX.
GSSAPI_HOST = os.environ.get('GSSAPI_HOST')
@ -366,6 +368,152 @@ class TestSCRAMSHA1(unittest.TestCase):
db.command('dbstats')
class TestSCRAM(unittest.TestCase):
@client_context.require_auth
@client_context.require_version_min(3, 7, 2)
def setUp(self):
self._SENSITIVE_COMMANDS = monitoring._SENSITIVE_COMMANDS
monitoring._SENSITIVE_COMMANDS = set([])
self.listener = WhiteListEventListener("saslStart")
def tearDown(self):
monitoring._SENSITIVE_COMMANDS = self._SENSITIVE_COMMANDS
client_context.client.testscram.command("dropAllUsersFromDatabase")
client_context.client.drop_database("testscram")
def test_scram(self):
host, port = client_context.host, client_context.port
client_context.create_user(
'testscram',
'sha1',
'pwd',
roles=['dbOwner'],
mechanisms=['SCRAM-SHA-1'])
client_context.create_user(
'testscram',
'sha256',
'pwd',
roles=['dbOwner'],
mechanisms=['SCRAM-SHA-256'])
client_context.create_user(
'testscram',
'both',
'pwd',
roles=['dbOwner'],
mechanisms=['SCRAM-SHA-1', 'SCRAM-SHA-256'])
client = rs_or_single_client_noauth(
event_listeners=[self.listener])
self.assertTrue(
client.testscram.authenticate('sha1', 'pwd'))
client.testscram.command('dbstats')
client.testscram.logout()
self.assertTrue(
client.testscram.authenticate(
'sha1', 'pwd', mechanism='SCRAM-SHA-1'))
client.testscram.command('dbstats')
client.testscram.logout()
self.assertRaises(
OperationFailure,
client.testscram.authenticate,
'sha1', 'pwd', mechanism='SCRAM-SHA-256')
self.assertTrue(
client.testscram.authenticate('sha256', 'pwd'))
client.testscram.command('dbstats')
client.testscram.logout()
self.assertTrue(
client.testscram.authenticate(
'sha256', 'pwd', mechanism='SCRAM-SHA-256'))
client.testscram.command('dbstats')
client.testscram.logout()
self.assertRaises(
OperationFailure,
client.testscram.authenticate,
'sha256', 'pwd', mechanism='SCRAM-SHA-1')
self.listener.results.clear()
self.assertTrue(
client.testscram.authenticate('both', 'pwd'))
started = self.listener.results['started'][0]
self.assertEqual(started.command.get('mechanism'), 'SCRAM-SHA-256')
client.testscram.command('dbstats')
client.testscram.logout()
self.assertTrue(
client.testscram.authenticate(
'both', 'pwd', mechanism='SCRAM-SHA-256'))
client.testscram.command('dbstats')
client.testscram.logout()
self.assertTrue(
client.testscram.authenticate(
'both', 'pwd', mechanism='SCRAM-SHA-1'))
client.testscram.command('dbstats')
client.testscram.logout()
self.assertRaises(
OperationFailure,
client.testscram.authenticate,
'not-a-user', 'pwd')
if HAVE_STRINGPREP:
client_context.create_user(
'testscram',
saslprep(u'\u2168'),
u'\u2168',
roles=['dbOwner'],
mechanisms=['SCRAM-SHA-256'])
self.assertTrue(
client.testscram.authenticate(u'\u2168', u'\u2168'))
client.testscram.command('dbstats')
client.testscram.logout()
self.assertTrue(
client.testscram.authenticate(
u'\u2168', u'\u2168', mechanism='SCRAM-SHA-256'))
client.testscram.command('dbstats')
client.testscram.logout()
self.assertRaises(
OperationFailure,
client.testscram.authenticate,
u'\u2168', u'\u2168', mechanism='SCRAM-SHA-1')
client = rs_or_single_client_noauth(
u'mongodb://\u2168:\u2168@%s:%d/testscram' % (host, port))
client.testscram.command('dbstats')
self.listener.results.clear()
client = rs_or_single_client_noauth(
'mongodb://both:pwd@%s:%d/testscram' % (host, port),
event_listeners=[self.listener])
client.testscram.command('dbstats')
started = self.listener.results['started'][0]
self.assertEqual(started.command.get('mechanism'), 'SCRAM-SHA-256')
client = rs_or_single_client_noauth(
'mongodb://both:pwd@%s:%d/testscram?authMechanism=SCRAM-SHA-1'
% (host, port))
client.testscram.command('dbstats')
client = rs_or_single_client_noauth(
'mongodb://both:pwd@%s:%d/testscram?authMechanism=SCRAM-SHA-256'
% (host, port))
client.testscram.command('dbstats')
if client_context.is_rs:
uri = ('mongodb://both:pwd@%s:%d/testscram'
'?replicaSet=%s' % (host, port,
client_context.replica_set_name))
client = single_client_noauth(uri)
client.testscram.command('dbstats')
db = client.get_database(
'testscram', read_preference=ReadPreference.SECONDARY)
db.command('dbstats')
class TestAuthURIOptions(unittest.TestCase):
@client_context.require_auth

73
test/test_database.py
View File

@ -45,6 +45,7 @@ from pymongo.errors import (CollectionInvalid,
from pymongo.mongo_client import MongoClient
from pymongo.read_concern import ReadConcern
from pymongo.read_preferences import ReadPreference
from pymongo.saslprep import HAVE_STRINGPREP
from pymongo.write_concern import WriteConcern
from test import (client_context,
SkipTest,
@ -503,12 +504,8 @@ class TestDatabase(IntegrationTest):
self.assertRaises(ConfigurationError, auth_db.add_user,
"user", "password", digestPassword=True)
extra = {}
if client_context.version.at_least(3, 7, 2):
extra['mechanisms'] = ['SCRAM-SHA-1']
# Add / authenticate / remove
auth_db.add_user("mike", "password", roles=["read"], **extra)
auth_db.add_user("mike", "password", roles=["read"])
self.addCleanup(remove_all_users, auth_db)
self.assertRaises(TypeError, check_auth, 5, "password")
self.assertRaises(TypeError, check_auth, "mike", 5)
@ -517,35 +514,34 @@ class TestDatabase(IntegrationTest):
self.assertRaises(OperationFailure, check_auth, "faker", "password")
check_auth("mike", "password")
# Unicode name and password.
check_auth(u"mike", u"password")
if not client_context.version.at_least(3, 7, 2) or HAVE_STRINGPREP:
# Unicode name and password.
check_auth(u"mike", u"password")
auth_db.remove_user("mike")
self.assertRaises(OperationFailure,
check_auth, "mike", "password")
auth_db.remove_user("mike")
self.assertRaises(
OperationFailure, check_auth, "mike", "password")
# Add / authenticate / change password
self.assertRaises(OperationFailure, check_auth, "Gustave", u"Dor\xe9")
auth_db.add_user("Gustave", u"Dor\xe9", roles=["read"], **extra)
check_auth("Gustave", u"Dor\xe9")
# Add / authenticate / change password
self.assertRaises(
OperationFailure, check_auth, "Gustave", u"Dor\xe9")
auth_db.add_user("Gustave", u"Dor\xe9", roles=["read"])
check_auth("Gustave", u"Dor\xe9")
# Change password.
auth_db.add_user("Gustave", "password", roles=["read"], **extra)
self.assertRaises(OperationFailure, check_auth, "Gustave", u"Dor\xe9")
check_auth("Gustave", u"password")
# Change password.
auth_db.add_user("Gustave", "password", roles=["read"])
self.assertRaises(
OperationFailure, check_auth, "Gustave", u"Dor\xe9")
check_auth("Gustave", u"password")
@client_context.require_auth
@ignore_deprecations
def test_make_user_readonly(self):
extra = {}
if client_context.version.at_least(3, 7, 2):
extra['mechanisms'] = ['SCRAM-SHA-1']
# "self.client" is logged in as root.
auth_db = self.client.pymongo_test
# Make a read-write user.
auth_db.add_user('jesse', 'pw', **extra)
auth_db.add_user('jesse', 'pw')
self.addCleanup(remove_all_users, auth_db)
# Check that we're read-write by default.
@ -556,7 +552,7 @@ class TestDatabase(IntegrationTest):
c.pymongo_test.collection.insert_one({})
# Make the user read-only.
auth_db.add_user('jesse', 'pw', read_only=True, **extra)
auth_db.add_user('jesse', 'pw', read_only=True)
c = rs_or_single_client_noauth(username='jesse',
password='pw',
@ -569,13 +565,9 @@ class TestDatabase(IntegrationTest):
@client_context.require_auth
@ignore_deprecations
def test_default_roles(self):
extra = {}
if client_context.version.at_least(3, 7, 2):
extra['mechanisms'] = ['SCRAM-SHA-1']
# "self.client" is logged in as root.
auth_admin = self.client.admin
auth_admin.add_user('test_default_roles', 'pass', **extra)
auth_admin.add_user('test_default_roles', 'pass')
self.addCleanup(client_context.drop_user, 'admin', 'test_default_roles')
info = auth_admin.command(
'usersInfo', 'test_default_roles')['users'][0]
@ -583,33 +575,29 @@ class TestDatabase(IntegrationTest):
self.assertEqual("root", info['roles'][0]['role'])
# Read only "admin" user
auth_admin.add_user('ro-admin', 'pass', read_only=True, **extra)
auth_admin.add_user('ro-admin', 'pass', read_only=True)
self.addCleanup(client_context.drop_user, 'admin', 'ro-admin')
info = auth_admin.command('usersInfo', 'ro-admin')['users'][0]
self.assertEqual("readAnyDatabase", info['roles'][0]['role'])
# "Non-admin" user
auth_db = self.client.pymongo_test
auth_db.add_user('user', 'pass', **extra)
auth_db.add_user('user', 'pass')
self.addCleanup(remove_all_users, auth_db)
info = auth_db.command('usersInfo', 'user')['users'][0]
self.assertEqual("dbOwner", info['roles'][0]['role'])
# Read only "Non-admin" user
auth_db.add_user('ro-user', 'pass', read_only=True, **extra)
auth_db.add_user('ro-user', 'pass', read_only=True)
info = auth_db.command('usersInfo', 'ro-user')['users'][0]
self.assertEqual("read", info['roles'][0]['role'])
@client_context.require_auth
@ignore_deprecations
def test_new_user_cmds(self):
extra = {}
if client_context.version.at_least(3, 7, 2):
extra['mechanisms'] = ['SCRAM-SHA-1']
# "self.client" is logged in as root.
auth_db = self.client.pymongo_test
auth_db.add_user("amalia", "password", roles=["userAdmin"], **extra)
auth_db.add_user("amalia", "password", roles=["userAdmin"])
self.addCleanup(client_context.drop_user, "pymongo_test", "amalia")
db = rs_or_single_client_noauth(username="amalia",
@ -618,7 +606,7 @@ class TestDatabase(IntegrationTest):
# This tests the ability to update user attributes.
db.add_user("amalia", "new_password",
customData={"secret": "koalas"}, **extra)
customData={"secret": "koalas"})
user_info = db.command("usersInfo", "amalia")
self.assertTrue(user_info["users"])
@ -629,10 +617,6 @@ class TestDatabase(IntegrationTest):
@client_context.require_auth
@ignore_deprecations
def test_authenticate_multiple(self):
extra = {}
if client_context.version.at_least(3, 7, 2):
extra['mechanisms'] = ['SCRAM-SHA-1']
# "self.client" is logged in as root.
self.client.drop_database("pymongo_test")
self.client.drop_database("pymongo_test1")
@ -650,12 +634,11 @@ class TestDatabase(IntegrationTest):
admin_db_auth.add_user(
'ro-admin',
'pass',
roles=["userAdmin", "readAnyDatabase"],
**extra)
roles=["userAdmin", "readAnyDatabase"])
self.addCleanup(client_context.drop_user, 'admin', 'ro-admin')
users_db_auth.add_user(
'user', 'pass', roles=["userAdmin", "readWrite"], **extra)
'user', 'pass', roles=["userAdmin", "readWrite"])
self.addCleanup(remove_all_users, users_db_auth)
# Regular user should be able to query its own db, but

8
test/test_session.py
View File

@ -273,15 +273,11 @@ class TestSession(IntegrationTest):
client = self.client
db = client.pymongo_test
extra = {'roles': ['read']}
if client_context.version.at_least(3, 7, 2):
extra['mechanisms'] = ['SCRAM-SHA-1']
self._test_ops(
client,
(db.add_user, ['session-test', 'pass'], extra),
(db.add_user, ['session-test', 'pass'], {'roles': ['read']}),
# Do it again to test updateUser command.
(db.add_user, ['session-test', 'pass'], extra),
(db.add_user, ['session-test', 'pass'], {'roles': ['read']}),
(db.remove_user, ['session-test'], {}))
def test_collection(self):

4
test/utils.py
View File

@ -111,9 +111,9 @@ def _connection_string(h, p, authenticate):
if h.startswith("mongodb://"):
return h
elif client_context.auth_enabled and authenticate:
return "mongodb://%s:%s@%s:%d" % (db_user, db_pwd, h, p)
return "mongodb://%s:%s@%s:%d" % (db_user, db_pwd, str(h), p)
else:
return "mongodb://%s:%d" % (h, p)
return "mongodb://%s:%d" % (str(h), p)
def _mongo_client(host, port, authenticate=True, direct=False, **kwargs):