ldap: small fixes and improvements over ldap authentication

rhodecode/authentication/base.py

@@ -1,825 +1,825 b''
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 Authentication modules
 """
 import socket
 import string
 import colander
 import copy
 import logging
 import time
 import traceback
 import warnings
 import functools
 
 from pyramid.threadlocal import get_current_registry
 
 from rhodecode.authentication import AuthenticationPluginRegistry
 from rhodecode.authentication.interface import IAuthnPluginRegistry
 from rhodecode.authentication.schema import AuthnPluginSettingsSchemaBase
 from rhodecode.lib import rc_cache
 from rhodecode.lib.statsd_client import StatsdClient
 from rhodecode.lib.auth import PasswordGenerator, _RhodeCodeCryptoBCrypt
 from rhodecode.lib.str_utils import safe_bytes
 from rhodecode.lib.utils2 import safe_int, safe_str
 from rhodecode.lib.exceptions import (LdapConnectionError, LdapUsernameError, LdapPasswordError)
 from rhodecode.model.db import User
 from rhodecode.model.meta import Session
 from rhodecode.model.settings import SettingsModel
 from rhodecode.model.user import UserModel
 from rhodecode.model.user_group import UserGroupModel
 
 
 log = logging.getLogger(__name__)
 
 # auth types that authenticate() function can receive
 VCS_TYPE = 'vcs'
 HTTP_TYPE = 'http'
 
 external_auth_session_key = 'rhodecode.external_auth'
 
 
 class hybrid_property(object):
     """
     a property decorator that works both for instance and class
     """
     def __init__(self, fget, fset=None, fdel=None, expr=None):
         self.fget = fget
         self.fset = fset
         self.fdel = fdel
         self.expr = expr or fget
         functools.update_wrapper(self, fget)
 
     def __get__(self, instance, owner):
         if instance is None:
             return self.expr(owner)
         else:
             return self.fget(instance)
 
     def __set__(self, instance, value):
         self.fset(instance, value)
 
     def __delete__(self, instance):
         self.fdel(instance)
 
 
 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 RhodeCodeAuthPluginBase(object):
     # UID is used to register plugin to the registry
     uid = None
 
     # cache the authentication request for N amount of seconds. Some kind
     # of authentication methods are very heavy and it's very efficient to cache
     # the result of a call. If it's set to None (default) cache is off
     AUTH_CACHE_TTL = None
     AUTH_CACHE = {}
 
     auth_func_attrs = {
         "username": "unique username",
         "firstname": "first name",
         "lastname": "last name",
         "email": "email address",
         "groups": '["list", "of", "groups"]',
         "user_group_sync":
             'True|False defines if returned user groups should be synced',
         "extern_name": "name in external source of record",
         "extern_type": "type of external source of record",
         "admin": 'True|False defines if user should be RhodeCode super admin',
         "active":
             'True|False defines active state of user internally for RhodeCode',
         "active_from_extern":
             "True|False|None, active state from the external auth, "
             "None means use definition from RhodeCode extern_type active value"
 
     }
     # set on authenticate() method and via set_auth_type func.
     auth_type = None
 
     # set on authenticate() method and via set_calling_scope_repo, this is a
     # calling scope repository when doing authentication most likely on VCS
     # operations
     acl_repo_name = None
 
     # List of setting names to store encrypted. Plugins may override this list
     # to store settings encrypted.
     _settings_encrypted = []
 
     # Mapping of python to DB settings model types. Plugins may override or
     # extend this mapping.
     _settings_type_map = {
         colander.String: 'unicode',
         colander.Integer: 'int',
         colander.Boolean: 'bool',
         colander.List: 'list',
     }
 
     # list of keys in settings that are unsafe to be logged, should be passwords
     # or other crucial credentials
     _settings_unsafe_keys = []
 
     def __init__(self, plugin_id):
         self._plugin_id = plugin_id
 
     def __str__(self):
         return self.get_id()
 
     def _get_setting_full_name(self, name):
         """
         Return the full setting name used for storing values in the database.
         """
         # TODO: johbo: Using the name here is problematic. It would be good to
         # introduce either new models in the database to hold Plugin and
         # PluginSetting or to use the plugin id here.
         return f'auth_{self.name}_{name}'
 
     def _get_setting_type(self, name):
         """
         Return the type of a setting. This type is defined by the SettingsModel
         and determines how the setting is stored in DB. Optionally the suffix
         `.encrypted` is appended to instruct SettingsModel to store it
         encrypted.
         """
         schema_node = self.get_settings_schema().get(name)
         db_type = self._settings_type_map.get(
             type(schema_node.typ), 'unicode')
         if name in self._settings_encrypted:
             db_type = f'{db_type}.encrypted'
         return db_type
 
     @classmethod
     def docs(cls):
         """
         Defines documentation url which helps with plugin setup
         """
         return ''
 
     @classmethod
     def icon(cls):
         """
         Defines ICON in SVG format for authentication method
         """
         return ''
 
     def is_enabled(self):
         """
         Returns true if this plugin is enabled. An enabled plugin can be
         configured in the admin interface but it is not consulted during
         authentication.
         """
         auth_plugins = SettingsModel().get_auth_plugins()
         return self.get_id() in auth_plugins
 
     def is_active(self, plugin_cached_settings=None):
         """
         Returns true if the plugin is activated. An activated plugin is
         consulted during authentication, assumed it is also enabled.
         """
         return self.get_setting_by_name(
             'enabled', plugin_cached_settings=plugin_cached_settings)
 
     def get_id(self):
         """
         Returns the plugin id.
         """
         return self._plugin_id
 
     def get_display_name(self, load_from_settings=False):
         """
         Returns a translation string for displaying purposes.
         if load_from_settings is set, plugin settings can override the display name
         """
         raise NotImplementedError('Not implemented in base class')
 
     def get_settings_schema(self):
         """
         Returns a colander schema, representing the plugin settings.
         """
         return AuthnPluginSettingsSchemaBase()
 
     def _propagate_settings(self, raw_settings):
         settings = {}
         for node in self.get_settings_schema():
             settings[node.name] = self.get_setting_by_name(
                 node.name, plugin_cached_settings=raw_settings)
         return settings
 
     def get_settings(self, use_cache=True):
         """
         Returns the plugin settings as dictionary.
         """
 
         raw_settings = SettingsModel().get_all_settings(cache=use_cache)
         settings = self._propagate_settings(raw_settings)
 
         return settings
 
     def get_setting_by_name(self, name, default=None, plugin_cached_settings=None):
         """
         Returns a plugin setting by name.
         """
         full_name = f'rhodecode_{self._get_setting_full_name(name)}'
         if plugin_cached_settings:
             plugin_settings = plugin_cached_settings
         else:
             plugin_settings = SettingsModel().get_all_settings()
 
         if full_name in plugin_settings:
             return plugin_settings[full_name]
         else:
             return default
 
     def create_or_update_setting(self, name, value):
         """
         Create or update a setting for this plugin in the persistent storage.
         """
         full_name = self._get_setting_full_name(name)
         type_ = self._get_setting_type(name)
         db_setting = SettingsModel().create_or_update_setting(
             full_name, value, type_)
         return db_setting.app_settings_value
 
     def log_safe_settings(self, settings):
         """
         returns a log safe representation of settings, without any secrets
         """
         settings_copy = copy.deepcopy(settings)
         for k in self._settings_unsafe_keys:
             if k in settings_copy:
                 del settings_copy[k]
         return settings_copy
 
     @hybrid_property
     def name(self):
         """
         Returns the name of this authentication plugin.
 
         :returns: string
         """
         raise NotImplementedError("Not implemented in base class")
 
     def get_url_slug(self):
         """
         Returns a slug which should be used when constructing URLs which refer
         to this plugin. By default it returns the plugin name. If the name is
         not suitable for using it in an URL the plugin should override this
         method.
         """
         return self.name
 
     @property
     def is_headers_auth(self):
         """
         Returns True if this authentication plugin uses HTTP headers as
         authentication method.
         """
         return False
 
     @hybrid_property
     def is_container_auth(self):
         """
         Deprecated method that indicates if this authentication plugin uses
         HTTP headers as authentication method.
         """
         warnings.warn(
             'Use is_headers_auth instead.', category=DeprecationWarning)
         return self.is_headers_auth
 
     @hybrid_property
     def allows_creating_users(self):
         """
         Defines if Plugin allows users to be created on-the-fly when
         authentication is called. Controls how external plugins should behave
         in terms if they are allowed to create new users, or not. Base plugins
         should not be allowed to, but External ones should be !
 
         :return: bool
         """
         return False
 
     def set_auth_type(self, auth_type):
         self.auth_type = auth_type
 
     def set_calling_scope_repo(self, acl_repo_name):
         self.acl_repo_name = acl_repo_name
 
     def allows_authentication_from(
             self, user, allows_non_existing_user=True,
             allowed_auth_plugins=None, allowed_auth_sources=None):
         """
         Checks if this authentication module should accept a request for
         the current user.
 
         :param user: user object fetched using plugin's get_user() method.
         :param allows_non_existing_user: if True, don't allow the
             user to be empty, meaning not existing in our database
         :param allowed_auth_plugins: if provided, users extern_type will be
             checked against a list of provided extern types, which are plugin
             auth_names in the end
         :param allowed_auth_sources: authentication type allowed,
             `http` or `vcs` default is both.
             defines if plugin will accept only http authentication vcs
             authentication(git/hg) or both
         :returns: boolean
         """
         if not user and not allows_non_existing_user:
             log.debug('User is empty but plugin does not allow empty users,'
                       'not allowed to authenticate')
             return False
 
         expected_auth_plugins = allowed_auth_plugins or [self.name]
         if user and (user.extern_type and
                      user.extern_type not in expected_auth_plugins):
             log.debug(
                 'User `%s` is bound to `%s` auth type. Plugin allows only '
                 '%s, skipping', user, user.extern_type, expected_auth_plugins)
 
             return False
 
         # by default accept both
         expected_auth_from = allowed_auth_sources or [HTTP_TYPE, VCS_TYPE]
         if self.auth_type not in expected_auth_from:
             log.debug('Current auth source is %s but plugin only allows %s',
                       self.auth_type, expected_auth_from)
             return False
 
         return True
 
     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. headers auth plugin to fetch user by environ params
 
         :param username: username if given to fetch from database
         :param kwargs: extra arguments needed for user fetching.
         """
 
         user = None
         log.debug(
             'Trying to fetch user `%s` from RhodeCode database', username)
         if username:
             user = User.get_by_username(username)
             if not user:
                 log.debug('User not found, fallback to fetch user in '
                           'case insensitive mode')
                 user = User.get_by_username(username, case_insensitive=True)
         else:
             log.debug('provided username:`%s` is empty skipping...', username)
         if not user:
             log.debug('User `%s` not found in database', username)
         else:
             log.debug('Got DB user:%s', user)
         return user
 
     def user_activation_state(self):
         """
         Defines user activation state when creating new users
 
         :returns: boolean
         """
         raise NotImplementedError("Not implemented in base class")
 
     def auth(self, userobj, username, passwd, 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: RhodeCodeAuthPluginBase.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
 
         :param userobj: userobj
         :param username: username
         :param passwd: plaintext password
         :param settings: plugin settings
         """
         auth = self.auth(userobj, username, passwd, settings, **kwargs)
         if auth:
             auth['_plugin'] = self.name
             auth['_ttl_cache'] = self.get_ttl_cache(settings)
             # check if hash should be migrated ?
             new_hash = auth.get('_hash_migrate')
             if new_hash:
                 # new_hash is a newly encrypted destination hash
                 self._migrate_hash_to_bcrypt(username, passwd, new_hash)
             if 'user_group_sync' not in auth:
                 auth['user_group_sync'] = False
             return self._validate_auth_return(auth)
         return auth
 
     def _migrate_hash_to_bcrypt(self, username, password, new_hash):
         new_hash_cypher = _RhodeCodeCryptoBCrypt()
         # extra checks, so make sure new hash is correct.
         password_as_bytes = safe_bytes(password)
 
         if new_hash and new_hash_cypher.hash_check(password_as_bytes, new_hash):
             cur_user = User.get_by_username(username)
             cur_user.password = new_hash
             Session().add(cur_user)
             Session().flush()
             log.info('Migrated user %s hash to bcrypt', cur_user)
 
     def _validate_auth_return(self, ret):
         if not isinstance(ret, dict):
             raise Exception('returned value from auth must be a dict')
         for k in self.auth_func_attrs:
             if k not in ret:
                 raise Exception('Missing %s attribute from returned data' % k)
         return ret
 
     def get_ttl_cache(self, settings=None):
         plugin_settings = settings or self.get_settings()
         # we set default to 30, we make a compromise here,
         # performance > security, mostly due to LDAP/SVN, majority
         # of users pick cache_ttl to be enabled
         from rhodecode.authentication import plugin_default_auth_ttl
         cache_ttl = plugin_default_auth_ttl
 
         if isinstance(self.AUTH_CACHE_TTL, int):
             # plugin cache set inside is more important than the settings value
             cache_ttl = self.AUTH_CACHE_TTL
-        elif plugin_settings.get('cache_ttl'):
+        elif 'cache_ttl' in plugin_settings:
             cache_ttl = safe_int(plugin_settings.get('cache_ttl'), 0)
 
         plugin_cache_active = bool(cache_ttl and cache_ttl > 0)
         return plugin_cache_active, cache_ttl
 
 
 class RhodeCodeExternalAuthPlugin(RhodeCodeAuthPluginBase):
 
     @hybrid_property
     def allows_creating_users(self):
         return True
 
     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):
         # at this point _authenticate calls plugin's `auth()` function
         auth = super()._authenticate(
             userobj, username, passwd, settings, **kwargs)
 
         if auth:
             # maybe plugin will clean the username ?
             # we should use the return value
             username = auth['username']
 
             # if external source tells us that user is not active, we should
             # skip rest of the process. This can prevent from creating users in
             # RhodeCode when using external authentication, but if it's
             # inactive user we shouldn't create that user anyway
             if auth['active_from_extern'] is False:
                 log.warning(
                     "User %s authenticated against %s, but is inactive",
                     username, self.__module__)
                 return None
 
             cur_user = User.get_by_username(username, case_insensitive=True)
             is_user_existing = cur_user is not None
 
             if is_user_existing:
                 log.debug('Syncing user `%s` from '
                           '`%s` plugin', username, self.name)
             else:
                 log.debug('Creating non existing user `%s` from '
                           '`%s` plugin', username, self.name)
 
             if self.allows_creating_users:
                 log.debug('Plugin `%s` allows to '
                           'create new users', self.name)
             else:
                 log.debug('Plugin `%s` does not allow to '
                           'create new users', self.name)
 
             user_parameters = {
                 'username': username,
                 'email': auth["email"],
                 'firstname': auth["firstname"],
                 'lastname': auth["lastname"],
                 'active': auth["active"],
                 'admin': auth["admin"],
                 'extern_name': auth["extern_name"],
                 'extern_type': self.name,
                 'plugin': self,
                 'allow_to_create_user': self.allows_creating_users,
             }
 
             if not is_user_existing:
                 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=16)
                 user_parameters['password'] = passwd
             else:
                 # Since the password is required by create_or_update method of
                 # UserModel, we need to set it explicitly.
                 # The create_or_update method is smart and recognises the
                 # password hashes as well.
                 user_parameters['password'] = cur_user.password
 
             # we either create or update users, we also pass the flag
             # that controls if this method can actually do that.
             # raises NotAllowedToCreateUserError if it cannot, and we try to.
             user = UserModel().create_or_update(**user_parameters)
             Session().flush()
             # 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
 
             if auth['user_group_sync']:
                 try:
                     groups = auth['groups'] or []
                     log.debug(
                         'Performing user_group sync based on set `%s` '
                         'returned by `%s` plugin', groups, self.name)
                     UserGroupModel().enforce_groups(user, groups, self.name)
                 except Exception:
                     # for any reason group syncing fails, we should
                     # proceed with login
                     log.error(traceback.format_exc())
 
             Session().commit()
         return auth
 
 
 class AuthLdapBase(object):
 
     @classmethod
     def _build_servers(cls, ldap_server_type, ldap_server, port, use_resolver=True):
 
         def host_resolver(host, port, full_resolve=True):
             """
             Main work for this function is to prevent ldap connection issues,
             and detect them early using a "greenified" sockets
             """
             host = host.strip()
             if not full_resolve:
                 return f'{host}:{port}'
 
             log.debug('LDAP: Resolving IP for LDAP host `%s`', host)
             try:
                 ip = socket.gethostbyname(host)
                 log.debug('LDAP: Got LDAP host `%s` ip %s', host, ip)
             except Exception:
                 raise LdapConnectionError(f'Failed to resolve host: `{host}`')
 
             log.debug('LDAP: Checking if IP %s is accessible', ip)
             s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
             try:
                 s.connect((ip, int(port)))
                 s.shutdown(socket.SHUT_RD)
                 log.debug('LDAP: connection to %s successful', ip)
             except Exception:
                 raise LdapConnectionError(
                     f'Failed to connect to host: `{host}:{port}`')
 
             return f'{host}:{port}'
 
         if len(ldap_server) == 1:
             # in case of single server use resolver to detect potential
             # connection issues
             full_resolve = True
         else:
             full_resolve = False
 
         return ', '.join(
             ["{}://{}".format(
                 ldap_server_type,
                 host_resolver(host, port, full_resolve=use_resolver and full_resolve))
              for host in ldap_server])
 
     @classmethod
     def _get_server_list(cls, servers):
-        return map(string.strip, servers.split(','))
+        return [s.strip() for s in servers.split(',')]
 
     @classmethod
     def get_uid(cls, username, server_addresses):
         uid = username
         for server_addr in server_addresses:
             uid = chop_at(username, "@%s" % server_addr)
         return uid
 
     @classmethod
     def validate_username(cls, username):
         if "," in username:
             raise LdapUsernameError(
                 f"invalid character `,` in username: `{username}`")
 
     @classmethod
     def validate_password(cls, username, password):
         if not password:
             msg = "Authenticating user %s with blank password not allowed"
             log.warning(msg, username)
             raise LdapPasswordError(msg)
 
 
 def loadplugin(plugin_id):
     """
     Loads and returns an instantiated authentication plugin.
     Returns the RhodeCodeAuthPluginBase subclass on success,
     or None on failure.
     """
     # TODO: Disusing pyramids thread locals to retrieve the registry.
     authn_registry = get_authn_registry()
     plugin = authn_registry.get_plugin(plugin_id)
     if plugin is None:
         log.error('Authentication plugin not found: "%s"', plugin_id)
     return plugin
 
 
 def get_authn_registry(registry=None) -> AuthenticationPluginRegistry:
     registry = registry or get_current_registry()
     authn_registry = registry.queryUtility(IAuthnPluginRegistry)
     return authn_registry
 
 
 def authenticate(username, password, environ=None, auth_type=None,
                  skip_missing=False, registry=None, acl_repo_name=None):
     """
     Authentication function used for access control,
     It tries to authenticate based on enabled authentication modules.
 
     :param username: username can be empty for headers auth
     :param password: password can be empty for headers auth
     :param environ: environ headers passed for headers auth
     :param auth_type: type of authentication, either `HTTP_TYPE` or `VCS_TYPE`
     :param skip_missing: ignores plugins that are in db but not in environment
     :param registry: pyramid registry
     :param acl_repo_name: name of repo for ACL checks
     :returns: None if auth failed, plugin_user dict if auth is correct
     """
     if not auth_type or auth_type not in [HTTP_TYPE, VCS_TYPE]:
         raise ValueError(f'auth type must be on of http, vcs got "{auth_type}" instead')
 
     auth_credentials = (username and password)
     headers_only = environ and not auth_credentials
 
     authn_registry = get_authn_registry(registry)
 
     plugins_to_check = authn_registry.get_plugins_for_authentication()
     log.debug('authentication: headers=%s, username_and_passwd=%s', headers_only, bool(auth_credentials))
     log.debug('Starting ordered authentication chain using %s plugins',
               [x.name for x in plugins_to_check])
 
     for plugin in plugins_to_check:
         plugin.set_auth_type(auth_type)
         plugin.set_calling_scope_repo(acl_repo_name)
 
         if headers_only and not plugin.is_headers_auth:
             log.debug('Auth type is for headers only and plugin `%s` is not '
                       'headers plugin, skipping...', plugin.get_id())
             continue
 
         log.debug('Trying authentication using ** %s **', plugin.get_id())
 
         # load plugin settings from RhodeCode database
         plugin_settings = plugin.get_settings()
         plugin_sanitized_settings = plugin.log_safe_settings(plugin_settings)
         log.debug('Plugin `%s` settings:%s', plugin.get_id(), plugin_sanitized_settings)
 
         # use plugin's method of user extraction.
         user = plugin.get_user(username, environ=environ,
                                settings=plugin_settings)
         display_user = user.username if user else username
         log.debug(
             'Plugin %s extracted user is `%s`', plugin.get_id(), display_user)
 
         if not plugin.allows_authentication_from(user):
             log.debug('Plugin %s does not accept user `%s` for authentication',
                       plugin.get_id(), display_user)
             continue
         else:
             log.debug('Plugin %s accepted user `%s` for authentication',
                       plugin.get_id(), display_user)
 
         log.info('Authenticating user `%s` using %s plugin',
                  display_user, plugin.get_id())
 
         plugin_cache_active, cache_ttl = plugin.get_ttl_cache(plugin_settings)
 
         log.debug('AUTH_CACHE_TTL for plugin `%s` active: %s (TTL: %s)',
                   plugin.get_id(), plugin_cache_active, cache_ttl)
 
         user_id = user.user_id if user else 'no-user'
         # don't cache for empty users
         plugin_cache_active = plugin_cache_active and user_id
         cache_namespace_uid = f'cache_user_auth.{rc_cache.PERMISSIONS_CACHE_VER}.{user_id}'
         region = rc_cache.get_or_create_region('cache_perms', cache_namespace_uid)
 
         @region.conditional_cache_on_arguments(namespace=cache_namespace_uid,
                                                expiration_time=cache_ttl,
                                                condition=plugin_cache_active)
         def compute_auth(
                 cache_name, plugin_name, username, password):
 
             # _authenticate is a wrapper for .auth() method of plugin.
             # it checks if .auth() sends proper data.
             # For RhodeCodeExternalAuthPlugin it also maps users to
             # Database and maps the attributes returned from .auth()
             # to RhodeCode database. If this function returns data
             # then auth is correct.
             log.debug('Running plugin `%s` _authenticate method '
                       'using username and password', plugin.get_id())
             return plugin._authenticate(
                 user, username, password, plugin_settings,
                 environ=environ or {})
 
         start = time.time()
         # for environ based auth, password can be empty, but then the validation is
         # on the server that fills in the env data needed for authentication
         plugin_user = compute_auth('auth', plugin.name, username, (password or ''))
 
         auth_time = time.time() - start
         log.debug('Authentication for plugin `%s` completed in %.4fs, '
                   'expiration time of fetched cache %.1fs.',
                   plugin.get_id(), auth_time, cache_ttl,
                   extra={"plugin": plugin.get_id(), "time": auth_time})
 
         log.debug('PLUGIN USER DATA: %s', plugin_user)
 
         statsd = StatsdClient.statsd
 
         if plugin_user:
             log.debug('Plugin returned proper authentication data')
             if statsd:
                 elapsed_time_ms = round(1000.0 * auth_time)  # use ms only
                 statsd.incr('rhodecode_login_success_total')
                 statsd.timing("rhodecode_login_timing.histogram", elapsed_time_ms,
                               tags=[f"plugin:{plugin.get_id()}"],
                               use_decimals=False
                 )
             return plugin_user
 
         # we failed to Auth because .auth() method didn't return proper user
         log.debug("User `%s` failed to authenticate against %s",
                   display_user, plugin.get_id())
         if statsd:
             statsd.incr('rhodecode_login_fail_total')
 
     # case when we failed to authenticate against all defined plugins
     return None
 
 
 def chop_at(s, sub, inclusive=False):
     """Truncate string ``s`` at the first occurrence of ``sub``.
 
     If ``inclusive`` is true, truncate just after ``sub`` rather than at it.
 
     >>> chop_at("plutocratic brats", "rat")
     'plutoc'
     >>> chop_at("plutocratic brats", "rat", True)
     'plutocrat'
     """
     pos = s.find(sub)
     if pos == -1:
         return s
     if inclusive:
         return s[:pos+len(sub)]
     return s[:pos]

rhodecode/authentication/plugins/auth_ldap.py

@@ -1,550 +1,550 b''
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 RhodeCode authentication plugin for LDAP
 """
 
 import logging
 import traceback
 
 import colander
 from rhodecode.translation import _
 from rhodecode.authentication.base import (
     RhodeCodeExternalAuthPlugin, AuthLdapBase, hybrid_property)
 from rhodecode.authentication.schema import AuthnPluginSettingsSchemaBase
 from rhodecode.authentication.routes import AuthnPluginResourceBase
 from rhodecode.lib.colander_utils import strip_whitespace
 from rhodecode.lib.exceptions import (
     LdapConnectionError, LdapUsernameError, LdapPasswordError, LdapImportError
 )
 from rhodecode.lib.str_utils import safe_str
 from rhodecode.model.db import User
 from rhodecode.model.validators import Missing
 
 log = logging.getLogger(__name__)
 
 try:
     import ldap
 except ImportError:
     # means that python-ldap is not installed, we use Missing object to mark
     # ldap lib is Missing
     ldap = Missing
 
 
 class LdapError(Exception):
     pass
 
 
 def plugin_factory(plugin_id, *args, **kwargs):
     """
     Factory function that is called during plugin discovery.
     It returns the plugin instance.
     """
     plugin = RhodeCodeAuthPlugin(plugin_id)
     return plugin
 
 
 class LdapAuthnResource(AuthnPluginResourceBase):
     pass
 
 
 class AuthLdap(AuthLdapBase):
     default_tls_cert_dir = '/etc/openldap/cacerts'
 
     scope_labels = {
         ldap.SCOPE_BASE: 'SCOPE_BASE',
         ldap.SCOPE_ONELEVEL: 'SCOPE_ONELEVEL',
         ldap.SCOPE_SUBTREE: 'SCOPE_SUBTREE',
     }
 
     def __init__(self, server, base_dn, port=389, bind_dn='', bind_pass='',
                  tls_kind='PLAIN', tls_reqcert='DEMAND', tls_cert_file=None,
                  tls_cert_dir=None, ldap_version=3,
                  search_scope='SUBTREE', attr_login='uid',
                  ldap_filter='', timeout=None):
         if ldap == Missing:
             raise LdapImportError("Missing or incompatible ldap library")
 
         self.debug = False
         self.timeout = timeout or 60 * 5
         self.ldap_version = ldap_version
         self.ldap_server_type = 'ldap'
 
         self.TLS_KIND = tls_kind
 
         if self.TLS_KIND == 'LDAPS':
             port = port or 636
             self.ldap_server_type += 's'
 
         OPT_X_TLS_DEMAND = 2
         self.TLS_REQCERT = getattr(ldap, 'OPT_X_TLS_%s' % tls_reqcert, OPT_X_TLS_DEMAND)
         self.TLS_CERT_FILE = tls_cert_file or ''
         self.TLS_CERT_DIR = tls_cert_dir or self.default_tls_cert_dir
 
         # split server into list
         self.SERVER_ADDRESSES = self._get_server_list(server)
         self.LDAP_SERVER_PORT = port
 
         # USE FOR READ ONLY BIND TO LDAP SERVER
         self.attr_login = attr_login
 
         self.LDAP_BIND_DN = safe_str(bind_dn)
         self.LDAP_BIND_PASS = safe_str(bind_pass)
 
         self.SEARCH_SCOPE = getattr(ldap, 'SCOPE_%s' % search_scope)
         self.BASE_DN = safe_str(base_dn)
         self.LDAP_FILTER = safe_str(ldap_filter)
 
     def _get_ldap_conn(self):
 
         if self.debug:
             ldap.set_option(ldap.OPT_DEBUG_LEVEL, 255)
 
         if self.TLS_CERT_FILE and hasattr(ldap, 'OPT_X_TLS_CACERTFILE'):
             ldap.set_option(ldap.OPT_X_TLS_CACERTFILE, self.TLS_CERT_FILE)
 
         elif hasattr(ldap, 'OPT_X_TLS_CACERTDIR'):
             ldap.set_option(ldap.OPT_X_TLS_CACERTDIR, self.TLS_CERT_DIR)
 
         if self.TLS_KIND != 'PLAIN':
             ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, self.TLS_REQCERT)
 
         ldap.set_option(ldap.OPT_REFERRALS, ldap.OPT_OFF)
         ldap.set_option(ldap.OPT_RESTART, ldap.OPT_ON)
 
         # init connection now
         ldap_servers = self._build_servers(
             self.ldap_server_type,  self.SERVER_ADDRESSES, self.LDAP_SERVER_PORT)
         log.debug('initializing LDAP connection to:%s', ldap_servers)
         ldap_conn = ldap.initialize(ldap_servers)
         ldap_conn.set_option(ldap.OPT_NETWORK_TIMEOUT, self.timeout)
         ldap_conn.set_option(ldap.OPT_TIMEOUT, self.timeout)
         ldap_conn.timeout = self.timeout
 
         if self.ldap_version == 2:
             ldap_conn.protocol = ldap.VERSION2
         else:
             ldap_conn.protocol = ldap.VERSION3
 
         if self.TLS_KIND == 'START_TLS':
             ldap_conn.start_tls_s()
 
         if self.LDAP_BIND_DN and self.LDAP_BIND_PASS:
             log.debug('Trying simple_bind with password and given login DN: %r',
                       self.LDAP_BIND_DN)
             ldap_conn.simple_bind_s(self.LDAP_BIND_DN, self.LDAP_BIND_PASS)
         log.debug('simple_bind successful')
         return ldap_conn
 
     def fetch_attrs_from_simple_bind(self, ldap_conn, dn, username, password):
         scope = ldap.SCOPE_BASE
         scope_label = self.scope_labels.get(scope)
         ldap_filter = '(objectClass=*)'
 
         try:
             log.debug('Trying authenticated search bind with dn: %r SCOPE: %s (and filter: %s)',
                       dn, scope_label, ldap_filter)
             ldap_conn.simple_bind_s(dn, safe_str(password))
             response = ldap_conn.search_ext_s(dn, scope, ldap_filter, attrlist=['*', '+'])
 
             if not response:
                 log.error('search bind returned empty results: %r', response)
                 return {}
             else:
                 _dn, attrs = response[0]
                 return attrs
 
         except ldap.INVALID_CREDENTIALS:
             log.debug("LDAP rejected password for user '%s': %s, org_exc:",
                       username, dn, exc_info=True)
 
     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.
 
         :param username: username
         :param password: password
         """
 
         uid = self.get_uid(username, self.SERVER_ADDRESSES)
         user_attrs = {}
         dn = ''
 
         self.validate_password(username, password)
         self.validate_username(username)
         scope_label = self.scope_labels.get(self.SEARCH_SCOPE)
 
         ldap_conn = None
         try:
             ldap_conn = self._get_ldap_conn()
-            filter_ = '(&%s(%s=%s))' % (
+            filter_ = '(&{}({}={}))'.format(
                 self.LDAP_FILTER, self.attr_login, username)
             log.debug("Authenticating %r filter %s and scope: %s",
                       self.BASE_DN, filter_, scope_label)
 
             ldap_objects = ldap_conn.search_ext_s(
                 self.BASE_DN, self.SEARCH_SCOPE, filter_, attrlist=['*', '+'])
 
             if not ldap_objects:
                 log.debug("No matching LDAP objects for authentication "
                           "of UID:'%s' username:(%s)", uid, username)
                 raise ldap.NO_SUCH_OBJECT()
 
             log.debug('Found %s matching ldap object[s], trying to authenticate on each one now...', len(ldap_objects))
             for (dn, _attrs) in ldap_objects:
                 if dn is None:
                     continue
 
                 user_attrs = self.fetch_attrs_from_simple_bind(
                     ldap_conn, dn, username, password)
 
                 if user_attrs:
                     log.debug('Got authenticated user attributes from DN:%s', dn)
                     break
             else:
                 raise LdapPasswordError(
                     f'Failed to authenticate user `{username}` with given password')
 
         except ldap.NO_SUCH_OBJECT:
             log.debug("LDAP says no such user '%s' (%s), org_exc:",
                       uid, username, exc_info=True)
             raise LdapUsernameError('Unable to find user')
         except ldap.SERVER_DOWN:
             org_exc = traceback.format_exc()
             raise LdapConnectionError(
                 "LDAP can't access authentication server, org_exc:%s" % org_exc)
         finally:
             if ldap_conn:
                 log.debug('ldap: connection release')
                 try:
                     ldap_conn.unbind_s()
                 except Exception:
                     # for any reason this can raise exception we must catch it
                     # to not crush the server
                     pass
 
         return dn, user_attrs
 
 
 class LdapSettingsSchema(AuthnPluginSettingsSchemaBase):
     tls_kind_choices = ['PLAIN', 'LDAPS', 'START_TLS']
     tls_reqcert_choices = ['NEVER', 'ALLOW', 'TRY', 'DEMAND', 'HARD']
     search_scope_choices = ['BASE', 'ONELEVEL', 'SUBTREE']
 
     host = colander.SchemaNode(
         colander.String(),
         default='',
         description=_('Host[s] of the LDAP Server \n'
                       '(e.g., 192.168.2.154, or ldap-server.domain.com.\n '
                       'Multiple servers can be specified using commas'),
         preparer=strip_whitespace,
         title=_('LDAP Host'),
         widget='string')
     port = colander.SchemaNode(
         colander.Int(),
         default=389,
         description=_('Custom port that the LDAP server is listening on. '
                       'Default value is: 389, use 636 for LDAPS (SSL)'),
         preparer=strip_whitespace,
         title=_('Port'),
         validator=colander.Range(min=0, max=65536),
         widget='int')
 
     timeout = colander.SchemaNode(
         colander.Int(),
         default=60 * 5,
         description=_('Timeout for LDAP connection'),
         preparer=strip_whitespace,
         title=_('Connection timeout'),
         validator=colander.Range(min=1),
         widget='int')
 
     dn_user = colander.SchemaNode(
         colander.String(),
         default='',
         description=_('Optional user DN/account to connect to LDAP if authentication is required. \n'
                       'e.g., cn=admin,dc=mydomain,dc=com, or '
                       'uid=root,cn=users,dc=mydomain,dc=com, or [e-mail]'),
         missing='',
         preparer=strip_whitespace,
         title=_('Bind account'),
         widget='string')
     dn_pass = colander.SchemaNode(
         colander.String(),
         default='',
         description=_('Password to authenticate for given user DN.'),
         missing='',
         preparer=strip_whitespace,
         title=_('Bind account password'),
         widget='password')
     tls_kind = colander.SchemaNode(
         colander.String(),
         default=tls_kind_choices[0],
         description=_('TLS Type'),
         title=_('Connection Security'),
         validator=colander.OneOf(tls_kind_choices),
         widget='select')
     tls_reqcert = colander.SchemaNode(
         colander.String(),
         default=tls_reqcert_choices[0],
         description=_('Require Cert over TLS?. Self-signed and custom '
                       'certificates can be used when\n `RhodeCode Certificate` '
                       'found in admin > settings > system info page is extended.'),
         title=_('Certificate Checks'),
         validator=colander.OneOf(tls_reqcert_choices),
         widget='select')
     tls_cert_file = colander.SchemaNode(
         colander.String(),
         default='',
         description=_('This specifies the PEM-format file path containing '
                       'certificates for use in TLS connection.\n'
                       'If not specified `TLS Cert dir` will be used'),
         title=_('TLS Cert file'),
         missing='',
         widget='string')
     tls_cert_dir = colander.SchemaNode(
         colander.String(),
         default=AuthLdap.default_tls_cert_dir,
         description=_('This specifies the path of a directory that contains individual '
                       'CA certificates in separate files.'),
         title=_('TLS Cert dir'),
         widget='string')
     base_dn = colander.SchemaNode(
         colander.String(),
         default='',
         description=_('Base DN to search. Dynamic bind is supported. Add `$login` marker '
                       'in it to be replaced with current user username \n'
                       '(e.g., dc=mydomain,dc=com, or ou=Users,dc=mydomain,dc=com)'),
         missing='',
         preparer=strip_whitespace,
         title=_('Base DN'),
         widget='string')
     filter = colander.SchemaNode(
         colander.String(),
         default='',
         description=_('Filter to narrow results \n'
                       '(e.g., (&(objectCategory=Person)(objectClass=user)), or \n'
                       '(memberof=cn=rc-login,ou=groups,ou=company,dc=mydomain,dc=com)))'),
         missing='',
         preparer=strip_whitespace,
         title=_('LDAP Search Filter'),
         widget='string')
 
     search_scope = colander.SchemaNode(
         colander.String(),
         default=search_scope_choices[2],
         description=_('How deep to search LDAP. If unsure set to SUBTREE'),
         title=_('LDAP Search Scope'),
         validator=colander.OneOf(search_scope_choices),
         widget='select')
     attr_login = colander.SchemaNode(
         colander.String(),
         default='uid',
         description=_('LDAP Attribute to map to user name (e.g., uid, or sAMAccountName)'),
         preparer=strip_whitespace,
         title=_('Login Attribute'),
         missing_msg=_('The LDAP Login attribute of the CN must be specified'),
         widget='string')
     attr_email = colander.SchemaNode(
         colander.String(),
         default='',
         description=_('LDAP Attribute to map to email address (e.g., mail).\n'
                       'Emails are a crucial part of RhodeCode. \n'
                       'If possible add a valid email attribute to ldap users.'),
         missing='',
         preparer=strip_whitespace,
         title=_('Email Attribute'),
         widget='string')
     attr_firstname = colander.SchemaNode(
         colander.String(),
         default='',
         description=_('LDAP Attribute to map to first name (e.g., givenName)'),
         missing='',
         preparer=strip_whitespace,
         title=_('First Name Attribute'),
         widget='string')
     attr_lastname = colander.SchemaNode(
         colander.String(),
         default='',
         description=_('LDAP Attribute to map to last name (e.g., sn)'),
         missing='',
         preparer=strip_whitespace,
         title=_('Last Name Attribute'),
         widget='string')
 
 
 class RhodeCodeAuthPlugin(RhodeCodeExternalAuthPlugin):
     uid = 'ldap'
     # used to define dynamic binding in the
     DYNAMIC_BIND_VAR = '$login'
     _settings_unsafe_keys = ['dn_pass']
 
     def includeme(self, config):
         config.add_authn_plugin(self)
         config.add_authn_resource(self.get_id(), LdapAuthnResource(self))
         config.add_view(
             'rhodecode.authentication.views.AuthnPluginViewBase',
             attr='settings_get',
             renderer='rhodecode:templates/admin/auth/plugin_settings.mako',
             request_method='GET',
             route_name='auth_home',
             context=LdapAuthnResource)
         config.add_view(
             'rhodecode.authentication.views.AuthnPluginViewBase',
             attr='settings_post',
             renderer='rhodecode:templates/admin/auth/plugin_settings.mako',
             request_method='POST',
             route_name='auth_home',
             context=LdapAuthnResource)
 
     def get_settings_schema(self):
         return LdapSettingsSchema()
 
     def get_display_name(self, load_from_settings=False):
         return _('LDAP')
 
     @classmethod
     def docs(cls):
         return "https://docs.rhodecode.com/RhodeCode-Enterprise/auth/auth-ldap.html"
 
     @hybrid_property
     def name(self):
         return "ldap"
 
     def use_fake_password(self):
         return True
 
     def user_activation_state(self):
         def_user_perms = User.get_default_user().AuthUser().permissions['global']
         return 'hg.extern_activate.auto' in def_user_perms
 
     def try_dynamic_binding(self, username, password, current_args):
         """
         Detects marker inside our original bind, and uses dynamic auth if
         present
         """
 
         org_bind = current_args['bind_dn']
         passwd = current_args['bind_pass']
 
-        def has_bind_marker(username):
-            if self.DYNAMIC_BIND_VAR in username:
+        def has_bind_marker(_username):
+            if self.DYNAMIC_BIND_VAR in _username:
                 return True
 
         # we only passed in user with "special" variable
         if org_bind and has_bind_marker(org_bind) and not passwd:
             log.debug('Using dynamic user/password binding for ldap '
                       'authentication. Replacing `%s` with username',
                       self.DYNAMIC_BIND_VAR)
             current_args['bind_dn'] = org_bind.replace(
                 self.DYNAMIC_BIND_VAR, username)
             current_args['bind_pass'] = password
 
         return current_args
 
     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: RhodeCodeAuthPluginBase.auth_func_attrs
         This is later validated for correctness
         """
 
         if not username or not password:
             log.debug('Empty username or password skipping...')
             return None
 
         ldap_args = {
             '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'),
             'tls_cert_file': settings.get('tls_cert_file'),
             'tls_cert_dir': settings.get('tls_cert_dir'),
             'search_scope': settings.get('search_scope'),
             'attr_login': settings.get('attr_login'),
             'ldap_version': 3,
             'ldap_filter': settings.get('filter'),
             'timeout': settings.get('timeout')
         }
 
         ldap_attrs = self.try_dynamic_binding(username, password, ldap_args)
 
         log.debug('Checking for ldap authentication.')
 
         try:
-            aldap = AuthLdap(**ldap_args)
-            (user_dn, ldap_attrs) = aldap.authenticate_ldap(username, password)
+            auth_ldap = AuthLdap(**ldap_args)
+            (user_dn, ldap_attrs) = auth_ldap.authenticate_ldap(username, password)
             log.debug('Got ldap DN response %s', user_dn)
 
-            def get_ldap_attr(k):
-                return ldap_attrs.get(settings.get(k), [''])[0]
+            def get_ldap_attr(k) -> str:
+                return safe_str(ldap_attrs.get(settings.get(k), [b''])[0])
 
             # old attrs fetched from RhodeCode database
             admin = getattr(userobj, 'admin', False)
             active = getattr(userobj, 'active', True)
             email = getattr(userobj, 'email', '')
             username = getattr(userobj, 'username', username)
             firstname = getattr(userobj, 'firstname', '')
             lastname = getattr(userobj, 'lastname', '')
             extern_type = getattr(userobj, 'extern_type', '')
 
             groups = []
 
             user_attrs = {
                 'username': username,
-                'firstname': safe_str(get_ldap_attr('attr_firstname') or firstname),
-                'lastname': safe_str(get_ldap_attr('attr_lastname') or lastname),
+                'firstname': get_ldap_attr('attr_firstname') or firstname,
+                'lastname': get_ldap_attr('attr_lastname') or lastname,
                 'groups': groups,
                 'user_group_sync': False,
                 'email': get_ldap_attr('attr_email') or email,
                 'admin': admin,
                 'active': active,
                 'active_from_extern': None,
                 'extern_name': user_dn,
                 'extern_type': extern_type,
             }
 
             log.debug('ldap user: %s', user_attrs)
             log.info('user `%s` authenticated correctly', user_attrs['username'],
                      extra={"action": "user_auth_ok", "auth_module": "auth_ldap", "username": user_attrs["username"]})
 
             return user_attrs
 
         except (LdapUsernameError, LdapPasswordError, LdapImportError):
             log.exception("LDAP related exception")
             return None
         except (Exception,):
             log.exception("Other exception")
             return None
 
 
 def includeme(config):
     plugin_id = f'egg:rhodecode-enterprise-ce#{RhodeCodeAuthPlugin.uid}'
     plugin_factory(plugin_id).includeme(config)

rhodecode/authentication/views.py

@@ -1,178 +1,183 b''
 # Copyright (C) 2012-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import colander
 import formencode.htmlfill
 import logging
 
 from pyramid.httpexceptions import HTTPFound
 from pyramid.renderers import render
 from pyramid.response import Response
 
 from rhodecode.apps._base import BaseAppView
 from rhodecode.authentication.base import get_authn_registry
 from rhodecode.lib import helpers as h
 from rhodecode.lib.auth import (
     LoginRequired, HasPermissionAllDecorator, CSRFRequired)
 from rhodecode.model.forms import AuthSettingsForm
 from rhodecode.model.meta import Session
 from rhodecode.model.settings import SettingsModel
 
 log = logging.getLogger(__name__)
 
 
 class AuthnPluginViewBase(BaseAppView):
 
     def load_default_context(self):
         c = self._get_local_tmpl_context()
         self.plugin = self.context.plugin
         return c
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     def settings_get(self, defaults=None, errors=None):
         """
         View that displays the plugin settings as a form.
         """
         c = self.load_default_context()
         defaults = defaults or {}
         errors = errors or {}
         schema = self.plugin.get_settings_schema()
 
         # Compute default values for the form. Priority is:
         # 1. Passed to this method 2. DB value 3. Schema default
         for node in schema:
             if node.name not in defaults:
                 defaults[node.name] = self.plugin.get_setting_by_name(
                     node.name, node.default)
 
         template_context = {
             'defaults': defaults,
             'errors': errors,
             'plugin': self.context.plugin,
             'resource': self.context,
         }
 
         return self._get_template_context(c, **template_context)
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     @CSRFRequired()
     def settings_post(self):
         """
         View that validates and stores the plugin settings.
         """
         _ = self.request.translate
         self.load_default_context()
         schema = self.plugin.get_settings_schema()
         data = self.request.params
 
         try:
             valid_data = schema.deserialize(data)
         except colander.Invalid as e:
             # Display error message and display form again.
             h.flash(
                 _('Errors exist when saving plugin settings. '
                   'Please check the form inputs.'),
                 category='error')
             defaults = {key: data[key] for key in data if key in schema}
             return self.settings_get(errors=e.asdict(), defaults=defaults)
 
         # Store validated data.
         for name, value in valid_data.items():
             self.plugin.create_or_update_setting(name, value)
         Session().commit()
         SettingsModel().invalidate_settings_cache()
 
+        authn_registry = get_authn_registry(self.request.registry)
+        authn_registry.invalidate_auth_plugins_cache()
+
         # Display success message and redirect.
         h.flash(_('Auth settings updated successfully.'), category='success')
         redirect_to = self.request.resource_path(self.context, route_name='auth_home')
 
         return HTTPFound(redirect_to)
 
 
 class AuthSettingsView(BaseAppView):
     def load_default_context(self):
         c = self._get_local_tmpl_context()
         return c
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     def index(self, defaults=None, errors=None, prefix_error=False):
         c = self.load_default_context()
 
         defaults = defaults or {}
         authn_registry = get_authn_registry(self.request.registry)
         enabled_plugins = SettingsModel().get_auth_plugins()
 
         # Create template context and render it.
         template_context = {
             'resource': self.context,
             'available_plugins': authn_registry.get_plugins(),
             'enabled_plugins': enabled_plugins,
         }
         html = render('rhodecode:templates/admin/auth/auth_settings.mako',
                       self._get_template_context(c, **template_context),
                       self.request)
 
         # Create form default values and fill the form.
         form_defaults = {
             'auth_plugins': ',\n'.join(enabled_plugins)
         }
         form_defaults.update(defaults)
         html = formencode.htmlfill.render(
             html,
             defaults=form_defaults,
             errors=errors,
             prefix_error=prefix_error,
             encoding="UTF-8",
             force_defaults=False)
 
         return Response(html)
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     @CSRFRequired()
     def auth_settings(self):
         _ = self.request.translate
         try:
             form = AuthSettingsForm(self.request.translate)()
             form_result = form.to_python(self.request.POST)
             plugins = ','.join(form_result['auth_plugins'])
             setting = SettingsModel().create_or_update_setting(
                 'auth_plugins', plugins)
             Session().add(setting)
             Session().commit()
             SettingsModel().invalidate_settings_cache()
             h.flash(_('Auth settings updated successfully.'), category='success')
         except formencode.Invalid as errors:
             e = errors.error_dict or {}
             h.flash(_('Errors exist when saving plugin setting. '
                       'Please check the form inputs.'), category='error')
             return self.index(
                 defaults=errors.value,
                 errors=e,
                 prefix_error=False)
         except Exception:
             log.exception('Exception in auth_settings')
             h.flash(_('Error occurred during update of auth settings.'),
                     category='error')
 
+        authn_registry = get_authn_registry(self.request.registry)
+        authn_registry.invalidate_auth_plugins_cache()
         redirect_to = self.request.resource_path(self.context, route_name='auth_home')
 
         return HTTPFound(redirect_to)

rhodecode/lib/middleware/vcs.py

@@ -1,295 +1,298 b''
 
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import gzip
 import shutil
 import logging
 import tempfile
 import urllib.parse
 
 from webob.exc import HTTPNotFound
 
 import rhodecode
 from rhodecode.lib.middleware.utils import get_path_info
 from rhodecode.lib.middleware.appenlight import wrap_in_appenlight_if_enabled
 from rhodecode.lib.middleware.simplegit import SimpleGit, GIT_PROTO_PAT
 from rhodecode.lib.middleware.simplehg import SimpleHg
 from rhodecode.lib.middleware.simplesvn import SimpleSvn
 from rhodecode.model.settings import VcsSettingsModel
 
 
 log = logging.getLogger(__name__)
 
 VCS_TYPE_KEY = '_rc_vcs_type'
 VCS_TYPE_SKIP = '_rc_vcs_skip'
 
 
 def is_git(environ):
     """
     Returns True if requests should be handled by GIT wsgi middleware
     """
     path_info = get_path_info(environ)
     is_git_path = GIT_PROTO_PAT.match(path_info)
     log.debug(
         'request path: `%s` detected as GIT PROTOCOL %s', path_info,
         is_git_path is not None)
 
     return is_git_path
 
 
 def is_hg(environ):
     """
     Returns True if requests target is mercurial server - header
     ``HTTP_ACCEPT`` of such request would start with ``application/mercurial``.
     """
     is_hg_path = False
 
     http_accept = environ.get('HTTP_ACCEPT')
 
     if http_accept and http_accept.startswith('application/mercurial'):
         query = urllib.parse.parse_qs(environ['QUERY_STRING'])
         if 'cmd' in query:
             is_hg_path = True
 
     path_info = get_path_info(environ)
     log.debug(
         'request path: `%s` detected as HG PROTOCOL %s', path_info,
         is_hg_path)
 
     return is_hg_path
 
 
 def is_svn(environ):
     """
     Returns True if requests target is Subversion server
     """
 
     http_dav = environ.get('HTTP_DAV', '')
     magic_path_segment = rhodecode.CONFIG.get(
         'rhodecode_subversion_magic_path', '/!svn')
     path_info = get_path_info(environ)
     is_svn_path = (
         'subversion' in http_dav or
         magic_path_segment in path_info
         or environ['REQUEST_METHOD'] in ['PROPFIND', 'PROPPATCH']
     )
     log.debug(
         'request path: `%s` detected as SVN PROTOCOL %s', path_info,
         is_svn_path)
 
     return is_svn_path
 
 
 class GunzipMiddleware(object):
     """
     WSGI middleware that unzips gzip-encoded requests before
     passing on to the underlying application.
     """
 
     def __init__(self, application):
         self.app = application
 
     def __call__(self, environ, start_response):
         accepts_encoding_header = environ.get('HTTP_CONTENT_ENCODING', b'')
 
         if b'gzip' in accepts_encoding_header:
             log.debug('gzip detected, now running gunzip wrapper')
             wsgi_input = environ['wsgi.input']
 
             if not hasattr(environ['wsgi.input'], 'seek'):
                 # The gzip implementation in the standard library of Python 2.x
                 # requires the '.seek()' and '.tell()' methods to be available
                 # on the input stream.  Read the data into a temporary file to
                 # work around this limitation.
 
                 wsgi_input = tempfile.SpooledTemporaryFile(64 * 1024 * 1024)
                 shutil.copyfileobj(environ['wsgi.input'], wsgi_input)
                 wsgi_input.seek(0)
 
             environ['wsgi.input'] = gzip.GzipFile(fileobj=wsgi_input, mode='r')
             # since we "Ungzipped" the content we say now it's no longer gzip
             # content encoding
             del environ['HTTP_CONTENT_ENCODING']
 
             # content length has changes ? or i'm not sure
             if 'CONTENT_LENGTH' in environ:
                 del environ['CONTENT_LENGTH']
         else:
             log.debug('content not gzipped, gzipMiddleware passing '
                       'request further')
         return self.app(environ, start_response)
 
 
 def is_vcs_call(environ):
     if VCS_TYPE_KEY in environ:
         raw_type = environ[VCS_TYPE_KEY]
         return raw_type and raw_type != VCS_TYPE_SKIP
     return False
 
 
 def detect_vcs_request(environ, backends):
     checks = {
         'hg': (is_hg, SimpleHg),
         'git': (is_git, SimpleGit),
         'svn': (is_svn, SimpleSvn),
     }
     handler = None
     # List of path views first chunk we don't do any checks
     white_list = [
         # favicon often requested by browsers
         'favicon.ico',
 
         # e.g /_file_store/download
         '_file_store++',
 
+        # login
+        "_admin/login",
+
         # _admin/api is safe too
         '_admin/api',
 
         # _admin/gist is safe too
         '_admin/gists++',
 
         # _admin/my_account is safe too
         '_admin/my_account++',
 
         # static files no detection
         '_static++',
 
         # debug-toolbar
         '_debug_toolbar++',
 
         # skip ops ping, status
         '_admin/ops/ping',
         '_admin/ops/status',
 
         # full channelstream connect should be VCS skipped
         '_admin/channelstream/connect',
 
         '++/repo_creating_check'
     ]
     path_info = get_path_info(environ)
     path_url = path_info.lstrip('/')
 
     for item in white_list:
         if item.endswith('++') and path_url.startswith(item[:-2]):
             log.debug('path `%s` in whitelist (match:%s), skipping...', path_url, item)
             return handler
         if item.startswith('++') and path_url.endswith(item[2:]):
             log.debug('path `%s` in whitelist (match:%s), skipping...', path_url, item)
             return handler
         if item == path_url:
             log.debug('path `%s` in whitelist (match:%s), skipping...', path_url, item)
             return handler
 
     if VCS_TYPE_KEY in environ:
         raw_type = environ[VCS_TYPE_KEY]
         if raw_type == VCS_TYPE_SKIP:
             log.debug('got `skip` marker for vcs detection, skipping...')
             return handler
 
         _check, handler = checks.get(raw_type) or [None, None]
         if handler:
             log.debug('got handler:%s from environ', handler)
 
     if not handler:
         log.debug('request start: checking if request for `%s` is of VCS type in order: %s', path_url, backends)
         for vcs_type in backends:
             vcs_check, _handler = checks[vcs_type]
             if vcs_check(environ):
                 log.debug('vcs handler found %s', _handler)
                 handler = _handler
                 break
 
     return handler
 
 
 class VCSMiddleware(object):
 
     def __init__(self, app, registry, config, appenlight_client):
         self.application = app
         self.registry = registry
         self.config = config
         self.appenlight_client = appenlight_client
         self.use_gzip = True
         # order in which we check the middlewares, based on vcs.backends config
         self.check_middlewares = config['vcs.backends']
 
     def vcs_config(self, repo_name=None):
         """
         returns serialized VcsSettings
         """
         try:
             return VcsSettingsModel(
                 repo=repo_name).get_ui_settings_as_config_obj()
         except Exception:
             pass
 
     def wrap_in_gzip_if_enabled(self, app, config):
         if self.use_gzip:
             app = GunzipMiddleware(app)
         return app
 
     def _get_handler_app(self, environ):
         app = None
         log.debug('VCSMiddleware: detecting vcs type.')
         handler = detect_vcs_request(environ, self.check_middlewares)
         if handler:
             app = handler(self.config, self.registry)
 
         return app
 
     def __call__(self, environ, start_response):
         # check if we handle one of interesting protocols, optionally extract
         # specific vcsSettings and allow changes of how things are wrapped
         vcs_handler = self._get_handler_app(environ)
         if vcs_handler:
             # translate the _REPO_ID into real repo NAME for usage
             # in middleware
 
             path_info = get_path_info(environ)
             environ['PATH_INFO'] = vcs_handler._get_by_id(path_info)
 
             # Set acl, url and vcs repo names.
             vcs_handler.set_repo_names(environ)
 
             # register repo config back to the handler
             vcs_conf = self.vcs_config(vcs_handler.acl_repo_name)
             # maybe damaged/non existent settings. We still want to
             # pass that point to validate on is_valid_and_existing_repo
             # and return proper HTTP Code back to client
             if vcs_conf:
                 vcs_handler.repo_vcs_config = vcs_conf
 
             # check for type, presence in database and on filesystem
             if not vcs_handler.is_valid_and_existing_repo(
                     vcs_handler.acl_repo_name,
                     vcs_handler.base_path,
                     vcs_handler.SCM):
                 return HTTPNotFound()(environ, start_response)
 
             environ['REPO_NAME'] = vcs_handler.url_repo_name
 
             # Wrap handler in middlewares if they are enabled.
             vcs_handler = self.wrap_in_gzip_if_enabled(
                 vcs_handler, self.config)
             vcs_handler, _ = wrap_in_appenlight_if_enabled(
                 vcs_handler, self.config, self.appenlight_client)
 
             return vcs_handler(environ, start_response)
 
         return self.application(environ, start_response)