Merge pull request #31 from dustinnguyen/master

Added ldap auth & ldap mock test

Vagrantfile

@@ -76,6 +76,9 @@ Vagrant.configure(2) do |config|
     echo "CREATE USER orlo WITH PASSWORD 'password'; CREATE DATABASE orlo OWNER orlo; " \
         | sudo -u postgres -i psql
 
+    # python-ldap dependencies
+    sudo apt-get install -y python-dev libldap2-dev libsasl2-dev libssl-dev
+
     # Build tools
     sudo apt-get -y install build-essential git-buildpackage debhelper python-dev dh-systemd python-virtualenv
     wget -P /tmp/ \

etc/passwd

@@ -1 +1 @@
-testuser:pbkdf2:sha1:1000$gE6GSSIL$6c8d99caf178c27973655764fd7f00c29cdaa575
+testuser:pbkdf2:sha1:1000$Xi0lFFta$faade1a2a59667086bd7ff40a47852e7400b9713

orlo/config.py

@@ -16,12 +16,14 @@
 
 config.add_section('security')
 config.set('security', 'enabled', 'false')
-config.set('security', 'method', 'file')
 config.set('security', 'passwd_file', os.path.dirname(__file__) + '/../etc/passwd')
 config.set('security', 'secret_key', 'change_me')
 # NOTE: orlo.__init__ checks that secret_key is not "change_me" when security is enabled
 # Do not change the default here without updating __init__ as well.
 config.set('security', 'token_ttl', '3600')
+config.set('security', 'ldap_server', 'localhost.localdomain')
+config.set('security', 'ldap_port', '389')
+config.set('security', 'user_base_dn', 'ou=people,ou=example,o=test')
 
 config.add_section('db')
 config.set('db', 'uri', 'postgres://orlo:password@localhost:5432/orlo')

orlo/user_auth.py

@@ -8,6 +8,7 @@
 from itsdangerous import (TimedJSONWebSignatureSerializer
                           as Serializer, BadSignature, SignatureExpired)
 from functools import update_wrapper, wraps
+import ldap
 
 
 # initialization
@@ -43,31 +44,11 @@ def wrapped(*args, **kwargs):
 
 
 class User(object):
-    def __init__(self, name):
-        self.name = name
-        self.password_hash = self.get_pw_ent(name)
+    def __init__(self, username, password):
+        self.username = username
+        self.password_hash = self.hash_password(password)
         self.confirmed = False
 
-    @staticmethod
-    def get_pw_ent(name):
-        if config.get('security', 'method') == 'file':
-            with open(config.get('security', 'passwd_file')) as f:
-                for line in f:
-                    line = line.strip()
-                    user = line.split(':')[0]
-                    if not user == name:
-                        continue
-
-                    # found user return password
-                    app.logger.debug("Found user {} in file".format(name))
-                    pw = ':'.join(line.split(':')[1:])
-                    return pw
-            # user not in passwd file return a hash that cannot occur
-            return '*'
-
-        # TODO implement LDAP
-        raise OrloAuthError("Unknown user_auth method, check security config")
-
     def hash_password(self, password):
         self.password_hash = generate_password_hash(password)
 
@@ -76,63 +57,103 @@ def verify_password(self, password):
         return rc
 
     def generate_auth_token(self, expiration=3600):
-        s = Serializer(app.config['SECRET_KEY'], expires_in=expiration)
-        return s.dumps({'id': self.name})
+        secret_key = config.get('security', 'secret_key')
+        s = Serializer(secret_key, expires_in=expiration)
+        return s.dumps({'id': self.username})
 
     @staticmethod
     def verify_auth_token(token):
-        s = Serializer(app.config['SECRET_KEY'])
+        app.logger.debug("Verify auth token called")
+        s = Serializer(config.get('security', 'secret_key'))
         try:
             data = s.loads(token)
         except SignatureExpired:
             return None  # valid token, but expired
         except BadSignature:
             return None  # invalid token
-        name = data['id']
-        return User(name)
+        user = data['id']
+        return user
 
 
 @user_auth.verify_password
-def verify_password(username=None, password=None):
-    app.logger.debug("Verify_password called")
-
-    if not password:
-        return False
-
-    user = User(username)
-    if not user.verify_password(password):
-        return False
-
-    set_current_user_as(user)
-    return True
+def verify_password(username_or_token=None, password=None):
+    # first try to authenticate by token
+    user = User.verify_auth_token(username_or_token)
+    if user:
+        set_current_user_as(user)
+        return True
+    elif not user:
+        password_file_user = verify_password_file(username_or_token, password)
+        if password_file_user:
+            set_current_user_as(username_or_token)
+            return True
+        else:
+            ldap_user = verify_ldap_access(username_or_token, password)
+            if ldap_user:
+                set_current_user_as(username_or_token)
+                return True
+            else:
+                return False
 
 
 @token_auth.verify_token
 def verify_token(token=None):
-    app.logger.debug("Verify_token called")
-
+    app.logger.debug("Verify token called")
     if not token:
         return False
-
     token_user = token_manager.verify(token)
     if token_user:
-        set_current_user_as(User(token_user))
+        set_current_user_as(token_user)
         return True
 
 
 def set_current_user_as(user):
     if not g.get('current_user'):
-        app.logger.debug('Setting current user to: {}'.format(user.name))
+        app.logger.debug('Setting current user to: {}'.format(user))
         g.current_user = user
 
 
+def verify_ldap_access(username, password):
+    app.logger.debug("Verify ldap called")
+    try:
+        ldap_server = config.get('security', 'ldap_server')
+        ldap_port = config.get('security', 'ldap_port')
+        user_base_dn = config.get('security', 'user_base_dn')
+        l = ldap.initialize('ldap://{}:{}'.format(ldap_server, ldap_port))
+        ldap_user = "uid=%s,%s" % (username, user_base_dn)
+        l.protocol_version = ldap.VERSION3
+        l.simple_bind_s(ldap_user, password)
+        return True
+    except ldap.LDAPError:
+        return False
+
+
+def verify_password_file(username=None, password=None):
+    app.logger.debug("Verify password file called")
+    password_file = config.get('security', 'passwd_file')
+    with open(password_file) as f:
+        for line in f:
+            line = line.strip()
+            user = line.split(':')[0]
+            if not user == username:
+                continue
+            # found user return password
+            elif user == username:
+                app.logger.debug("Found user {} in file".format(username))
+                pw = ':'.join(line.split(':')[1:])
+                checked_password = check_password_hash(pw, password)
+                if checked_password:
+                    return True
+                else:
+                    return False
+
+
 @user_auth.error_handler
 @token_auth.error_handler
 def auth_error():
     """
     Authentication error
     """
-    # raise OrloAuthError("Not authorized")
     response = jsonify({'error': 'not authorized'})
     response.status_code = 401
     return response
@@ -145,9 +166,8 @@ def get_token():
     Get a token
     """
     ttl = config.getint('security', 'token_ttl')
-    token = token_manager.generate(g.current_user.name, ttl)
+    token = token_manager.generate(g.current_user, ttl)
     return jsonify({
         'token': token.decode('ascii'),
         'duration': config.get('security', 'token_ttl')
     })
-

requirements.txt

@@ -14,4 +14,5 @@ Werkzeug==0.9.4
 itsdangerous==0.23
 passlib==1.6.1
 psycopg2==2.6.1
+python-ldap>=2.4.25
 pytz

requirements_testing.txt

@@ -18,3 +18,7 @@ pytest
 pytz
 requests>=2.4.2
 sqlalchemy-utils==0.31.0
+mock==1.3.0
+mockldap==0.2.6
+python-ldap==2.4.25
+six==1.10.0

tests/test_auth.py

@@ -8,10 +8,12 @@
 from werkzeug.datastructures import Headers
 from werkzeug.test import Client
 import base64
+import ldap
+from mockldap import MockLdap
 
 __author__ = 'alforbes'
 
-USER = 'testuser'
+USERNAME = 'testuser'
 PASSWORD = 'blah'
 
 
@@ -31,11 +33,23 @@ def auth_required():
     response.status_code = 200
     return response
 
+@orlo.app.route('/test/user')
+@conditional_auth(user_auth.login_required)
+def get_resource():
+    return jsonify({'data': 'Hello, %s!' % g.current_user})
 
 class OrloAuthTest(TestCase):
     """
     Base test class to setup the app
     """
+    top = ('o=test', {'o': ['test']})
+    example = ('ou=example,o=test', {'ou': ['example']})
+    people = ('ou=people,ou=example,o=test', {'ou': ['other']})
+    ldapuser = ('uid=ldapuser,ou=people,ou=example,o=test', {'uid': ['ldapuser'], 'userPassword': ['ldapuserpw']})
+    # This is the content of our mock LDAP directory. It takes the form
+    # {dn: {attr: [value, ...], ...}, ...}.
+    directory = dict([top, example, people, ldapuser])
+
     def create_app(self):
         self.app = orlo.app
         self.app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite://'
@@ -46,20 +60,40 @@ def create_app(self):
 
         return self.app
 
+    @classmethod
+    def setUpClass(self):
+        # We only need to create the MockLdap instance once. The content we
+        # pass in will be used for all LDAP connections.
+        self.mockldap = MockLdap(self.directory)
+
+    @classmethod
+    def tearDownClass(self):
+        del self.mockldap
+
     def setUp(self):
         db.create_all()
+        self.mockldap.start()
+        self.ldapobj = self.mockldap['ldap://localhost/']
         self.orig_security_enabled = orlo.config.get('security', 'enabled')
-        self.orig_security_method = orlo.config.get('security', 'method')
+        self.orig_security_secret_key = orlo.config.set('security', 'secret_key')
+        self.orig_security_ldap_server = orlo.config.set('security', 'ldap_server')
+        self.orig_security_ldap_port = orlo.config.set('security', 'ldap_port')
+        self.orig_security_user_base_dn = orlo.config.set('security', 'user_base_dn')
         orlo.config.set('security', 'enabled', 'true')
-        orlo.config.set('security', 'method', 'file')
+        orlo.config.set('security', 'secret_key', 'It does not matter how slowly you go so long as you do not stop')
+        orlo.config.set('security', 'ldap_server', 'localhost')
+        orlo.config.set('security', 'ldap_port', '389')
+        orlo.config.set('security', 'user_base_dn', 'ou=people,ou=example,o=test')
 
     def tearDown(self):
         db.session.remove()
         db.drop_all()
+        self.mockldap.stop()
+        del self.ldapobj
         orlo.config.set('security', 'enabled', self.orig_security_enabled)
-        orlo.config.set('security', 'method', self.orig_security_method)
+        orlo.config.set('security', 'secret_key', self.orig_security_secret_key)
 
-    def get_with_basic_auth(self, path, username='testuser', password='blabla'):
+    def get_with_basic_auth(self, path, username='testuser', password='blah'):
         """
         Do a request with basic auth
 
@@ -74,6 +108,21 @@ def get_with_basic_auth(self, path, username='testuser', password='blabla'):
         response = Client.open(self.client, path=path, headers=h)
         return response
 
+    def get_with_ldap_auth(self, path, username='ldapuser', password='ldapuserpw'):
+        """
+        Do a request with ldap auth
+
+        :param path:
+        :param username:
+        :param password:
+        """
+        h = Headers()
+        h.add('Authorization', 'Basic ' + base64.b64encode(
+            '{u}:{p}'.format(u=username, p=password)
+        ))
+        response = Client.open(self.client, path=path, headers=h)
+        return response
+
     def get_with_token_auth(self, path, token):
         """
         Do a request with token auth
@@ -164,6 +213,12 @@ def test_with_login(self):
         response = self.get_with_basic_auth('/test/auth_required')
         self.assert200(response)
 
+    def test_with_ldap_login(self):
+        response = self.get_with_ldap_auth(
+            '/test/auth_required', username='ldapuser', password='ldapuserpw'
+        )
+        self.assert200(response)
+
     def test_with_bad_login(self):
         response = self.get_with_basic_auth(
             '/test/auth_required', username='bad_user', password='foobar'