Add ldap example user sync and auth

README.md

@@ -16,7 +16,7 @@ Development setup
 
 ```bash
 python setup.py develop
-pip install -r dev_requirements.txt
+pip install -e .[dev]
 ```
 
 Setup mysql schema:

configs/config.yaml

@@ -22,6 +22,10 @@ session:
 auth:
   debug: True
   module: 'oncall.auth.modules.debug'
+#   module: 'oncall.auth.modules.ldap_example'
+#   ldap_url: 'ldaps://example.com'
+#   ldap_user_suffix: '@example.biz'
+#   ldap_cert_path: '/etc/ldap_cert.pem'
 notifier:
   skipsend: True
 healthcheck_path: /tmp/status
@@ -87,4 +91,17 @@ iris_plan_integration:
 #   oauth_access_token: 'foo'
 #
 # user_sync:
-#   module: 'oncall.user_sync.slack'
+#   module: 'oncall.user_sync.ldap_sync'
+# ldap_sync:
+#   url: 'ldaps://example.com'
+#   base: 'ou=Staff Users,dc=multiforest,dc=biz'
+#   user: 'CN=example,DC=multiforest,DC=biz'
+#   password: 'password'
+#   cert_path: '/etc/ldap_cert.pem'
+#   query: '(&(objectClass=userProxy)(employeeId=*))'
+#   attrs:
+#     username: 'sAMAccountName'
+#     full_name: 'displayName'
+#     mail: 'mail'
+#     mobile: 'mobile'
+#   image_url: 'https://image.example.com/api/%s/picture'

dev_requirements.txt

@@ -1,9 +0,0 @@
-pytest
-pytest-mock
-requests
-gunicorn
-flake8
-Sphinx==1.5.6
-sphinxcontrib-httpdomain
-sphinx_rtd_theme
-sphinx-autobuild

setup.py

@@ -32,6 +32,20 @@ setuptools.setup(
         'irisclient',
         'slackclient',
     ],
+    extras_require={
+        'ldap': ['python-ldap'],
+        'dev': [
+            'pytest',
+            'pytest-mock',
+            'requests',
+            'gunicorn',
+            'flake8',
+            'Sphinx==1.5.6',
+            'sphinxcontrib-httpdomain',
+            'sphinx_rtd_theme',
+            'sphinx-autobuild',
+        ],
+    },
     entry_points={
         'console_scripts': [
             'oncall-dev = oncall.bin.run_server:main',

src/oncall/auth/modules/ldap_example.py

@@ -0,0 +1,33 @@
+# Copyright (c) LinkedIn Corporation. All rights reserved. Licensed under the BSD-2 Clause license.
+# See LICENSE in the project root for license information.
+import ldap
+
+
+class Authenticator:
+    def __init__(self, config):
+        if config.get('debug'):
+            self.authenticate = self.debug_auth
+            return
+        self.ldap_url = config['ldap_url']
+        self.cert_path = config['ldap_cert_path']
+        self.user_suffix = config['ldap_user_suffix']
+        self.authenticate = self.ldap_auth
+
+    def ldap_auth(self, username, password):
+        ldap.set_option(ldap.OPT_X_TLS_CACERTFILE, self.cert_path)
+        connection = ldap.initialize(self.ldap_url)
+        connection.set_option(ldap.OPT_REFERRALS, 0)
+
+        try:
+            if password:
+                connection.simple_bind_s(username + self.user_suffix, password)
+            else:
+                return False
+        except ldap.INVALID_CREDENTIALS:
+            return False
+        except ldap.SERVER_DOWN:
+            return None
+        return True
+
+    def debug_auth(self, username, password):
+        return True

src/oncall/user_sync/ldap_sync.py

@@ -0,0 +1,309 @@
+from gevent import monkey, sleep, spawn
+monkey.patch_all()  # NOQA
+
+import sys
+import time
+import yaml
+import logging
+import ldap
+
+from oncall import metrics
+from ldap.controls import SimplePagedResultsControl
+from datetime import datetime
+from pytz import timezone
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.exc import SQLAlchemyError, IntegrityError
+from phonenumbers import format_number, parse, PhoneNumberFormat
+from phonenumbers.phonenumberutil import NumberParseException
+
+
+logger = logging.getLogger()
+formatter = logging.Formatter('%(asctime)s %(levelname)s %(name)s %(message)s')
+ch = logging.StreamHandler()
+ch.setLevel(logging.DEBUG)
+ch.setFormatter(formatter)
+logger.setLevel(logging.INFO)
+logger.addHandler(ch)
+
+stats = {
+  'ldap_found': 0,
+  'sql_errors': 0,
+  'users_added': 0,
+  'users_failed_to_add': 0,
+  'users_failed_to_update': 0,
+  'users_purged': 0,
+  'user_contacts_updated': 0,
+  'user_names_updated': 0,
+  'user_photos_updated': 0,
+  'users_reactivated': 0,
+  'users_failed_to_reactivate': 0,
+}
+
+LDAP_SETTINGS = {}
+
+
+def normalize_phone_number(num):
+    return format_number(parse(num.decode('utf-8'), 'US'), PhoneNumberFormat.INTERNATIONAL)
+
+
+def get_predefined_users(config):
+    users = {}
+
+    try:
+        config_users = config['sync_script']['preset_users']
+    except KeyError:
+        return {}
+
+    for user in config_users:
+        users[user['name']] = user
+        for key in ['sms', 'call']:
+            try:
+                users[user['name']][key] = normalize_phone_number(users[user['name']][key])
+            except (NumberParseException, KeyError, AttributeError):
+                users[user['name']][key] = None
+
+    return users
+
+
+def timestamp_to_human_str(timestamp, tz):
+    dt = datetime.fromtimestamp(timestamp, timezone(tz))
+    return ' '.join([dt.strftime('%Y-%m-%d %H:%M:%S'), tz])
+
+
+def prune_user(engine, username):
+    global stats
+    stats['users_purged'] += 1
+
+    try:
+        engine.execute('DELETE FROM `user` WHERE `name` = %s', username)
+        logger.info('Deleted inactive user %s', username)
+
+    # The user has messages or some other user data which should be preserved. Just mark as inactive.
+    except IntegrityError:
+        logger.info('Marking user %s inactive', username)
+        engine.execute('UPDATE `user` SET `active` = FALSE WHERE `name` = %s', username)
+
+    except SQLAlchemyError as e:
+        logger.error('Deleting user %s failed: %s', username, e)
+        stats['sql_errors'] += 1
+
+
+def fetch_ldap():
+    ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_ALLOW)
+    l = ldap.initialize(LDAP_SETTINGS['url'])
+    l.set_option(ldap.OPT_X_TLS_CACERTFILE, LDAP_SETTINGS['cert_path'])
+    l.simple_bind_s(LDAP_SETTINGS['user'], LDAP_SETTINGS['password'])
+
+    req_ctrl = SimplePagedResultsControl(True, size=1000, cookie='')
+
+    known_ldap_resp_ctrls = {
+        SimplePagedResultsControl.controlType: SimplePagedResultsControl,
+    }
+
+    base = LDAP_SETTINGS['base']
+    attrs = ['distinguishedName'] + LDAP_SETTINGS['attrs'].values()
+    query = LDAP_SETTINGS['query']
+
+    users = {}
+    dn_map = {}
+
+    while True:
+        msgid = l.search_ext(base, ldap.SCOPE_SUBTREE, query, attrs, serverctrls=[req_ctrl])
+        rtype, rdata, rmsgid, serverctrls = l.result3(msgid, resp_ctrl_classes=known_ldap_resp_ctrls)
+        logger.info('Loaded %d entries from ldap.' % len(rdata))
+        for dn, ldap_dict in rdata:
+            if LDAP_SETTINGS['attrs']['mail'] not in ldap_dict:
+                logger.error('ERROR: invalid ldap entry for dn: %s' % dn)
+                continue
+
+            username = ldap_dict['sAMAccountName'][0]
+
+            mobile = ldap_dict.get(LDAP_SETTINGS['attrs']['mobile'])
+            mail = ldap_dict.get(LDAP_SETTINGS['attrs']['mail'])
+            name = ldap_dict.get(LDAP_SETTINGS['attrs']['full_name'])[0]
+
+            if mobile:
+                try:
+                    mobile = normalize_phone_number(mobile[0])
+                except NumberParseException:
+                    mobile = None
+                except UnicodeEncodeError:
+                    mobile = None
+
+            if mail:
+                mail = mail[0]
+                slack = mail.split('@')[0]
+            else:
+                slack = None
+
+            contacts = {'call': mobile, 'sms': mobile, 'email': mail, 'slack': slack, 'name': name}
+            dn_map[dn] = username
+            users[username] = contacts
+
+        pctrls = [
+            c for c in serverctrls if c.controlType == SimplePagedResultsControl.controlType
+            ]
+
+        cookie = pctrls[0].cookie
+        if not cookie:
+            break
+        req_ctrl.cookie = cookie
+
+    return users
+
+
+def sync(config, engine):
+    Session = sessionmaker(bind=engine)
+    session = Session()
+    oncall_users = {}
+    users_query = '''SELECT `user`.`name` as `name`, `contact_mode`.`name` as `mode`, `user_contact`.`destination`,
+                            `user`.`full_name`, `user`.`photo_url`
+                     FROM `user`
+                     LEFT OUTER JOIN `user_contact` ON `user`.`id` = `user_contact`.`user_id`
+                     LEFT OUTER JOIN `contact_mode` ON `user_contact`.`mode_id` = `contact_mode`.`id`
+                     ORDER BY `user`.`name`'''
+    for row in engine.execute(users_query):
+        contacts = oncall_users.setdefault(row.name, {})
+        if row.mode is None or row.destination is None:
+            continue
+        contacts[row.mode] = row.destination
+        contacts['full_name'] = row.full_name
+        contacts['photo_url'] = row.photo_url
+
+    oncall_usernames = set(oncall_users)
+
+    # users from ldap and config file
+    ldap_users = fetch_ldap()
+    stats['ldap_found'] += len(ldap_users)
+    ldap_users.update(get_predefined_users(config))
+    ldap_usernames = set(ldap_users)
+
+    # set of ldap users not in oncall
+    users_to_insert = ldap_usernames - oncall_usernames
+    # set of existing oncall users that are in ldap
+    users_to_update = oncall_usernames & ldap_usernames
+    # set of users in oncall but not ldap, assumed to be inactive
+    inactive_users = oncall_usernames - ldap_usernames
+    # users who need to be deactivated
+    rows = engine.execute('SELECT name FROM user WHERE active = TRUE AND name IN %s', inactive_users)
+    users_to_purge = (user.name for user in rows)
+
+    # set of inactive oncall users who appear in ldap
+    rows = engine.execute('SELECT name FROM user WHERE active = FALSE AND name IN %s', ldap_usernames)
+    users_to_reactivate = (user.name for user in rows)
+
+    # get objects needed for insertion
+    modes = dict(list(session.execute('SELECT `name`, `id` FROM `contact_mode`')))
+
+    user_add_sql = 'INSERT INTO `user` (`name`, `full_name`, `photo_url`) VALUES (%s, %s, %s)'
+
+    # insert users that need to be
+    logger.debug('Users to insert:')
+    for username in users_to_insert:
+        logger.debug('Inserting %s' % username)
+        full_name = ldap_users[username].pop('name')
+        try:
+            user_id = engine.execute(user_add_sql, (username, full_name, LDAP_SETTINGS['image_url'] % username)).lastrowid
+        except SQLAlchemyError:
+            stats['users_failed_to_add'] += 1
+            stats['sql_errors'] += 1
+            logger.exception('Failed to add user %s' % username)
+            continue
+        stats['users_added'] += 1
+        for key, value in ldap_users[username].iteritems():
+            if value and key in modes:
+                logger.debug('\t%s -> %s' % (key, value))
+                user_contact_add_sql = 'INSERT INTO `user_contact` (`user_id`, `mode_id`, `destination`) VALUES (%s, %s, %s)'
+                engine.execute(user_contact_add_sql, (user_id, modes[key], value))
+
+    # update users that need to be
+    contact_update_sql = 'UPDATE user_contact SET destination = %s WHERE user_id = (SELECT id FROM user WHERE name = %s) AND mode_id = %s'
+    contact_insert_sql = 'INSERT INTO user_contact (user_id, mode_id, destination) VALUES ((SELECT id FROM user WHERE name = %s), %s, %s)'
+    contact_delete_sql = 'DELETE FROM user_contact WHERE user_id = (SELECT id FROM user WHERE name = %s) AND mode_id = %s'
+    name_update_sql = 'UPDATE user SET full_name = %s WHERE name = %s'
+    photo_update_sql = 'UPDATE user SET photo_url = %s WHERE name = %s'
+    logger.debug('Users to update:')
+    for username in users_to_update:
+        logger.debug(username)
+        try:
+            db_contacts = oncall_users[username]
+            ldap_contacts = ldap_users[username]
+            full_name = ldap_contacts.pop('name')
+            if full_name != db_contacts.get('full_name'):
+                engine.execute(name_update_sql, (full_name, username))
+                stats['user_names_updated'] += 1
+            if not db_contacts.get('photo_url'):
+                engine.execute(photo_update_sql, (LDAP_SETTINGS['image_url'] % username, username))
+                stats['user_photos_updated'] += 1
+            for mode in modes:
+                if mode in ldap_contacts and ldap_contacts[mode]:
+                    if mode in db_contacts:
+                        if ldap_contacts[mode] != db_contacts[mode]:
+                            logger.debug('\tupdating %s', mode)
+                            engine.execute(contact_update_sql, (ldap_contacts[mode], username, modes[mode]))
+                            stats['user_contacts_updated'] += 1
+                    else:
+                        logger.debug('\tadding %s', mode)
+                        engine.execute(contact_insert_sql, (username, modes[mode], ldap_contacts[mode]))
+                        stats['user_contacts_updated'] += 1
+                elif mode in db_contacts:
+                    logger.debug('\tdeleting %s', mode)
+                    engine.execute(contact_delete_sql, (username, modes[mode]))
+                    stats['user_contacts_updated'] += 1
+                else:
+                    logger.debug('\tmissing %s', mode)
+        except SQLAlchemyError:
+            stats['users_failed_to_update'] += 1
+            stats['sql_errors'] += 1
+            logger.exception('Failed to update user %s' % username)
+            continue
+
+    logger.debug('Users to mark as inactive:')
+    for username in users_to_purge:
+        prune_user(engine, username)
+
+    logger.debug('Users to reactivate:')
+    for username in users_to_reactivate:
+        logger.debug(username)
+        try:
+            engine.execute('UPDATE user SET active = TRUE WHERE name = %s', username)
+            stats['users_reactivated'] += 1
+        except SQLAlchemyError:
+            stats['users_failed_to_reactivate'] += 1
+            stats['sql_errors'] += 1
+            logger.exception('Failed to reactivate user %s', username)
+
+    session.commit()
+    session.close()
+
+
+def metrics_sender():
+    while True:
+        metrics.emit_metrics()
+        sleep(60)
+
+
+def main(config):
+    global LDAP_SETTINGS
+
+    LDAP_SETTINGS = config['ldap_sync']
+
+    metrics.init(config, 'oncall-ldap-user-sync', stats)
+    spawn(metrics_sender)
+    # Default sleep one hour
+    sleep_time = config.get('user_sync_sleep_time', 3600)
+    engine = create_engine(config['db']['conn']['str'] % config['db']['conn']['kwargs'],
+                           **config['db']['kwargs'])
+    while 1:
+        logger.info('Starting user sync loop at %s' % time.time())
+        sync(config, engine)
+        logger.info('Sleeping for %s seconds' % sleep_time)
+        sleep(sleep_time)
+
+
+if __name__ == '__main__':
+    config_path = sys.argv[1]
+    with open(config_path, 'r') as config_file:
+        config = yaml.load(config_file)
+    main(config)