pkgs/webapps/mediagoblin/ldap_fix.py

   1 # GNU MediaGoblin -- federated, autonomous media hosting
   2 # Copyright (C) 2011, 2012 MediaGoblin contributors.  See AUTHORS.
   3 #
   4 # This program is free software: you can redistribute it and/or modify
   5 # it under the terms of the GNU Affero General Public License as published by
   6 # the Free Software Foundation, either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU Affero General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16 from ldap3 import Server, Connection, SUBTREE
  17 from ldap3.core.exceptions import LDAPException
  18 import logging
  19
  20 import six
  21
  22 from mediagoblin.tools import pluginapi
  23
  24 _log = logging.getLogger(__name__)
  25
  26
  27 class LDAP(object):
  28     def __init__(self):
  29         self.ldap_settings = pluginapi.get_config('mediagoblin.plugins.ldap')
  30
  31     def _connect(self, server):
  32         _log.info('Connecting to {0}.'.format(server['LDAP_SERVER_URI']))
  33         self.server = Server(server['LDAP_SERVER_URI'])
  34
  35         if 'LDAP_START_TLS' in server and server['LDAP_START_TLS'] == 'true':
  36             _log.info('Initiating TLS')
  37             self.server.start_tls()
  38
  39     def _manager_auth(self, settings, username, password):
  40         conn = Connection(self.server,
  41                 settings['LDAP_BIND_DN'],
  42                 settings['LDAP_BIND_PW'],
  43                 auto_bind=True)
  44         found = conn.search(
  45                 search_base=settings['LDAP_SEARCH_BASE'],
  46                 search_filter=settings['LDAP_SEARCH_FILTER'].format(username=username),
  47                 search_scope=SUBTREE,
  48                 attributes=[settings['EMAIL_SEARCH_FIELD']])
  49         if (not found) or len(conn.entries) > 1:
  50             return False, None
  51
  52         user = conn.entries[0]
  53         user_dn = user.entry_dn
  54         try:
  55             email = user.entry_attributes_as_dict[settings['EMAIL_SEARCH_FIELD']][0]
  56         except KeyError:
  57             email = None
  58
  59         Connection(self.server, user_dn, password, auto_bind=True)
  60
  61         return username, email
  62
  63     def _direct_auth(self, settings, username, password):
  64         user_dn = settings['LDAP_USER_DN_TEMPLATE'].format(username=username)
  65         conn = Connection(self.server, user_dn, password, auto_bind=True)
  66         email_found = conn.search(
  67                 search_base=settings['LDAP_SEARCH_BASE'],
  68                 search_filter='uid={0}'.format(username),
  69                 search_scope=SUBTREE,
  70                 attributes=[settings['EMAIL_SEARCH_FIELD']])
  71
  72         if email_found:
  73             try:
  74                 email = conn.entries[0].entry_attributes_as_dict[settings['EMAIL_SEARCH_FIELD']][0]
  75             except KeyError:
  76                 email = None
  77
  78         return username, email
  79
  80     def login(self, username, password):
  81         for k, v in six.iteritems(self.ldap_settings):
  82             try:
  83                 self._connect(v)
  84
  85                 if 'LDAP_BIND_DN' in v:
  86                     return self._manager_auth(v, username, password)
  87                 else:
  88                     return self._direct_auth(v, username, password)
  89
  90             except LDAPException as e:
  91                 _log.info(e)
  92
  93         return False, None