quay/test/test_ldap.py

import unittest

from app import app
from initdb import setup_database_for_testing, finished_database_for_testing
from data.users import LDAPUsers
from data import model
from mockldap import MockLdap
from mock import patch
from contextlib import contextmanager

def _create_ldap(requires_email=True):
  base_dn = ['dc=quay', 'dc=io']
  admin_dn = 'uid=testy,ou=employees,dc=quay,dc=io'
  admin_passwd = 'password'
  user_rdn = ['ou=employees']
  uid_attr = 'uid'
  email_attr = 'mail'
  secondary_user_rdns = ['ou=otheremployees']

  ldap = LDAPUsers('ldap://localhost', base_dn, admin_dn, admin_passwd, user_rdn,
                   uid_attr, email_attr, secondary_user_rdns=secondary_user_rdns,
                   requires_email=requires_email, force_no_pagination=True)
  return ldap

@contextmanager
def mock_ldap(requires_email=True):
  mock_data = {
    'dc=quay,dc=io': {'dc': ['quay', 'io']},
    'ou=employees,dc=quay,dc=io': {
      'dc': ['quay', 'io'],
      'ou': 'employees'
    },
    'ou=otheremployees,dc=quay,dc=io': {
      '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'],
      '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'],
      'memberOf': ['cn=AwesomeFolk,dc=quay,dc=io'],
    },
    'uid=nomail,ou=employees,dc=quay,dc=io': {
      'dc': ['quay', 'io'],
      'ou': 'employees',
      'uid': ['nomail'],
      'userPassword': ['somepass']
    },
    'uid=cool.user,ou=employees,dc=quay,dc=io': {
      'dc': ['quay', 'io'],
      'ou': 'employees',
      'uid': ['cool.user', 'referred'],
      'userPassword': ['somepass'],
      'mail': ['foo@bar.com']
    },
    'uid=referred,ou=employees,dc=quay,dc=io': {
      'uid': ['referred'],
      '_referral': 'ldap:///uid=cool.user,ou=employees,dc=quay,dc=io'
    },
    'uid=invalidreferred,ou=employees,dc=quay,dc=io': {
      'uid': ['invalidreferred'],
      '_referral': 'ldap:///uid=someinvaliduser,ou=employees,dc=quay,dc=io'
    },
    'uid=multientry,ou=subgroup1,ou=employees,dc=quay,dc=io': {
      'uid': ['multientry'],
      'mail': ['foo@bar.com'],
      'userPassword': ['somepass'],
    },
    'uid=multientry,ou=subgroup2,ou=employees,dc=quay,dc=io': {
      'uid': ['multientry'],
      'another': ['key']
    },
    'uid=secondaryuser,ou=otheremployees,dc=quay,dc=io': {
      'dc': ['quay', 'io'],
      'ou': 'otheremployees',
      'uid': ['secondaryuser'],
      'userPassword': ['somepass'],
      'mail': ['foosecondary@bar.com']
    },
  }

  if not requires_email:
    for path in mock_data:
      mock_data[path].pop('mail', None)

  mockldap = MockLdap(mock_data)

  def initializer(uri, trace_level=0):
    obj = mockldap[uri]

    # Seed to "support" wildcard queries, which MockLDAP does not support natively.
    cool_block = {
      'dc': ['quay', 'io'],
      'ou': 'employees',
      'uid': ['cool.user', 'referred'],
      'userPassword': ['somepass'],
      'mail': ['foo@bar.com']
    }

    if not requires_email:
      cool_block.pop('mail', None)

    obj.search_s.seed('ou=employees,dc=quay,dc=io', 2, '(|(uid=cool*)(mail=cool*))')([
      ('uid=cool.user,ou=employees,dc=quay,dc=io', cool_block)
    ])

    obj.search_s.seed('ou=otheremployees,dc=quay,dc=io', 2, '(|(uid=cool*)(mail=cool*))')([])

    obj.search_s.seed('ou=employees,dc=quay,dc=io', 2, '(|(uid=unknown*)(mail=unknown*))')([])
    obj.search_s.seed('ou=otheremployees,dc=quay,dc=io', 2,
                      '(|(uid=unknown*)(mail=unknown*))')([])
    return obj

  mockldap.start()
  with patch('ldap.initialize', new=initializer):
    yield _create_ldap(requires_email=requires_email)
  mockldap.stop()


class TestLDAP(unittest.TestCase):
  def setUp(self):
    setup_database_for_testing(self)
    self.app = app.test_client()
    self.ctx = app.test_request_context()
    self.ctx.__enter__()

  def tearDown(self):
    finished_database_for_testing(self)
    self.ctx.__exit__(True, None, None)

  def test_invalid_admin_password(self):
    base_dn = ['dc=quay', 'dc=io']
    admin_dn = 'uid=testy,ou=employees,dc=quay,dc=io'
    admin_passwd = 'INVALIDPASSWORD'
    user_rdn = ['ou=employees']
    uid_attr = 'uid'
    email_attr = 'mail'

    with mock_ldap():
      ldap = LDAPUsers('ldap://localhost', base_dn, admin_dn, admin_passwd, user_rdn,
                       uid_attr, email_attr)

      # Try to login.
      (response, err_msg) = ldap.verify_and_link_user('someuser', 'somepass')
      self.assertIsNone(response)
      self.assertEquals('LDAP Admin dn or password is invalid', err_msg)

  def test_login(self):
    with mock_ldap() as ldap:
      # Verify we can login.
      (response, _) = ldap.verify_and_link_user('someuser', 'somepass')
      self.assertEquals(response.username, 'someuser')
      self.assertTrue(model.user.has_user_prompt(response, 'confirm_username'))

      # Verify we can confirm the user.
      (response, _) = ldap.confirm_existing_user('someuser', 'somepass')
      self.assertEquals(response.username, 'someuser')

  def test_login_secondary(self):
    with mock_ldap() as ldap:
      # Verify we can login.
      (response, _) = ldap.verify_and_link_user('secondaryuser', 'somepass')
      self.assertEquals(response.username, 'secondaryuser')

      # Verify we can confirm the user.
      (response, _) = ldap.confirm_existing_user('secondaryuser', 'somepass')
      self.assertEquals(response.username, 'secondaryuser')

  def test_invalid_password(self):
    with mock_ldap() as ldap:
      # Verify we cannot login with an invalid password.
      (response, err_msg) = ldap.verify_and_link_user('someuser', 'invalidpass')
      self.assertIsNone(response)
      self.assertEquals(err_msg, 'Invalid password')

      # Verify we cannot confirm the user.
      (response, err_msg) = ldap.confirm_existing_user('someuser', 'invalidpass')
      self.assertIsNone(response)
      self.assertEquals(err_msg, 'Invalid user')

  def test_missing_mail(self):
    with mock_ldap() as ldap:
      (response, err_msg) = ldap.get_user('nomail')
      self.assertIsNone(response)
      self.assertEquals('Missing mail field "mail" in user record', err_msg)

  def test_missing_mail_allowed(self):
    with mock_ldap(requires_email=False) as ldap:
      (response, _) = ldap.get_user('nomail')
      self.assertEquals(response.username, 'nomail')

  def test_confirm_different_username(self):
    with mock_ldap() as ldap:
      # Verify that the user is logged in and their username was adjusted.
      (response, _) = ldap.verify_and_link_user('cool.user', 'somepass')
      self.assertEquals(response.username, 'cool_user')

      # Verify we can confirm the user's quay username.
      (response, _) = ldap.confirm_existing_user('cool_user', 'somepass')
      self.assertEquals(response.username, 'cool_user')

      # Verify that we *cannot* confirm the LDAP username.
      (response, _) = ldap.confirm_existing_user('cool.user', 'somepass')
      self.assertIsNone(response)

  def test_referral(self):
    with mock_ldap() as ldap:
      (response, _) = ldap.verify_and_link_user('referred', 'somepass')
      self.assertEquals(response.username, 'cool_user')

      # Verify we can confirm the user's quay username.
      (response, _) = ldap.confirm_existing_user('cool_user', 'somepass')
      self.assertEquals(response.username, 'cool_user')

  def test_invalid_referral(self):
    with mock_ldap() as ldap:
      (response, _) = ldap.verify_and_link_user('invalidreferred', 'somepass')
      self.assertIsNone(response)

  def test_multientry(self):
    with mock_ldap() as ldap:
      (response, _) = ldap.verify_and_link_user('multientry', 'somepass')
      self.assertEquals(response.username, 'multientry')

  def test_login_empty_userdn(self):
    with mock_ldap():
      base_dn = ['ou=employees', 'dc=quay', 'dc=io']
      admin_dn = 'uid=testy,ou=employees,dc=quay,dc=io'
      admin_passwd = 'password'
      user_rdn = []
      uid_attr = 'uid'
      email_attr = 'mail'
      secondary_user_rdns = ['ou=otheremployees']

      ldap = LDAPUsers('ldap://localhost', base_dn, admin_dn, admin_passwd, user_rdn,
                       uid_attr, email_attr, secondary_user_rdns=secondary_user_rdns)

      # Verify we can login.
      (response, _) = ldap.verify_and_link_user('someuser', 'somepass')
      self.assertEquals(response.username, 'someuser')

      # Verify we can confirm the user.
      (response, _) = ldap.confirm_existing_user('someuser', 'somepass')
      self.assertEquals(response.username, 'someuser')

  def test_link_user(self):
    with mock_ldap() as ldap:
      # Link someuser.
      user, error_message = ldap.link_user('someuser')
      self.assertIsNone(error_message)
      self.assertIsNotNone(user)
      self.assertEquals('someuser', user.username)

      # Link again. Should return the same user record.
      user_again, _ = ldap.link_user('someuser')
      self.assertEquals(user_again.id, user.id)

      # Confirm someuser.
      result, _ = ldap.confirm_existing_user('someuser', 'somepass')
      self.assertIsNotNone(result)
      self.assertEquals('someuser', result.username)
      self.assertTrue(model.user.has_user_prompt(user, 'confirm_username'))

  def test_query(self):
    with mock_ldap() as ldap:
      # Lookup cool.
      (response, federated_id, error_message) = ldap.query_users('cool')
      self.assertIsNone(error_message)
      self.assertEquals(1, len(response))
      self.assertEquals('ldap', federated_id)

      user_info = response[0]
      self.assertEquals("cool.user", user_info.username)
      self.assertEquals("foo@bar.com", user_info.email)

      # Lookup unknown.
      (response, federated_id, error_message) = ldap.query_users('unknown')
      self.assertIsNone(error_message)
      self.assertEquals(0, len(response))
      self.assertEquals('ldap', federated_id)

  def test_timeout(self):
    base_dn = ['dc=quay', 'dc=io']
    admin_dn = 'uid=testy,ou=employees,dc=quay,dc=io'
    admin_passwd = 'password'
    user_rdn = ['ou=employees']
    uid_attr = 'uid'
    email_attr = 'mail'
    secondary_user_rdns = ['ou=otheremployees']

    with self.assertRaisesRegexp(Exception, "Can't contact LDAP server"):
      ldap = LDAPUsers('ldap://localhost', base_dn, admin_dn, admin_passwd, user_rdn,
                       uid_attr, email_attr, secondary_user_rdns=secondary_user_rdns,
                       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)

  def test_check_group_lookup_args(self):
    with mock_ldap() as ldap:
      (result, err) = ldap.check_group_lookup_args({'group_dn': 'cn=invalid'},
                                                   disable_pagination=True)
      self.assertFalse(result)
      self.assertIsNotNone(err)

      (result, err) = ldap.check_group_lookup_args({'group_dn': 'cn=AwesomeFolk'},
                                                   disable_pagination=True)
      self.assertTrue(result)
      self.assertIsNone(err)

  def test_metadata(self):
    with mock_ldap() as ldap:
      assert 'base_dn' in ldap.service_metadata()

if __name__ == '__main__':
  unittest.main()