vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Initial LDAP group member iteration support

Browse source 
Add interface for group member iteration on internal auth providers and implement support in the LDAP interface.
This commit is contained in:
Joseph Schorr 2017-02-16 15:16:47 -05:00
parent df235d9315
commit d718829f5d
4 changed files with 141 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
data/users/__init__.py
View file

@ -186,6 +186,16 @@ class UserAuthentication(object):
""" Verifies that the given username and password credentials are valid. """
return self.state.verify_credentials(username_or_email, password)
def iterate_group_members(self, group_lookup_args, page_size=None, disable_pagination=False):
""" Returns a tuple of an iterator over all the members of the group matching the given lookup
args dictionary, or the error that occurred if the initial call failed or is unsupported.
The format of the lookup args dictionary is specific to the implementation.
Each result in the iterator is a tuple of (UserInformation, error_message), and only
one will be not-None.
"""
return self.state.iterate_group_members(group_lookup_args, page_size=page_size,
disable_pagination=disable_pagination)
def verify_and_link_user(self, username_or_email, password, basic_auth=False):
""" Verifies that the given username and password credentials are valid and, if so,
creates or links the database user to the federated identity. """

73
data/users/externalldap.py
View file

@ -2,6 +2,8 @@ import ldap
import logging
import os
from ldap.controls import SimplePagedResultsControl
from collections import namedtuple
from data.users.federated import FederatedUsers, UserInformation
from util.itertoolrecipes import take
@ -10,6 +12,7 @@ logger = logging.getLogger(__name__)
_DEFAULT_NETWORK_TIMEOUT = 10.0 # seconds
_DEFAULT_TIMEOUT = 10.0 # seconds
_DEFAULT_PAGE_SIZE = 500
class LDAPConnectionBuilder(object):
@ -84,6 +87,7 @@ class LDAPUsers(FederatedUsers):
# Create the set of full DN paths.
self._user_dns = [get_full_rdn(relative_dn) for relative_dn in relative_user_dns]
self._base_dn = ','.join(base_dn)
def _get_ldap_referral_dn(self, referral_exception):
logger.debug('Got referral: %s', referral_exception.args[0])
@ -174,7 +178,7 @@ class LDAPUsers(FederatedUsers):
with_mail = [result for result in with_dns if result.attrs.get(self._email_attr)]
return (with_mail[0] if with_mail else with_dns[0], None)
def _credential_for_user(self, response):
def _build_user_information(self, response):
if not response.get(self._uid_attr):
return (None, 'Missing uid field "%s" in user record' % self._uid_attr)
@ -194,7 +198,7 @@ class LDAPUsers(FederatedUsers):
logger.debug('Found user for LDAP username or email %s', username_or_email)
_, found_response = found_user
return self._credential_for_user(found_response)
return self._build_user_information(found_response)
def query_users(self, query, limit=20):
""" Queries LDAP for matching users. """
@ -208,7 +212,7 @@ class LDAPUsers(FederatedUsers):
final_results = []
for result in results[0:limit]:
credentials, err_msg = self._credential_for_user(result.attrs)
credentials, err_msg = self._build_user_information(result.attrs)
if err_msg is not None:
continue
@ -253,4 +257,65 @@ class LDAPUsers(FederatedUsers):
logger.debug('Invalid LDAP credentials')
return (None, 'Invalid password')
return self._credential_for_user(found_response)
return self._build_user_information(found_response)
def iterate_group_members(self, group_lookup_args, page_size=None, disable_pagination=False):
try:
with self._ldap.get_connection():
pass
except ldap.INVALID_CREDENTIALS:
return (None, 'LDAP Admin dn or password is invalid')
group_dn = group_lookup_args['group_dn']
page_size = page_size or _DEFAULT_PAGE_SIZE
return (self._iterate_members(group_dn, page_size, disable_pagination), None)
def _iterate_members(self, group_dn, page_size, disable_pagination):
with self._ldap.get_connection() as conn:
lc = ldap.controls.libldap.SimplePagedResultsControl(criticality=True, size=page_size,
cookie='')
search_flt = '(memberOf=%s,%s)' % (group_dn, self._base_dn)
for user_search_dn in self._user_dns:
# Conduct the initial search for users that are a member of the group.
if disable_pagination:
msgid = conn.search(user_search_dn, ldap.SCOPE_SUBTREE, search_flt)
else:
msgid = conn.search_ext(user_search_dn, ldap.SCOPE_SUBTREE, search_flt, serverctrls=[lc])
while True:
if disable_pagination:
_, rdata = conn.result(msgid)
else:
_, rdata, _, serverctrls = conn.result3(msgid)
# Yield any users found.
for userdata in rdata:
yield self._build_user_information(userdata[1])
# If pagination is disabled, nothing more to do.
if disable_pagination:
break
# Filter down the controls with which the server responded, looking for the paging
# control type. If not found, then the server does not support pagination and we already
# got all of the results.
pctrls = [control for control in serverctrls
if control.controlType == ldap.controls.SimplePagedResultsControl.controlType]
if pctrls:
# Server supports pagination. Update the cookie so the next search finds the next page,
# then conduct the next search.
cookie = lc.cookie = pctrls[0].cookie
if cookie:
msgid = conn.search_ext(user_search_dn, ldap.SCOPE_SUBTREE, search_flt,
serverctrls=[lc])
continue
else:
# No additional results.
break
else:
# Pagintation is not supported.
logger.debug('Pagination is not supported for this LDAP server')
break

60
data/users/federated.py
View file

@ -37,33 +37,6 @@ class FederatedUsers(object):
""" If implemented, get_user must be implemented as well. """
return (None, 'Not supported')
def _get_federated_user(self, username, email):
db_user = model.user.verify_federated_login(self._federated_service, username)
if not db_user:
# We must create the user in our db
valid_username = None
for valid_username in generate_valid_usernames(username):
if model.user.is_username_unique(valid_username):
break
if not valid_username:
logger.error('Unable to pick a username for user: %s', username)
return (None, 'Unable to pick a username. Please report this to your administrator.')
prompts = model.user.get_default_user_prompts(features)
db_user = model.user.create_federated_user(valid_username, email, self._federated_service,
username,
set_password_notification=False,
email_required=self._requires_email,
prompts=prompts)
else:
# Update the db attributes from the federated service.
if email:
db_user.email = email
db_user.save()
return (db_user, None)
def link_user(self, username_or_email):
(credentials, err_msg) = self.get_user(username_or_email)
if credentials is None:
@ -98,3 +71,36 @@ class FederatedUsers(object):
return (None, err_msg)
return (db_user, None)
def iterate_group_members(self, group_lookup_args, page_size=None, disable_pagination=False):
""" Returns an iterator over all the members of the group matching the given lookup args
dictionary. The format of the lookup args dictionary is specific to the implementation.
"""
return (None, 'Not supported')
def _get_federated_user(self, username, email):
db_user = model.user.verify_federated_login(self._federated_service, username)
if not db_user:
# We must create the user in our db
valid_username = None
for valid_username in generate_valid_usernames(username):
if model.user.is_username_unique(valid_username):
break
if not valid_username:
logger.error('Unable to pick a username for user: %s', username)
return (None, 'Unable to pick a username. Please report this to your administrator.')
prompts = model.user.get_default_user_prompts(features)
db_user = model.user.create_federated_user(valid_username, email, self._federated_service,
username,
set_password_notification=False,
email_required=self._requires_email,
prompts=prompts)
else:
# Update the db attributes from the federated service.
if email:
db_user.email = email
db_user.save()
return (db_user, None)

32
test/test_ldap.py
View file

@ -34,18 +34,25 @@ def mock_ldap(requires_email=True):
'dc': ['quay', 'io'],
'ou': 'otheremployees'
},
'cn=AwesomeFolk,dc=quay,dc=io': {
'dc': ['quay', 'io'],
'cn': 'AwesomeFolk'
},
'uid=testy,ou=employees,dc=quay,dc=io': {
'dc': ['quay', 'io'],
'ou': 'employees',
'uid': 'testy',
'userPassword': ['password']
'uid': ['testy'],
'userPassword': ['password'],
'mail': ['bar@baz.com'],
'memberOf': ['cn=AwesomeFolk,dc=quay,dc=io'],
},
'uid=someuser,ou=employees,dc=quay,dc=io': {
'dc': ['quay', 'io'],
'ou': 'employees',
'uid': ['someuser'],
'userPassword': ['somepass'],
'mail': ['foo@bar.com']
'mail': ['foo@bar.com'],
'memberOf': ['cn=AwesomeFolk,dc=quay,dc=io'],
},
'uid=nomail,ou=employees,dc=quay,dc=io': {
'dc': ['quay', 'io'],
@ -301,6 +308,25 @@ class TestLDAP(unittest.TestCase):
requires_email=False, timeout=5)
ldap.query_users('cool')
def test_iterate_group_members(self):
with mock_ldap() as ldap:
(it, err) = ldap.iterate_group_members({'group_dn': 'cn=AwesomeFolk'},
disable_pagination=True)
self.assertIsNone(err)
results = list(it)
self.assertEquals(2, len(results))
first = results[0][0]
self.assertEquals('testy', first.id)
self.assertEquals('testy', first.username)
self.assertEquals('bar@baz.com', first.email)
second = results[1][0]
self.assertEquals('someuser', second.id)
self.assertEquals('someuser', second.username)
self.assertEquals('foo@bar.com', second.email)
if __name__ == '__main__':
unittest.main()