Setting up LDAP support

-----------------------

RhodeCode starting from version 1.1 supports ldap authentication. In order

    easy_install python-ldap

    pip install python-ldap

.. note::

   python-ldap requires some certain libs on your system, so before installing 

   it check that you have at least `openldap`, and `sasl` libraries.

 Connection settings

 Enable LDAP          = checked

 Host                 = host.example.org

 Port                 = 389

 Account              = <account>

 Password             = <password>

 Certificate Checks   = DEMAND

 Search settings

 Base DN              = CN=users,DC=host,DC=example,DC=org

 LDAP Filter          = (&(objectClass=user)(!(objectClass=computer)))

 LDAP Search Scope    = SUBTREE

Password : optional

    Only required if the LDAP server does not allow anonymous browsing of

    records.

.. _Enable LDAPS:

.. _Certificate Checks:

Certificate Checks : optional

    How SSL certificates verification is handled - this is only useful when

    `Enable LDAPS`_ is enabled.  Only DEMAND or HARD offer full SSL security while

from formencode import htmlfill

from pylons import request, response, session, tmpl_context as c, url

from pylons.controllers.util import abort, redirect

from pylons.i18n.translation import _

from rhodecode.lib.base import BaseController, render

from rhodecode.lib import helpers as h

from rhodecode.lib.auth import LoginRequired, HasPermissionAllDecorator

from rhodecode.model.forms import LdapSettingsForm

log = logging.getLogger(__name__)

class LdapSettingsController(BaseController):

    def __before__(self):

        c.admin_user = session.get('admin_user')

        c.admin_username = session.get('admin_username')

        c.search_scope_choices = self.search_scope_choices

        c.tls_reqcert_choices = self.tls_reqcert_choices

        c.tls_kind_choices = self.tls_kind_choices

        super(LdapSettingsController, self).__before__()

    def index(self):

        c.search_scope_cur = defaults.get('ldap_search_scope')

        c.tls_reqcert_cur = defaults.get('ldap_tls_reqcert')

        c.tls_kind_cur = defaults.get('ldap_tls_kind')

        return htmlfill.render(

                    render('admin/ldap/ldap.html'),

                    encoding="UTF-8",

                    force_defaults=True,)

    def ldap_settings(self):

        """POST ldap create and store ldap settings"""

        _form = LdapSettingsForm([x[0] for x in self.tls_reqcert_choices],

                                 [x[0] for x in self.search_scope_choices],

                                 [x[0] for x in self.tls_kind_choices])()

        try:

            form_result = _form.to_python(dict(request.POST))

            try:

                for k, v in form_result.items():

                    if k.startswith('ldap_'):

                        setting.app_settings_value = v

                        self.sa.add(setting)

                self.sa.commit()

                h.flash(_('Ldap settings updated successfully'),

                    category='success')

                raise

        except LdapImportError:

            h.flash(_('Unable to activate ldap. The "python-ldap" library '

                      'is missing.'), category='warning')

        except formencode.Invalid, errors:

            return htmlfill.render(

                render('admin/ldap/ldap.html'),

                defaults=errors.value,

                prefix_error=False,

                encoding="UTF-8")

        except Exception:

            log.error(traceback.format_exc())

            h.flash(_('error occurred during update of ldap settings'),

                    category='error')

from rhodecode.lib.auth import LoginRequired, HasPermissionAllDecorator, \

    HasPermissionAnyDecorator, NotAnonymous

from rhodecode.lib.base import BaseController, render

from rhodecode.lib.celerylib import tasks, run_task

from rhodecode.lib.utils import repo2db_mapper, invalidate_cache, \

    set_rhodecode_config, repo_name_slug

from rhodecode.model.forms import UserForm, ApplicationSettingsForm, \

    ApplicationUiSettingsForm

from rhodecode.model.scm import ScmModel

from rhodecode.model.user import UserModel

log = logging.getLogger(__name__)

class SettingsController(BaseController):

    @HasPermissionAllDecorator('hg.admin')

    def index(self, format='html'):

        """GET /admin/settings: All items in the collection"""

        # url('admin_settings')

        defaults.update(self.get_hg_ui_settings())

        return htmlfill.render(

            render('admin/settings/settings.html'),

            defaults=defaults,

            encoding="UTF-8",

            force_defaults=False

            h.flash(_('Whoosh reindex task scheduled'), category='success')

        if setting_id == 'global':

            application_form = ApplicationSettingsForm()()

            try:

                form_result = application_form.to_python(dict(request.POST))

                try:

                    hgsettings1.app_settings_value = \

                        form_result['rhodecode_title']

                    hgsettings2.app_settings_value = \

                        form_result['rhodecode_realm']

                    hgsettings3.app_settings_value = \

                        form_result['rhodecode_ga_code']

                    self.sa.add(hgsettings1)

                    self.sa.add(hgsettings2)

                    self.sa.add(hgsettings3)

from rhodecode.lib.exceptions import LdapPasswordError, LdapUsernameError

from rhodecode.lib.utils import get_repo_slug

from rhodecode.lib.auth_ldap import AuthLdap

from rhodecode.model import meta

from rhodecode.model.user import UserModel

log = logging.getLogger(__name__)

class PasswordGenerator(object):

    """This is a simple class for generating password from

    firstly checks for db authentication then if ldap is enabled for ldap

    authentication, also creates ldap user if not in database

    :param username: username

    :param password: password

    """

    user_model = UserModel()

    user = user_model.get_by_username(username, cache=False)

    log.debug('Authenticating user using RhodeCode account')

    if user is not None and not user.ldap_dn:

        if user.active:

                                            case_insensitive=True)

        if user_obj is not None and not user_obj.ldap_dn:

            log.debug('this user already exists as non ldap')

            return False

        #======================================================================

        # FALLBACK TO LDAP AUTH IF ENABLE

        #======================================================================

        if str2bool(ldap_settings.get('ldap_active')):

            log.debug("Authenticating user using ldap")

            kwargs = {

            try:

                aldap = AuthLdap(**kwargs)

                (user_dn, ldap_attrs) = aldap.authenticate_ldap(username,

                                                                password)

                log.debug('Got ldap DN response %s', user_dn)

                user_attrs = {

                    }

                if user_model.create_ldap(username, password, user_dn,

                                          user_attrs):

                    log.info('created new ldap user %s', username)

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import logging

log = logging.getLogger(__name__)

try:

    import ldap

except ImportError:

    pass

class AuthLdap(object):

    def __init__(self, server, base_dn, port=389, bind_dn='', bind_pass='',

                 ldap_filter='(&(objectClass=user)(!(objectClass=computer)))',

                 search_scope='SUBTREE',

                 attr_login='uid'):

        self.ldap_version = ldap_version

        ldap_server_type = 'ldap'

        self.BASE_DN = base_dn

        self.LDAP_FILTER = ldap_filter

        self.SEARCH_SCOPE = ldap.__dict__['SCOPE_' + search_scope]

        self.attr_login = attr_login

    def authenticate_ldap(self, username, password):

        """Authenticate a user via LDAP and return his/her LDAP properties.

        Raises AuthenticationError if the credentials are rejected, or

        EnvironmentError if the LDAP server can't be reached.

            if self.TLS_KIND == 'START_TLS':

                server.start_tls_s()

            if self.LDAP_BIND_DN and self.LDAP_BIND_PASS:

                server.simple_bind_s(self.LDAP_BIND_DN, self.LDAP_BIND_PASS)

            log.debug("Authenticating %r filt %s at %s", self.BASE_DN,

                      filt, self.LDAP_SERVER)

            lobjects = server.search_ext_s(self.BASE_DN, self.SEARCH_SCOPE,

                                           filt)

            if not lobjects:

                raise ldap.NO_SUCH_OBJECT()

            for (dn, _attrs) in lobjects:

                try:

                    server.simple_bind_s(dn, password)

                    break

                except ldap.INVALID_CREDENTIALS, e:

                    log.debug("LDAP rejected password for user '%s' (%s): %s",

                              uid, username, dn)

                raise LdapPasswordError()

        except ldap.NO_SUCH_OBJECT, e:

            log.debug("LDAP says no such user '%s' (%s)", uid, username)

            raise LdapUsernameError()

        except ldap.SERVER_DOWN, e:

        return (dn, attrs)

from vcs.backends.base import BaseChangeset

from vcs.utils.lazy import LazyProperty

from rhodecode.model import meta

from rhodecode.model.caching_query import FromCache

from rhodecode.model.repo import RepoModel

from rhodecode.model.user import UserModel

log = logging.getLogger(__name__)

def set_rhodecode_config(config):

    """Updates pylons config with new settings from database

    :param config:

    """

    for k, v in hgsettings.items():

        config[k] = v

def invalidate_cache(cache_key, *args):

    def __repr__(self):

        return "<%s('%s:%s')>" % (self.__class__.__name__,

                                  self.app_settings_name, self.app_settings_value)

    @classmethod

    def get_app_settings(cls, cache=False):

        ret = Session.query(cls)

        if cache:

            ret = ret.options(FromCache("sql_cache_short", "get_hg_settings"))

    def get_ldap_settings(cls, cache=False):

        ret = Session.query(cls)\

                .filter(cls.app_settings_name.startswith('ldap_'))\

                .all()

        fd = {}

        for row in ret:

        return fd

class RhodeCodeUi(Base):

    __tablename__ = 'rhodecode_ui'

    __table_args__ = {'useexisting':True}

from rhodecode.tests import *

class TestLdapSettingsController(TestController):

    def test_index(self):

        # Test response...

from rhodecode.lib.auth import get_crypt_password, check_password

from rhodecode.tests import *

class TestAdminSettingsController(TestController):

    def test_index(self):

        response = self.app.get(url('admin_settings'))

        # Test response...

                                                 rhodecode_title=old_title,

                                                 rhodecode_realm=old_realm,

                                                 rhodecode_ga_code=new_ga_code

                                                 ))

        assert 'Updated application settings' in response.session['flash'][0][1], 'no flash message about success of change'

        response = response.follow()

        assert """_gaq.push(['_setAccount', '%s']);""" % new_ga_code in response.body

    def test_ga_code_inactive(self):

        self.log_user()

                                                 rhodecode_title=old_title,

                                                 rhodecode_realm=old_realm,

                                                 rhodecode_ga_code=new_ga_code

                                                 ))

        assert 'Updated application settings' in response.session['flash'][0][1], 'no flash message about success of change'

        response = response.follow()

        assert """_gaq.push(['_setAccount', '%s']);""" % new_ga_code not in response.body

    def test_title_change(self):

                                                 rhodecode_realm=old_realm,

                                                 rhodecode_ga_code=''

                                                 ))

        assert 'Updated application settings' in response.session['flash'][0][1], 'no flash message about success of change'

        response = response.follow()

        assert """<h1><a href="/">%s</a></h1>""" % new_title in response.body

    def test_my_account(self):