quay/data/users/externalldap.py

import ldap
import logging
import os
import itertools

from collections import namedtuple
from data.users.federated import FederatedUsers, UserInformation

logger = logging.getLogger(__name__)


class LDAPConnectionBuilder(object):
  def __init__(self, ldap_uri, user_dn, user_pw, allow_tls_fallback=False):
    self._ldap_uri = ldap_uri
    self._user_dn = user_dn
    self._user_pw = user_pw
    self._allow_tls_fallback = allow_tls_fallback

  def get_connection(self):
    return LDAPConnection(self._ldap_uri, self._user_dn, self._user_pw, self._allow_tls_fallback)


class LDAPConnection(object):
  def __init__(self, ldap_uri, user_dn, user_pw, allow_tls_fallback=False):
    self._ldap_uri = ldap_uri
    self._user_dn = user_dn
    self._user_pw = user_pw
    self._allow_tls_fallback = allow_tls_fallback
    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)

    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()

def _take(n, iterable):
  "Return first n items of the iterable as a list"
  return list(itertools.islice(iterable, n))


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):
    super(LDAPUsers, self).__init__('ldap', requires_email)
    self._ldap = LDAPConnectionBuilder(ldap_uri, admin_dn, admin_passwd, allow_tls_fallback)
    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]

  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 _credential_for_user(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._credential_for_user(found_response)

  def query_users(self, query, limit=20):
    """ Queries LDAP for matching users. """
    if not query:
      return (None, '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, err_msg)

    final_results = []
    for result in results[0:limit]:
      credentials, err_msg = self._credential_for_user(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, 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._credential_for_user(found_response)