# 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 logging

import traceback

import importlib

from kallithea.lib.utils2 import str2bool

from kallithea.lib.compat import formatted_json, hybrid_property

from kallithea.lib.auth import PasswordGenerator

from kallithea.model.user import UserModel

from kallithea.model.db import Setting, User

from kallithea.model.meta import Session

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)

class KallitheaAuthPluginBase(object):

    auth_func_attrs = {

        "username": "unique username",

        "firstname": "first name",

        "lastname": "last name",

        "email": "email address",

        "groups": '["list", "of", "groups"]',

        "extern_name": "name in external source of record",

        "admin": 'True|False defines if user should be Kallithea admin',

    }

    @property

    def validators(self):

        """

        Exposes Kallithea validators modules

        """

        # this is a hack to overcome issues with pylons threadlocals and

        # translator object _() not being registered properly.

        class LazyCaller(object):

            def __init__(self, name):

                self.validator_name = name

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

                from kallithea.model import validators as v

                obj = getattr(v, self.validator_name)

                #log.debug('Initializing lazy formencode object: %s', obj)

                return LazyFormencode(obj, *args, **kwargs)

        class ProxyGet(object):

            def __getattribute__(self, name):

                return LazyCaller(name)

        return ProxyGet()

    @hybrid_property

    def name(self):

        """

        Returns the name of this authentication plugin.

        :returns: string

        """

        raise NotImplementedError("Not implemented in base class")

    @hybrid_property

    def is_container_auth(self):

        """

        Returns bool if this module uses container auth.

        This property will trigger an automatic call to authenticate on

        a visit to the website or during a push/pull.

        :returns: bool

        """

        return False

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

        """

                "name": "OPTION_NAME",

                "type": "[bool|password|string|int|select]",

                ["values": ["opt1", "opt2", ...]]

                "validator": "expr"

                "description": "A short description of the option" [,

                "default": Default Value],

                ["formname": "Friendly Name for Forms"]

            } [, ...]

        ]

        This is used to interrogate the authentication plugin as to what

        settings it expects to be present and configured.

        'type' is a shorthand notation for what kind of value this option is.

        This is primarily used by the auth web form to control how the option

        is configured.

                bool : checkbox

                password : password input box

                string : input box

                select : single select dropdown

        'validator' is an lazy instantiated form field validator object, ala

        formencode. You need to *call* this object to init the validators.

        All calls to Kallithea validators should be used through self.validators

        which is a lazy loading proxy of formencode module.

        """

        raise NotImplementedError("Not implemented in base class")

    def plugin_settings(self):

        """

        This method is called by the authentication framework, not the .settings()

        method. This method adds a few default settings (e.g., "enabled"), so that

        plugin authors don't have to maintain a bunch of boilerplate.

        OVERRIDING THIS METHOD WILL CAUSE YOUR PLUGIN TO FAIL.

        """

        rcsettings = self.settings()

        rcsettings.insert(0, {

            "name": "enabled",

            "validator": self.validators.StringBoolean(if_missing=False),

            "type": "bool",

            "description": "Enable or Disable this Authentication Plugin",

            "formname": "Enabled"

            }

        )

        return rcsettings

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

        """

        Given a user object (which may be None), username, a plaintext password,

        and a settings object (containing all the keys needed as listed in settings()),

        authenticate this user's login attempt.

        Return None on failure. On success, return a dictionary with keys from

        KallitheaAuthPluginBase.auth_func_attrs.

        This is later validated for correctness.

        """

        raise NotImplementedError("not implemented in base class")

    def _authenticate(self, userobj, username, passwd, settings, **kwargs):

        """

        Wrapper to call self.auth() that validates call on it

        """

        user_data = self.auth(userobj, username, passwd, settings, **kwargs)

        if user_data is not None:

            return self._validate_auth_return(user_data)

        return None

    def _validate_auth_return(self, user_data):

        if not isinstance(user_data, dict):

            raise Exception('returned value from auth must be a dict')

        for k in self.auth_func_attrs:

            if k not in user_data:

                raise Exception('Missing %s attribute from returned data' % k)

        return user_data

class KallitheaExternalAuthPlugin(KallitheaAuthPluginBase):

    def use_fake_password(self):

        """

        Return a boolean that indicates whether or not we should set the user's

        password to a random value when it is authenticated by this plugin.

        If your plugin provides authentication, then you will generally want this.

        :returns: boolean

        """

        raise NotImplementedError("Not implemented in base class")

    def _authenticate(self, userobj, username, passwd, settings, **kwargs):

        user_data = super(KallitheaExternalAuthPlugin, self)._authenticate(

            userobj, username, passwd, settings, **kwargs)

        if user_data is not None:

            if self.use_fake_password():

                # Randomize the PW because we don't need it, but don't want

                # them blank either

                passwd = PasswordGenerator().gen_password(length=8)

            log.debug('Updating or creating user info from %s plugin',

                      self.name)

            user = UserModel().create_or_update(

                username=user_data['username'],

                password=passwd,

                email=user_data["email"],

                firstname=user_data["firstname"],

                lastname=user_data["lastname"],

                admin=user_data["admin"],

                extern_name=user_data["extern_name"],

            )

            # enforce user is just in given groups, all of them has to be ones

            # created from plugins. We store this info in _group_data JSON field

            groups = user_data['groups'] or []

            UserGroupModel().enforce_groups(user, groups, self.name)

            Session().commit()

        return user_data

def loadplugin(plugin):

    """

    Imports, instantiates, and returns the authentication plugin in the module named by plugin

    (e.g., plugin='kallithea.lib.auth_modules.auth_internal'). Returns an instance of the

    KallitheaAuthPluginBase subclass on success, raises exceptions on failure.

    raises:

        AttributeError -- no KallitheaAuthPlugin class in the module

        TypeError -- if the KallitheaAuthPlugin is not a subclass of ours KallitheaAuthPluginBase

        ImportError -- if we couldn't import the plugin at all

    """

    log.debug("Importing %s", plugin)

    if not plugin.startswith(u'kallithea.lib.auth_modules.auth_'):

        parts = plugin.split(u'.lib.auth_modules.auth_', 1)

        if len(parts) == 2:

            _module, pn = parts

            plugin = u'kallithea.lib.auth_modules.auth_' + pn

    PLUGIN_CLASS_NAME = "KallitheaAuthPlugin"

    try:

        module = importlib.import_module(plugin)

    except (ImportError, TypeError):

        log.error(traceback.format_exc())

        # TODO: make this more error prone, if by some accident we screw up

        # the plugin name, the crash is pretty bad and hard to recover

        raise

    log.debug("Loaded auth plugin from %s (module:%s, file:%s)",

              plugin, module.__name__, module.__file__)

    pluginclass = getattr(module, PLUGIN_CLASS_NAME)

    if not issubclass(pluginclass, KallitheaAuthPluginBase):

        raise TypeError("Authentication class %s.KallitheaAuthPlugin is not "

                        "a subclass of %s" % (plugin, KallitheaAuthPluginBase))

    plugin = pluginclass()

    if plugin.plugin_settings.im_func != KallitheaAuthPluginBase.plugin_settings.im_func:

        raise TypeError("Authentication class %s.KallitheaAuthPluginBase "

                        "has overridden the plugin_settings method, which is "

                        "forbidden." % plugin)

def get_auth_plugins():

    """Return a list of instances of plugins that are available and enabled"""

    auth_plugins = []

    for plugin_name in Setting.get_by_name("auth_plugins").app_settings_value:

        try:

            plugin = loadplugin(plugin_name)

        except Exception:

            log.exception('Failed to load authentication module %s' % (plugin_name))

        else:

            auth_plugins.append(plugin)

    return auth_plugins

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

        log.debug('Settings for auth plugin %s:\n%s', plugin_name, formatted_json(plugin_settings))

        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 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 as the plugin has accepted the user, .username field should

            # have a valid username, so use it for authentication purposes.

            if user is not None:

                username = user.username

        log.info('Authenticating user using %s plugin', module)

        # _authenticate is a wrapper for .auth() method of plugin.

        # it checks if .auth() sends proper data. For KallitheaExternalAuthPlugin

        # it also maps users to Database and maps the attributes returned

        # from .auth() to Kallithea database. If this function returns data

        # then auth is correct.

        user_data = plugin._authenticate(user, username, password,

                                           plugin_settings,

                                           environ=environ or {})

        log.debug('Plugin user data: %s', user_data)

        if user_data is not None:

            log.debug('Plugin returned proper authentication data')

            return user_data

        # we failed to Auth because .auth() method didn't return the user

        if username:

            log.warning("User `%s` failed to authenticate against %s",

                        username, module)

    return None

def get_managed_fields(user):

    """return list of fields that are managed by the user's auth source, usually some of

    """

    auth_plugins = get_auth_plugins()

    for plugin in auth_plugins:

        module = plugin.__class__.__module__

        log.debug('testing %s (%s) with auth plugin %s', user, user.extern_type, module)

        if plugin.name == user.extern_type:

            return plugin.get_managed_fields()

    log.error('no auth plugin %s found for %s', user.extern_type, user)

    return [] # TODO: Fail badly instead of allowing everything to be edited?

            {

                "name": "email_header",

                "validator": self.validators.UnicodeString(strip=True),

                "type": "string",

                "description": "Optional request header to extract the email from",

                "default": "",

                "formname": "Email header"

            },

            {

                "name": "firstname_header",

                "validator": self.validators.UnicodeString(strip=True),

                "type": "string",

                "description": "Optional request header to extract the first name from",

                "default": "",

                "formname": "Firstname header"

            },

            {

                "name": "lastname_header",

                "validator": self.validators.UnicodeString(strip=True),

                "type": "string",

                "description": "Optional request header to extract the last name from",

                "default": "",

                "formname": "Lastname header"

            },

            {

                "name": "fallback_header",

                "validator": self.validators.UnicodeString(strip=True),

                "type": "string",

                "description": "Request header to extract the user from when main one fails",

                "default": "HTTP_X_FORWARDED_USER",

                "formname": "Fallback header"

            },

            {

                "name": "clean_username",

                "validator": self.validators.StringBoolean(if_missing=False),

                "type": "bool",

                "description": "Perform cleaning of user, if passed user has @ in username "

                               "then first part before @ is taken. "

                               "If there's \\ in the username only the part after \\ is taken",

                "default": "True",

                "formname": "Clean username"

            },

        ]

        return settings

    def use_fake_password(self):

        return True

    def _clean_username(self, username):

        # Removing realm and domain from username

        username = username.partition('@')[0]

        username = username.rpartition('\\')[2]

        return username

    def _get_username(self, environ, settings):

        username = None

        environ = environ or {}

        if not environ:

            log.debug('got empty environ: %s', environ)

        settings = settings or {}

        if settings.get('header'):

            header = settings.get('header')

            username = environ.get(header)

            log.debug('extracted %s:%s', header, username)

        # fallback mode

        if not username and settings.get('fallback_header'):

            header = settings.get('fallback_header')

            username = environ.get(header)

            log.debug('extracted %s:%s', header, username)

        if username and str2bool(settings.get('clean_username')):

            log.debug('Received username %s from container', username)

            username = self._clean_username(username)

            log.debug('New cleanup user is: %s', username)

        return username

    def get_user(self, username=None, **kwargs):

        """

        Helper method for user fetching in plugins, by default it's using

        simple fetch by username, but this method can be customized in plugins

        eg. container auth plugin to fetch user by environ params

        :param username: username if given to fetch

        :param kwargs: extra arguments needed for user fetching.

        """

        environ = kwargs.get('environ') or {}

        settings = kwargs.get('settings') or {}

        username = self._get_username(environ, settings)

        # we got the username, so use default method now

        return super(KallitheaAuthPlugin, self).get_user(username)

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

        """

        Gets the container_auth username (or email). It tries to get username

        from REMOTE_USER if this plugin is enabled, if that fails

        it tries to get username from HTTP_X_FORWARDED_USER if fallback header

        is set. clean_username extracts the username from this data if it's

        having @ in it.

        Return None on failure. On success, return a dictionary of the form:

            see: KallitheaAuthPluginBase.auth_func_attrs

        :param userobj:

        :param username:

        :param password:

        :param settings:

        :param kwargs:

        """

        environ = kwargs.get('environ')

        if not environ:

            log.debug('Empty environ data skipping...')

            return None

        if not userobj:

            userobj = self.get_user('', environ=environ, settings=settings)

        # we don't care passed username/password for container auth plugins.

        # only way to log in is using environ

        username = None

        if userobj:

            username = safe_str(getattr(userobj, 'username'))

        if not username:

            # we don't have any objects in DB, user doesn't exist, extract

            # username from environ based on the settings

            username = self._get_username(environ, settings)

        # if cannot fetch username, it's a no-go for this plugin to proceed

        if not username:

            return None

        # old attrs fetched from Kallithea database

        admin = getattr(userobj, 'admin', False)

        email = environ.get(settings.get('email_header'), getattr(userobj, 'email', ''))

        firstname = environ.get(settings.get('firstname_header'), getattr(userobj, 'firstname', ''))

        lastname = environ.get(settings.get('lastname_header'), getattr(userobj, 'lastname', ''))

        user_data = {

            'username': username,

            'firstname': safe_unicode(firstname or username),

            'lastname': safe_unicode(lastname or ''),

            'groups': [],

            'email': email or '',

            'admin': admin or False,

            'extern_name': username,

        }

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

        return user_data

    def get_managed_fields(self):

        fields = ['username', 'password']

        if(Setting.get_by_name('auth_container_email_header').app_settings_value):

            fields.append('email')

        if(Setting.get_by_name('auth_container_firstname_header').app_settings_value):

            fields.append('firstname')

        if(Setting.get_by_name('auth_container_lastname_header').app_settings_value):

            fields.append('lastname')

        return fields

                "description": "The protocol used to connect to the Atlassian CROWD server.",

                "formname": "Protocol"

            },

            {

                "name": "host",

                "validator": self.validators.UnicodeString(strip=True),

                "type": "string",

                "description": "The FQDN or IP of the Atlassian CROWD Server",

                "default": "127.0.0.1",

                "formname": "Host"

            },

            {

                "name": "port",

                "validator": self.validators.Number(strip=True),

                "type": "int",

                "description": "The Port in use by the Atlassian CROWD Server",

                "default": 8095,

                "formname": "Port"

            },

            {

                "name": "app_name",

                "validator": self.validators.UnicodeString(strip=True),

                "type": "string",

                "description": "The Application Name to authenticate to CROWD",

                "default": "",

                "formname": "Application Name"

            },

            {

                "name": "app_password",

                "validator": self.validators.UnicodeString(strip=True),

                "type": "string",

                "description": "The password to authenticate to CROWD",

                "default": "",

                "formname": "Application Password"

            },

            {

                "name": "admin_groups",

                "validator": self.validators.UnicodeString(strip=True),

                "type": "string",

                "description": "A comma separated list of group names that identify users as Kallithea Administrators",

                "formname": "Admin Groups"

            }

        ]

        return settings

    def use_fake_password(self):

        return True

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

        """

        Given a user object (which may be null), username, a plaintext password,

        and a settings object (containing all the keys needed as listed in settings()),

        authenticate this user's login attempt.

        Return None on failure. On success, return a dictionary of the form:

            see: KallitheaAuthPluginBase.auth_func_attrs

        This is later validated for correctness

        """

        if not username or not password:

            log.debug('Empty username or password skipping...')

            return None

        log.debug("Crowd settings: \n%s", formatted_json(settings))

        server = CrowdServer(**settings)

        server.set_credentials(settings["app_name"], settings["app_password"])

        crowd_user = server.user_auth(username, password)

        log.debug("Crowd returned: \n%s", formatted_json(crowd_user))

        if not crowd_user["status"]:

            log.error('Crowd authentication as %s returned no status', username)

            return None

        if not crowd_user.get('active'):

            log.error('Crowd authentication as %s returned in-active user', username)

            return None

        res = server.user_groups(crowd_user["name"])

        log.debug("Crowd groups: \n%s", formatted_json(res))

        crowd_user["groups"] = [x["name"] for x in res["groups"]]

        # old attrs fetched from Kallithea database

        admin = getattr(userobj, 'admin', False)

        email = getattr(userobj, 'email', '')

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

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

        user_data = {

            'username': crowd_user["name"] or username,

            'firstname': crowd_user["first-name"] or firstname,

            'lastname': crowd_user["last-name"] or lastname,

            'groups': crowd_user["groups"],

            'email': crowd_user["email"] or email,

            'admin': admin,

            'extern_name': crowd_user["name"],

        }

        # set an admin if we're in admin_groups of crowd

        for group in settings["admin_groups"].split(","):

            if group in user_data["groups"]:

                user_data["admin"] = True

        log.debug("Final crowd user object: \n%s", formatted_json(user_data))

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

        return user_data

    def get_managed_fields(self):

        return ['username', 'firstname', 'lastname', 'email', 'password']

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

"""

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

from kallithea.lib.compat import formatted_json, hybrid_property

from kallithea.model.db import User

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,

        }

    def get_managed_fields(self):

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

        return []

                "name": "filter",

                "validator": self.validators.UnicodeString(strip=True),

                "type": "string",

                "description": "Filter to narrow results (e.g., ou=Users, etc)",

                "formname": "LDAP Search Filter"

            },

            {

                "name": "search_scope",

                "validator": self.validators.OneOf(self._search_scopes),

                "type": "select",

                "values": self._search_scopes,

                "description": "How deep to search LDAP",

                "formname": "LDAP Search Scope"

            },

            {

                "name": "attr_login",

                "validator": self.validators.AttrLoginValidator(not_empty=True, strip=True),

                "type": "string",

                "description": "LDAP Attribute to map to user name",

                "formname": "Login Attribute"

            },

            {

                "name": "attr_firstname",

                "validator": self.validators.UnicodeString(strip=True),

                "type": "string",

                "description": "LDAP Attribute to map to first name",

                "formname": "First Name Attribute"

            },

            {

                "name": "attr_lastname",

                "validator": self.validators.UnicodeString(strip=True),

                "type": "string",

                "description": "LDAP Attribute to map to last name",

                "formname": "Last Name Attribute"

            },

            {

                "name": "attr_email",

                "validator": self.validators.UnicodeString(strip=True),

                "type": "string",

                "description": "LDAP Attribute to map to email address",

                "formname": "Email Attribute"

            }

        ]

        return settings

    def use_fake_password(self):

        return True

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

        """

        Given a user object (which may be null), username, a plaintext password,

        and a settings object (containing all the keys needed as listed in settings()),

        authenticate this user's login attempt.

        Return None on failure. On success, return a dictionary of the form:

            see: KallitheaAuthPluginBase.auth_func_attrs

        This is later validated for correctness

        """

        if not username or not password:

            log.debug('Empty username or password skipping...')

            return None

        kwargs = {

            'server': settings.get('host', ''),

            'base_dn': settings.get('base_dn', ''),

            'port': settings.get('port'),

            'bind_dn': settings.get('dn_user'),

            'bind_pass': settings.get('dn_pass'),

            'tls_kind': settings.get('tls_kind'),

            'tls_reqcert': settings.get('tls_reqcert'),

            'cacertdir': settings.get('cacertdir'),

            'ldap_filter': settings.get('filter'),

            'search_scope': settings.get('search_scope'),

            'attr_login': settings.get('attr_login'),