Implement LDAP backend

README.md

@@ -1,6 +1,37 @@
 # LDAP Authentication Backend for StackStorm Enterprise Edition
 
-Requirements:
+### Requirements
 ```
-sudo apt-get install -y python-dev libldap2-dev libsasl2-dev libssl-dev
+sudo apt-get install -y python-dev libldap2-dev libsasl2-dev libssl-dev ldap-utils
+```
+
+### Configuration Options
+
+| option   | required | default | description                               |
+|----------|----------|---------|-------------------------------------------|
+| users_ou | yes      |         | OU of the user accounts                   |
+| host     | yes      |         | Hostname of the LDAP server               |
+| port     | yes      |         | Port of the LDAP server                   |
+| use_ssl  | no       | false   | Use LDAPS to connect                      |
+| use_tls  | no       | false   | Start TLS on LDAP to connect              |
+| cacert   | no       | None    | CA cert to validate certificate           |
+| id_attr  | no       | uid     | Field name of the user ID attribute       |
+| scope    | no       | subtree | Search scope (base, onelevel, or subtree) |
+
+### Configuration Example
+
+Please refer to the [standalone mode](http://docs.stackstorm.com/config/authentication.html#setup-standalone-mode) in the configuration section for authentication for basic setup concept. The following is an example of the auth section in the StackStorm configuration file for the LDAP backend.
+
+```
+[auth]
+mode = standalone
+backend = ldap
+backend_kwargs = {"users_ou": "ou=users,dc=example,dc=com", "host": "identity.example.com", "port": 636, "use_ssl": true, "cacert": "/path/to/cacert.pem"}
+enable = True
+debug = False
+use_ssl = True
+cert = /path/to/mycert.crt
+key = /path/to/mycert.key
+logging = /path/to/st2auth.logging.conf
+api_url = http://myhost.example.com:9101/
 ```

st2auth_enterprise_ldap_backend/ldap_backend.py

@@ -17,16 +17,105 @@
 
 from __future__ import absolute_import
 
+import os
 import logging
 
 import ldap
+import ldapurl
 
 __all__ = [
     'LDAPAuthenticationBackend'
 ]
 
 LOG = logging.getLogger(__name__)
 
+SEARCH_SCOPES = {
+    'base': ldapurl.LDAP_SCOPE_BASE,
+    'onelevel': ldapurl.LDAP_SCOPE_ONELEVEL,
+    'subtree': ldapurl.LDAP_SCOPE_SUBTREE
+}
+
 
 class LDAPAuthenticationBackend(object):
-    pass
+
+    def __init__(self, users_ou, host, port=389, scope='subtree',
+                 id_attr='uid', use_ssl=False, use_tls=False, cacert=None):
+
+        if not host:
+            raise ValueError('Hostname for the LDAP server is not provided.')
+
+        self._host = host
+
+        if port:
+            self._port = port
+        elif not port and not use_ssl:
+            LOG.warn('Default port 389 is used for the LDAP query.')
+            self._port = 389
+        elif not port and use_ssl:
+            LOG.warn('Default port 636 is used for the LDAP query over SSL.')
+            self._port = 636
+
+        if not users_ou:
+            raise ValueError('Users OU for the LDAP query is not provided.')
+
+        self._users_ou = users_ou
+
+        if scope not in SEARCH_SCOPES.keys():
+            raise ValueError('Scope value for the LDAP query must be one of '
+                             '%s.' % str(SEARCH_SCOPES.keys()))
+
+        self._scope = SEARCH_SCOPES[scope]
+
+        if not id_attr:
+            LOG.warn('Default to "uid" for the user attribute in the LDAP query.')
+
+        self._id_attr = id_attr or 'uid'
+
+        if use_ssl and use_tls:
+            raise ValueError('SSL and TLS cannot be both true.')
+
+        self._use_ssl = use_ssl
+        self._use_tls = use_tls
+
+        if cacert and not os.path.isfile(cacert): 
+            raise ValueError('Unable to find the cacert file "%s" for the LDAP connection.' % cacert)
+
+        self._cacert = cacert
+
+    def authenticate(self, username, password):
+        try:
+            # Use CA cert bundle to validate certificate if present.
+            if self._use_ssl or self._use_tls:
+                if self._cacert:
+                    ldap.set_option(ldap.OPT_X_TLS_CACERTFILE, self._cacert)
+                else:
+                    ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)
+
+            # Setup connection and options.
+            protocol = 'ldaps' if self._use_ssl else 'ldap'
+            endpoint = '%s://%s:%d' % (protocol, self._host, self._port)
+            connection = ldap.initialize(endpoint)
+            connection.set_option(ldap.OPT_DEBUG_LEVEL, 255)
+            connection.set_option(ldap.OPT_REFERRALS, 0)
+            connection.set_option(ldap.OPT_PROTOCOL_VERSION, ldap.VERSION3)
+
+            if self._use_tls:
+                connection.start_tls_s()
+
+            try:
+                # Bind using given username and password.
+                user_dn = '%s=%s,%s' % (self._id_attr, username, self._users_ou)
+                connection.simple_bind_s(user_dn, password)
+                LOG.info('Successfully authenticated user "%s".' % username)
+                return True
+            except Exception as e:
+                LOG.exception('Failed authenticating user "%s".' % username)
+                return False
+            finally:
+                connection.unbind_s()
+        except ldap.LDAPError as e:
+            LOG.exception('Unexpected LDAP configuration or connection error.')
+            return False
+
+    def get_user(self, username):
+        pass

tests/unit/test_authentication.py

@@ -0,0 +1,186 @@
+# Licensed to the StackStorm, Inc ('StackStorm') under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import ldap
+import mock
+import unittest2
+
+from st2auth_enterprise_ldap_backend import ldap_backend
+
+
+LDAP_HOST = 'ldap202.uswest2.stackstorm.net'
+LDAPS_PORT = 636
+LDAP_USERS_OU = 'ou=users,dc=staging,dc=stackstorm,dc=net'
+LDAP_CACERT = '/home/wcchan/Downloads/cacert.pem'
+LDAP_USER_UID = 'jon'
+LDAP_USER_PASSWD = 'snow123'
+LDAP_USER_BAD_PASSWD = 'snow1234'
+
+
+class LDAPBackendAuthenticationTest(unittest2.TestCase):
+
+    def test_null_host(self):
+        self.assertRaises(ValueError, ldap_backend.LDAPAuthenticationBackend,
+                          users_ou=LDAP_USERS_OU, host=None)
+
+    def test_null_users_ou(self):
+        self.assertRaises(ValueError, ldap_backend.LDAPAuthenticationBackend,
+                          users_ou=None, host=LDAP_HOST)
+
+    def test_null_port(self):
+        backend = ldap_backend.LDAPAuthenticationBackend(
+            users_ou=LDAP_USERS_OU, host=LDAP_HOST, port=None)
+
+        self.assertEqual(389, backend._port)
+
+        backend = ldap_backend.LDAPAuthenticationBackend(
+            users_ou=LDAP_USERS_OU, host=LDAP_HOST, port=None, use_ssl=True)
+
+        self.assertEqual(636, backend._port)
+
+        backend = ldap_backend.LDAPAuthenticationBackend(
+            users_ou=LDAP_USERS_OU, host=LDAP_HOST, port=9090, use_ssl=True)
+
+        self.assertEqual(9090, backend._port)
+
+    def test_scope(self):
+        for scope in ['base', 'onelevel', 'subtree']:
+            backend = ldap_backend.LDAPAuthenticationBackend(
+                users_ou=LDAP_USERS_OU, host=LDAP_HOST, scope=scope)
+
+            self.assertEqual(ldap_backend.SEARCH_SCOPES[scope], backend._scope)
+
+    def test_bad_scope(self):
+        self.assertRaises(ValueError, ldap_backend.LDAPAuthenticationBackend,
+                          users_ou=LDAP_USERS_OU, host=LDAP_HOST, scope='foo')
+
+    def test_null_id_attr(self):
+        backend = ldap_backend.LDAPAuthenticationBackend(
+            users_ou=LDAP_USERS_OU, host=LDAP_HOST, id_attr=None)
+
+        self.assertEqual('uid', backend._id_attr)
+
+    def test_both_ssl_tls_true(self):
+        self.assertRaises(ValueError, ldap_backend.LDAPAuthenticationBackend,
+                          users_ou=LDAP_USERS_OU, host=LDAP_HOST,
+                          use_ssl=True, use_tls=True)
+
+    def test_bad_cacert_file(self):
+        self.assertRaises(ValueError, ldap_backend.LDAPAuthenticationBackend,
+                          users_ou=LDAP_USERS_OU, host=LDAP_HOST,
+                          cacert='/tmp/foobar')
+
+    @mock.patch.object(
+        ldap, 'initialize',
+        mock.MagicMock(side_effect=ldap.LDAPError()))
+    def test_connection_error(self):
+        backend = ldap_backend.LDAPAuthenticationBackend(
+            users_ou=LDAP_USERS_OU, host=LDAP_HOST)
+
+        authenticated = backend.authenticate(LDAP_USER_UID, LDAP_USER_PASSWD)
+        self.assertFalse(authenticated)
+
+    @mock.patch.object(
+        ldap.ldapobject.SimpleLDAPObject, 'simple_bind_s',
+        mock.MagicMock(return_value=None))
+    def test_authenticate(self):
+        backend = ldap_backend.LDAPAuthenticationBackend(
+            users_ou=LDAP_USERS_OU, host=LDAP_HOST)
+
+        authenticated = backend.authenticate(LDAP_USER_UID, LDAP_USER_PASSWD)
+        self.assertTrue(authenticated)
+
+    @mock.patch.object(
+        ldap.ldapobject.SimpleLDAPObject, 'simple_bind_s',
+        mock.MagicMock(side_effect=Exception()))
+    def test_authenticate_failure(self):
+        backend = ldap_backend.LDAPAuthenticationBackend(
+            users_ou=LDAP_USERS_OU, host=LDAP_HOST)
+
+        authenticated = backend.authenticate(LDAP_USER_UID, LDAP_USER_BAD_PASSWD)
+        self.assertFalse(authenticated)
+
+    @mock.patch.object(
+        ldap.ldapobject.SimpleLDAPObject, 'simple_bind_s',
+        mock.MagicMock(return_value=None))
+    def test_ssl_authenticate(self):
+        backend = ldap_backend.LDAPAuthenticationBackend(
+            users_ou=LDAP_USERS_OU, host=LDAP_HOST,
+            port=LDAPS_PORT, use_ssl=True)
+
+        authenticated = backend.authenticate(LDAP_USER_UID, LDAP_USER_PASSWD)
+        self.assertTrue(authenticated)
+
+    @mock.patch.object(
+        ldap.ldapobject.SimpleLDAPObject, 'simple_bind_s',
+        mock.MagicMock(side_effect=Exception()))
+    def test_ssl_authenticate_failure(self):
+        backend = ldap_backend.LDAPAuthenticationBackend(
+            users_ou=LDAP_USERS_OU, host=LDAP_HOST,
+            port=LDAPS_PORT, use_ssl=True)
+
+        authenticated = backend.authenticate(LDAP_USER_UID, LDAP_USER_BAD_PASSWD)
+        self.assertFalse(authenticated)
+
+    @mock.patch.object(
+        ldap.ldapobject.SimpleLDAPObject, 'simple_bind_s',
+        mock.MagicMock(return_value=None))
+    def test_ssl_authenticate_validate_cert(self):
+        backend = ldap_backend.LDAPAuthenticationBackend(
+            users_ou=LDAP_USERS_OU, host=LDAP_HOST,
+            port=LDAPS_PORT, use_ssl=True, cacert=LDAP_CACERT)
+
+        authenticated = backend.authenticate(LDAP_USER_UID, LDAP_USER_PASSWD)
+        self.assertTrue(authenticated)
+
+    @mock.patch.object(
+        ldap.ldapobject.SimpleLDAPObject, 'start_tls_s',
+        mock.MagicMock(return_value=None))
+    @mock.patch.object(
+        ldap.ldapobject.SimpleLDAPObject, 'simple_bind_s',
+        mock.MagicMock(return_value=None))
+    def test_tls_authenticate(self):
+        backend = ldap_backend.LDAPAuthenticationBackend(
+            users_ou=LDAP_USERS_OU, host=LDAP_HOST, use_tls=True)
+
+        authenticated = backend.authenticate(LDAP_USER_UID, LDAP_USER_PASSWD)
+        self.assertTrue(authenticated)
+
+    @mock.patch.object(
+        ldap.ldapobject.SimpleLDAPObject, 'start_tls_s',
+        mock.MagicMock(return_value=None))
+    @mock.patch.object(
+        ldap.ldapobject.SimpleLDAPObject, 'simple_bind_s',
+        mock.MagicMock(side_effect=Exception()))
+    def test_tls_authenticate_failure(self):
+        backend = ldap_backend.LDAPAuthenticationBackend(
+            users_ou=LDAP_USERS_OU, host=LDAP_HOST, use_tls=True)
+
+        authenticated = backend.authenticate(LDAP_USER_UID, LDAP_USER_BAD_PASSWD)
+        self.assertFalse(authenticated)
+
+    @mock.patch.object(
+        ldap.ldapobject.SimpleLDAPObject, 'start_tls_s',
+        mock.MagicMock(return_value=None))
+    @mock.patch.object(
+        ldap.ldapobject.SimpleLDAPObject, 'simple_bind_s',
+        mock.MagicMock(return_value=None))
+    def test_tls_authenticate_validate_cert(self):
+        backend = ldap_backend.LDAPAuthenticationBackend(
+            users_ou=LDAP_USERS_OU, host=LDAP_HOST,
+            use_tls=True, cacert=LDAP_CACERT)
+
+        authenticated = backend.authenticate(LDAP_USER_UID, LDAP_USER_PASSWD)
+        self.assertTrue(authenticated)