This repository has been archived on 2020-03-24. You can view files and clone it, but cannot push or open issues or pull requests.
quay/data/users/externalldap.py

324 lines
12 KiB
Python
Raw Normal View History

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
logger = logging.getLogger(__name__)
_DEFAULT_NETWORK_TIMEOUT = 10.0 # seconds
_DEFAULT_TIMEOUT = 10.0 # seconds
_DEFAULT_PAGE_SIZE = 1000
class LDAPConnectionBuilder(object):
def __init__(self, ldap_uri, user_dn, user_pw, allow_tls_fallback=False,
timeout=None, network_timeout=None):
self._ldap_uri = ldap_uri
self._user_dn = user_dn
self._user_pw = user_pw
self._allow_tls_fallback = allow_tls_fallback
self._timeout = timeout
self._network_timeout = network_timeout
def get_connection(self):
return LDAPConnection(self._ldap_uri, self._user_dn, self._user_pw, self._allow_tls_fallback,
self._timeout, self._network_timeout)
class LDAPConnection(object):
def __init__(self, ldap_uri, user_dn, user_pw, allow_tls_fallback=False,
timeout=None, network_timeout=None):
self._ldap_uri = ldap_uri
self._user_dn = user_dn
self._user_pw = user_pw
self._allow_tls_fallback = allow_tls_fallback
self._timeout = timeout
self._network_timeout = network_timeout
self._conn = None
def __enter__(self):
trace_level = 2 if os.environ.get('USERS_DEBUG') == '1' else 0
self._conn = ldap.initialize(self._ldap_uri, trace_level=trace_level)
self._conn.set_option(ldap.OPT_REFERRALS, 1)
self._conn.set_option(ldap.OPT_NETWORK_TIMEOUT,
self._network_timeout or _DEFAULT_NETWORK_TIMEOUT)
self._conn.set_option(ldap.OPT_TIMEOUT, self._timeout or _DEFAULT_TIMEOUT)
if self._allow_tls_fallback:
logger.debug('TLS Fallback enabled in LDAP')
self._conn.set_option(ldap.OPT_X_TLS_TRY, 1)
self._conn.simple_bind_s(self._user_dn, self._user_pw)
return self._conn
def __exit__(self, exc_type, value, tb):
self._conn.unbind_s()
class LDAPUsers(FederatedUsers):
_LDAPResult = namedtuple('LDAPResult', ['dn', 'attrs'])
def __init__(self, ldap_uri, base_dn, admin_dn, admin_passwd, user_rdn, uid_attr, email_attr,
allow_tls_fallback=False, secondary_user_rdns=None, requires_email=True,
timeout=None, network_timeout=None):
super(LDAPUsers, self).__init__('ldap', requires_email)
self._ldap = LDAPConnectionBuilder(ldap_uri, admin_dn, admin_passwd, allow_tls_fallback,
timeout, network_timeout)
self._ldap_uri = ldap_uri
self._uid_attr = uid_attr
self._email_attr = email_attr
self._allow_tls_fallback = allow_tls_fallback
self._requires_email = requires_email
# Note: user_rdn is a list of RDN pieces (for historical reasons), and secondary_user_rds
# is a list of RDN strings.
relative_user_dns = [','.join(user_rdn)] + (secondary_user_rdns or [])
def get_full_rdn(relative_dn):
prefix = relative_dn.split(',') if relative_dn else []
return ','.join(prefix + base_dn)
# 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])
if not referral_exception.args[0] or not referral_exception.args[0].get('info'):
logger.debug('LDAP referral missing info block')
return None
referral_info = referral_exception.args[0]['info']
if not referral_info.startswith('Referral:\n'):
logger.debug('LDAP referral missing Referral header')
return None
referral_uri = referral_info[len('Referral:\n'):]
if not referral_uri.startswith('ldap:///'):
logger.debug('LDAP referral URI does not start with ldap:///')
return None
referral_dn = referral_uri[len('ldap:///'):]
return referral_dn
def _ldap_user_search_with_rdn(self, conn, username_or_email, user_search_dn):
query = u'(|({0}={2})({1}={2}))'.format(self._uid_attr, self._email_attr,
username_or_email)
logger.debug('Conducting user search: %s under %s', query, user_search_dn)
try:
return (conn.search_s(user_search_dn, ldap.SCOPE_SUBTREE, query.encode('utf-8')), None)
except ldap.REFERRAL as re:
referral_dn = self._get_ldap_referral_dn(re)
if not referral_dn:
return (None, 'Failed to follow referral when looking up username')
try:
subquery = u'(%s=%s)' % (self._uid_attr, username_or_email)
return (conn.search_s(referral_dn, ldap.SCOPE_BASE, subquery), None)
except ldap.LDAPError:
logger.debug('LDAP referral search exception')
return (None, 'Username not found')
except ldap.LDAPError:
logger.debug('LDAP search exception')
return (None, 'Username not found')
def _ldap_user_search(self, username_or_email, limit=20):
if not username_or_email:
return (None, 'Empty username/email')
# Verify the admin connection works first. We do this here to avoid wrapping
# the entire block in the INVALID CREDENTIALS check.
try:
with self._ldap.get_connection():
pass
except ldap.INVALID_CREDENTIALS:
return (None, 'LDAP Admin dn or password is invalid')
with self._ldap.get_connection() as conn:
logger.debug('Incoming username or email param: %s', username_or_email.__repr__())
for user_search_dn in self._user_dns:
(pairs, err_msg) = self._ldap_user_search_with_rdn(conn, username_or_email, user_search_dn)
if pairs is not None and len(pairs) > 0:
break
if err_msg is not None:
return (None, err_msg)
logger.debug('Found matching pairs: %s', pairs)
results = [LDAPUsers._LDAPResult(*pair) for pair in take(limit, pairs)]
# Filter out pairs without DNs. Some LDAP impls will return such pairs.
with_dns = [result for result in results if result.dn]
return (with_dns, None)
def _ldap_single_user_search(self, username_or_email):
with_dns, err_msg = self._ldap_user_search(username_or_email)
if err_msg is not None:
return (None, err_msg)
# Make sure we have at least one result.
if len(with_dns) < 1:
return (None, 'Username not found')
# If we have found a single pair, then return it.
if len(with_dns) == 1:
return (with_dns[0], None)
# Otherwise, there are multiple pairs with DNs, so find the one with the mail
# attribute (if any).
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 _build_user_information(self, response):
if not response.get(self._uid_attr):
return (None, 'Missing uid field "%s" in user record' % self._uid_attr)
if self._requires_email and not response.get(self._email_attr):
return (None, 'Missing mail field "%s" in user record' % self._email_attr)
username = response[self._uid_attr][0].decode('utf-8')
email = response.get(self._email_attr, [None])[0]
return (UserInformation(username=username, email=email, id=username), None)
def get_user(self, username_or_email):
""" Looks up a username or email in LDAP. """
logger.debug('Looking up LDAP username or email %s', username_or_email)
(found_user, err_msg) = self._ldap_single_user_search(username_or_email)
if err_msg is not None:
return (None, err_msg)
logger.debug('Found user for LDAP username or email %s', username_or_email)
_, found_response = found_user
return self._build_user_information(found_response)
def query_users(self, query, limit=20):
""" Queries LDAP for matching users. """
if not query:
return (None, self.federated_service, 'Empty query')
logger.debug('Got query %s with limit %s', query, limit)
(results, err_msg) = self._ldap_user_search(query + '*', limit=limit)
if err_msg is not None:
return (None, self.federated_service, err_msg)
final_results = []
for result in results[0:limit]:
credentials, err_msg = self._build_user_information(result.attrs)
if err_msg is not None:
continue
final_results.append(credentials)
logger.debug('For query %s found results %s', query, final_results)
return (final_results, self.federated_service, None)
def verify_credentials(self, username_or_email, password):
""" Verify the credentials with LDAP. """
# Make sure that even if the server supports anonymous binds, we don't allow it
if not password:
return (None, 'Anonymous binding not allowed')
(found_user, err_msg) = self._ldap_single_user_search(username_or_email)
if found_user is None:
return (None, err_msg)
found_dn, found_response = found_user
logger.debug('Found user for LDAP username %s; validating password', username_or_email)
logger.debug('DN %s found: %s', found_dn, found_response)
# First validate the password by binding as the user
try:
with LDAPConnection(self._ldap_uri, found_dn, password.encode('utf-8'),
self._allow_tls_fallback):
pass
except ldap.REFERRAL as re:
referral_dn = self._get_ldap_referral_dn(re)
if not referral_dn:
return (None, 'Invalid username')
try:
with LDAPConnection(self._ldap_uri, referral_dn, password.encode('utf-8'),
self._allow_tls_fallback):
pass
except ldap.INVALID_CREDENTIALS:
logger.debug('Invalid LDAP credentials')
return (None, 'Invalid password')
except ldap.INVALID_CREDENTIALS:
logger.debug('Invalid LDAP credentials')
return (None, 'Invalid password')
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)
attributes = [self._uid_attr, self._email_attr]
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, attrlist=attributes)
else:
msgid = conn.search_ext(user_search_dn, ldap.SCOPE_SUBTREE, search_flt, serverctrls=[lc],
attrlist=attributes)
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