~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

pluggable authentication controller for Kallithea

This file was forked by the Kallithea project in July 2014.

Original author and date, and relevant copyright and licensing information is below:

:created_on: Nov 26, 2010

:author: akesterson

"""

import logging

import traceback

import formencode.htmlfill

from tg import request

from tg import tmpl_context as c

from tg.i18n import ugettext as _

from webob.exc import HTTPFound

from kallithea.config.routing import url

from kallithea.lib import auth_modules

from kallithea.lib import helpers as h

from kallithea.lib.auth import HasPermissionAnyDecorator, LoginRequired

from kallithea.lib.base import BaseController, render

from kallithea.model.db import Setting

from kallithea.model.forms import AuthSettingsForm

from kallithea.model.meta import Session

log = logging.getLogger(__name__)

class AuthSettingsController(BaseController):

    @LoginRequired()

    @HasPermissionAnyDecorator('hg.admin')

    def _before(self, *args, **kwargs):

        super(AuthSettingsController, self)._before(*args, **kwargs)

    def __load_defaults(self):

        c.available_plugins = [

            'kallithea.lib.auth_modules.auth_internal',

            'kallithea.lib.auth_modules.auth_container',

            'kallithea.lib.auth_modules.auth_ldap',

            'kallithea.lib.auth_modules.auth_crowd',

            'kallithea.lib.auth_modules.auth_pam'

        ]

        self.enabled_plugins = auth_modules.get_auth_plugins()

    def __render(self, defaults, errors):

        c.defaults = {}

        c.plugin_settings = {}

        c.plugin_shortnames = {}

        for plugin in self.enabled_plugins:

            module = plugin.__class__.__module__

            c.plugin_shortnames[module] = plugin.name

            c.plugin_settings[module] = plugin.plugin_settings()

            for v in c.plugin_settings[module]:

                fullname = "auth_%s_%s" % (plugin.name, v["name"])

                if "default" in v:

                    c.defaults[fullname] = v["default"]

                # Current values will be the default on the form, if there are any

                setting = Setting.get_by_name(fullname)

                if setting is not None:

                    c.defaults[fullname] = setting.app_settings_value

        if defaults:

            c.defaults.update(defaults)

        # we want to show , separated list of enabled plugins

        c.defaults['auth_plugins'] = ','.join(c.enabled_plugin_names)

        return formencode.htmlfill.render(

            render('admin/auth/auth_settings.html'),

            defaults=c.defaults,

            errors=errors,

            prefix_error=False,

            encoding="UTF-8",

            force_defaults=False)

    def index(self):

        self.__load_defaults()

        return self.__render(defaults=None, errors=None)

    def auth_settings(self):

        """POST create and store auth settings"""

        self.__load_defaults()

        # First, parse only the plugin list (not the plugin settings).

        _auth_plugins_validator = AuthSettingsForm([]).fields['auth_plugins']

        try:

            new_enabled_plugins = _auth_plugins_validator.to_python(request.POST.get('auth_plugins'))

        except formencode.Invalid:

            # User provided an invalid plugin list. Just fall back to

            # the list of currently enabled plugins. (We'll re-validate

            # and show an error message to the user, below.)

            pass

        else:

            # Hide plugins that the user has asked to be disabled, but

            # do not show plugins that the user has asked to be enabled

            # (yet), since that'll cause validation errors and/or wrong

            # settings being applied (e.g. checkboxes being cleared),

            # since the plugin settings will not be in the POST data.

            c.enabled_plugin_names = [p for p in c.enabled_plugin_names if p in new_enabled_plugins]

        # Next, parse everything including plugin settings.

        _form = AuthSettingsForm(c.enabled_plugin_names)()

        try:

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

            for k, v in form_result.items():

# -*- coding: utf-8 -*-

# 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/>.

"""

Authentication modules

"""

import importlib

import logging

import traceback

from kallithea.lib.auth import AuthUser, PasswordGenerator

from kallithea.lib.utils2 import str2bool

from kallithea.model.db import Setting, User

from kallithea.model.meta import Session

from kallithea.model.user import UserModel

from kallithea.model.user_group import UserGroupModel

log = logging.getLogger(__name__)

class LazyFormencode(object):

    def __init__(self, formencode_obj, *args, **kwargs):

        self.formencode_obj = formencode_obj

        self.args = args

        self.kwargs = kwargs

    def __call__(self, *args, **kwargs):

        from inspect import isfunction

        formencode_obj = self.formencode_obj

        if isfunction(formencode_obj):

            # case we wrap validators into functions

            formencode_obj = self.formencode_obj(*args, **kwargs)

        return formencode_obj(*self.args, **self.kwargs)

def authenticate(username, password, environ=None):

    """

    Authentication function used for access control,

    It tries to authenticate based on enabled authentication modules.

    :param username: username can be empty for container auth

    :param password: password can be empty for container auth

    :param environ: environ headers passed for container auth

    :returns: None if auth failed, user_data dict if auth is correct

    """

    auth_plugins = get_auth_plugins()

    for plugin in auth_plugins:

        module = plugin.__class__.__module__

        log.debug('Trying authentication using %s', module)

        # load plugin settings from Kallithea database

        plugin_name = plugin.name

        plugin_settings = {}

        for v in plugin.plugin_settings():

            conf_key = "auth_%s_%s" % (plugin_name, v["name"])

            setting = Setting.get_by_name(conf_key)

            plugin_settings[v["name"]] = setting.app_settings_value if setting else None

        if not str2bool(plugin_settings["enabled"]):

            log.info("Authentication plugin %s is disabled, skipping for %s",

                     module, username)

            continue

        # use plugin's method of user extraction.

        user = plugin.get_user(username, environ=environ,

                               settings=plugin_settings)

        log.debug('Plugin %s extracted user `%s`', module, user)

        if user is not None and not user.active: # give up, way before creating AuthUser

            log.error("Rejecting authentication of in-active user %s", user)

            continue

        if not plugin.accepts(user):

            log.debug('Plugin %s does not accept user `%s` for authentication',

                      module, user)

            continue

        else:

            log.debug('Plugin %s accepted user `%s` for authentication',

                      module, user)

            # The user might have tried to authenticate using their email address,

            # then the username variable wouldn't contain a valid username.

# 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/>.

"""

kallithea.lib.auth_modules.auth_internal

~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Kallithea authentication plugin for built in internal auth

This file was forked by the Kallithea project in July 2014.

Original author and date, and relevant copyright and licensing information is below:

:created_on: Created on Nov 17, 2012

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import logging

from kallithea.lib import auth_modules

log = logging.getLogger(__name__)

class KallitheaAuthPlugin(auth_modules.KallitheaAuthPluginBase):

    def __init__(self):

        pass

    @hybrid_property

    def name(self):

        # Also found as kallithea.lib.model.db.User.DEFAULT_AUTH_TYPE

        return 'internal'

    def settings(self):

        return []

    def accepts(self, user, accepts_empty=True):

        """

        Custom accepts for this auth that doesn't accept empty users. We

        know that user exists in database.

        """

        return super(KallitheaAuthPlugin, self).accepts(user,

                                                        accepts_empty=False)

    def auth(self, userobj, username, password, settings, **kwargs):

        if not userobj:

            log.debug('userobj was:%s skipping', userobj)

            return None

        if userobj.extern_type != self.name:

            log.warning("userobj:%s extern_type mismatch got:`%s` expected:`%s`",

                     userobj, userobj.extern_type, self.name)

            return None

        if not username:

            log.debug('Empty username - skipping...')

            return None

        user_data = {

            "username": userobj.username,

            "firstname": userobj.firstname,

            "lastname": userobj.lastname,

            "groups": [],

            "email": userobj.email,

            "admin": userobj.admin,

            "extern_name": userobj.user_id,

        }

        from kallithea.lib import auth

        password_match = auth.check_password(password, userobj.password)

        if userobj.is_default_user:

            log.info('user %s authenticated correctly as anonymous user',

                     username)

            return user_data

        elif userobj.username == username and password_match:

            log.info('user %s authenticated correctly', user_data['username'])

            return user_data

        log.error("user %s had a bad password", username)

        return None

    def get_managed_fields(self):

        # Note: 'username' should only be editable (at least for user) if self registration is enabled

        return []

#

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

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

"""

kallithea.lib.auth_modules.auth_pam

~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Kallithea authentication library for PAM

This file was forked by the Kallithea project in July 2014.

Original author and date, and relevant copyright and licensing information is below:

:created_on: Created on Apr 09, 2013

:author: Alexey Larikov

"""

import grp

import logging

import pwd

import re

import socket

import threading

import time

from kallithea.lib import auth_modules

try:

    from pam import authenticate as pam_authenticate

except ImportError:

    # work around pam.authenticate missing in python-pam 1.8.*

    from pam import pam

    pam_authenticate = pam().authenticate

log = logging.getLogger(__name__)

# Cache to store PAM authenticated users

_auth_cache = dict()

_pam_lock = threading.Lock()

class KallitheaAuthPlugin(auth_modules.KallitheaExternalAuthPlugin):

    # PAM authnetication can be slow. Repository operations involve a lot of

    # auth calls. Little caching helps speedup push/pull operations significantly

    AUTH_CACHE_TTL = 4

    def __init__(self):

        firstname = getattr(userobj, 'firstname', '')

        lastname = getattr(userobj, 'lastname', '')

        user_data = {

            'username': username,

            'firstname': firstname,

            'lastname': lastname,

            'groups': [g.gr_name for g in grp.getgrall() if username in g.gr_mem],

            'email': email,

            'admin': admin,

            'extern_name': username,

        }

        try:

            user_pw_data = pwd.getpwnam(username)

            regex = settings["gecos"]

            match = re.search(regex, user_pw_data.pw_gecos)

            if match:

                user_data["firstname"] = match.group('first_name')

                user_data["lastname"] = match.group('last_name')

        except Exception:

            log.warning("Cannot extract additional info for PAM user %s", username)

            pass

        log.info('user %s authenticated correctly', user_data['username'])

        return user_data

    def get_managed_fields(self):

        return ['username', 'password']