authentication: run modernize for python3

rhodecode/authentication/__init__.py

@@ -1,99 +1,97 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 logging
 import importlib
 
 from pyramid.authentication import SessionAuthenticationHelper
 
 from rhodecode.authentication.registry import AuthenticationPluginRegistry
 from rhodecode.authentication.routes import root_factory
 from rhodecode.authentication.routes import AuthnRootResource
 from rhodecode.apps._base import ADMIN_PREFIX
 from rhodecode.model.settings import SettingsModel
 
 log = logging.getLogger(__name__)
 
 legacy_plugin_prefix = 'py:'
 plugin_default_auth_ttl = 30
 
 
 def _import_legacy_plugin(plugin_id):
     module_name = plugin_id.split(legacy_plugin_prefix, 1)[-1]
     module = importlib.import_module(module_name)
     return module.plugin_factory(plugin_id=plugin_id)
 
 
 def discover_legacy_plugins(config, prefix=legacy_plugin_prefix):
     """
     Function that imports the legacy plugins stored in the 'auth_plugins'
     setting in database which are using the specified prefix. Normally 'py:' is
     used for the legacy plugins.
     """
 
     log.debug('authentication: running legacy plugin discovery for prefix %s',
               legacy_plugin_prefix)
     try:
         auth_plugins = SettingsModel().get_setting_by_name('auth_plugins')
         enabled_plugins = auth_plugins.app_settings_value
         legacy_plugins = [id_ for id_ in enabled_plugins if id_.startswith(prefix)]
     except Exception:
         legacy_plugins = []
 
     for plugin_id in legacy_plugins:
         log.debug('Legacy plugin discovered: "%s"', plugin_id)
         try:
             plugin = _import_legacy_plugin(plugin_id)
             config.include(plugin.includeme)
         except Exception as e:
             log.exception(
                 'Exception while loading legacy authentication plugin '
                 '"%s": %s', plugin_id, e)
 
 
 def includeme(config):
 
     config.set_security_policy(SessionAuthenticationHelper())
 
     # Create authentication plugin registry and add it to the pyramid registry.
     authn_registry = AuthenticationPluginRegistry(config.get_settings())
     config.add_directive('add_authn_plugin', authn_registry.add_authn_plugin)
     config.registry.registerUtility(authn_registry)
 
     # Create authentication traversal root resource.
     authn_root_resource = root_factory()
     config.add_directive('add_authn_resource',
                          authn_root_resource.add_authn_resource)
 
     # Add the authentication traversal route.
     config.add_route('auth_home',
                      ADMIN_PREFIX + '/auth*traverse',
                      factory=root_factory)
     # Add the authentication settings root views.
     config.add_view('rhodecode.authentication.views.AuthSettingsView',
                     attr='index',
                     request_method='GET',
                     route_name='auth_home',
                     context=AuthnRootResource)
     config.add_view('rhodecode.authentication.views.AuthSettingsView',
                     attr='auth_settings',
                     request_method='POST',
                     route_name='auth_home',
                     context=AuthnRootResource)

rhodecode/authentication/base.py

@@ -1,826 +1,826 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 'auth_{}_{}'.format(self.name, name)
+        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 = '{}.encrypted'.format(db_type)
+            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 = 'rhodecode_{}'.format(self._get_setting_full_name(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'):
             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(RhodeCodeExternalAuthPlugin, self)._authenticate(
+        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 '{}:{}'.format(host, port)
+                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('Failed to resolve host: `{}`'.format(host))
+                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(
-                    'Failed to connect to host: `{}:{}`'.format(host, port))
+                    f'Failed to connect to host: `{host}:{port}`')
 
-            return '{}:{}'.format(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(','))
 
     @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(
-                "invalid character `,` in username: `{}`".format(username))
+                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 = 'cache_user_auth.{}'.format(user_id)
+        cache_namespace_uid = f'cache_user_auth.{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=["plugin:{}".format(plugin.get_id())],
+                              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/interface.py

@@ -1,29 +1,27 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/
 
 from zope.interface import Interface
 
 
 class IAuthnPluginRegistry(Interface):
     """
     Interface for the authentication plugin registry. Currently this is only
     used to register and retrieve it via pyramids registry.
     """
     pass

rhodecode/authentication/plugins/__init__.py

@@ -1,19 +1,17 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/

rhodecode/authentication/plugins/auth_crowd.py

@@ -1,297 +1,295 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/
 
 """
 RhodeCode authentication plugin for Atlassian CROWD
 """
 
 
 import colander
 import base64
 import logging
 import urllib.request
 import urllib.error
 import urllib.parse
 
 from rhodecode.translation import _
 from rhodecode.authentication.base import (
     RhodeCodeExternalAuthPlugin, 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.ext_json import json, formatted_json
 from rhodecode.model.db import User
 
 log = logging.getLogger(__name__)
 
 
 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 CrowdAuthnResource(AuthnPluginResourceBase):
     pass
 
 
 class CrowdSettingsSchema(AuthnPluginSettingsSchemaBase):
     host = colander.SchemaNode(
         colander.String(),
         default='127.0.0.1',
         description=_('The FQDN or IP of the Atlassian CROWD Server'),
         preparer=strip_whitespace,
         title=_('Host'),
         widget='string')
     port = colander.SchemaNode(
         colander.Int(),
         default=8095,
         description=_('The Port in use by the Atlassian CROWD Server'),
         preparer=strip_whitespace,
         title=_('Port'),
         validator=colander.Range(min=0, max=65536),
         widget='int')
     app_name = colander.SchemaNode(
         colander.String(),
         default='',
         description=_('The Application Name to authenticate to CROWD'),
         preparer=strip_whitespace,
         title=_('Application Name'),
         widget='string')
     app_password = colander.SchemaNode(
         colander.String(),
         default='',
         description=_('The password to authenticate to CROWD'),
         preparer=strip_whitespace,
         title=_('Application Password'),
         widget='password')
     admin_groups = colander.SchemaNode(
         colander.String(),
         default='',
         description=_('A comma separated list of group names that identify '
                       'users as RhodeCode Administrators'),
         missing='',
         preparer=strip_whitespace,
         title=_('Admin Groups'),
         widget='string')
 
 
 class CrowdServer(object):
     def __init__(self, *args, **kwargs):
         """
         Create a new CrowdServer object that points to IP/Address 'host',
         on the given port, and using the given method (https/http). user and
         passwd can be set here or with set_credentials. If unspecified,
         "version" defaults to "latest".
 
         example::
 
             cserver = CrowdServer(host="127.0.0.1",
                                   port="8095",
                                   user="some_app",
                                   passwd="some_passwd",
                                   version="1")
         """
         if 'port' not in kwargs:
             kwargs["port"] = "8095"
         self._logger = kwargs.get("logger", logging.getLogger(__name__))
         self._uri = "%s://%s:%s/crowd" % (kwargs.get("method", "http"),
                                     kwargs.get("host", "127.0.0.1"),
                                     kwargs.get("port", "8095"))
         self.set_credentials(kwargs.get("user", ""),
                              kwargs.get("passwd", ""))
         self._version = kwargs.get("version", "latest")
         self._url_list = None
         self._appname = "crowd"
 
     def set_credentials(self, user, passwd):
         self.user = user
         self.passwd = passwd
         self._make_opener()
 
     def _make_opener(self):
         mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
         mgr.add_password(None, self._uri, self.user, self.passwd)
         handler = urllib.request.HTTPBasicAuthHandler(mgr)
         self.opener = urllib.request.build_opener(handler)
 
     def _request(self, url, body=None, headers=None,
                  method=None, noformat=False,
                  empty_response_ok=False):
         _headers = {"Content-type": "application/json",
                     "Accept": "application/json"}
         if self.user and self.passwd:
-            authstring = base64.b64encode("%s:%s" % (self.user, self.passwd))
+            authstring = base64.b64encode("{}:{}".format(self.user, self.passwd))
             _headers["Authorization"] = "Basic %s" % authstring
         if headers:
             _headers.update(headers)
         log.debug("Sent crowd: \n%s"
                   % (formatted_json({"url": url, "body": body,
                                            "headers": _headers})))
         request = urllib.request.Request(url, body, _headers)
         if method:
             request.get_method = lambda: method
 
         global msg
         msg = ""
         try:
             ret_doc = self.opener.open(request)
             msg = ret_doc.read()
             if not msg and empty_response_ok:
                 ret_val = {}
                 ret_val["status"] = True
                 ret_val["error"] = "Response body was empty"
             elif not noformat:
                 ret_val = json.loads(msg)
                 ret_val["status"] = True
             else:
                 ret_val = msg
         except Exception as e:
             if not noformat:
                 ret_val = {"status": False,
                            "body": body,
-                           "error": "{}\n{}".format(e, msg)}
+                           "error": f"{e}\n{msg}"}
             else:
                 ret_val = None
         return ret_val
 
     def user_auth(self, username, password):
         """Authenticate a user against crowd. Returns brief information about
         the user."""
         url = ("%s/rest/usermanagement/%s/authentication?username=%s"
                % (self._uri, self._version, username))
         body = json.dumps({"value": password})
         return self._request(url, body)
 
     def user_groups(self, username):
         """Retrieve a list of groups to which this user belongs."""
         url = ("%s/rest/usermanagement/%s/user/group/nested?username=%s"
                % (self._uri, self._version, username))
         return self._request(url)
 
 
 class RhodeCodeAuthPlugin(RhodeCodeExternalAuthPlugin):
     uid = 'crowd'
     _settings_unsafe_keys = ['app_password']
 
     def includeme(self, config):
         config.add_authn_plugin(self)
         config.add_authn_resource(self.get_id(), CrowdAuthnResource(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=CrowdAuthnResource)
         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=CrowdAuthnResource)
 
     def get_settings_schema(self):
         return CrowdSettingsSchema()
 
     def get_display_name(self, load_from_settings=False):
         return _('CROWD')
 
     @classmethod
     def docs(cls):
         return "https://docs.rhodecode.com/RhodeCode-Enterprise/auth/auth-crowd.html"
 
     @hybrid_property
     def name(self):
-        return u"crowd"
+        return "crowd"
 
     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 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
 
         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"]:
             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 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', '')
 
         user_attrs = {
             'username': username,
             'firstname': crowd_user["first-name"] or firstname,
             'lastname': crowd_user["last-name"] or lastname,
             'groups': crowd_user["groups"],
             'user_group_sync': True,
             'email': crowd_user["email"] or email,
             'admin': admin,
             'active': active,
             'active_from_extern': crowd_user.get('active'),
             'extern_name': crowd_user["name"],
             'extern_type': extern_type,
         }
 
         # set an admin if we're in admin_groups of crowd
         for group in settings["admin_groups"]:
             if group in user_attrs["groups"]:
                 user_attrs["admin"] = True
         log.debug("Final crowd user object: \n%s", formatted_json(user_attrs))
         log.info('user `%s` authenticated correctly', user_attrs['username'])
         return user_attrs
 
 
 def includeme(config):
-    plugin_id = 'egg:rhodecode-enterprise-ce#{}'.format(RhodeCodeAuthPlugin.uid)
+    plugin_id = f'egg:rhodecode-enterprise-ce#{RhodeCodeAuthPlugin.uid}'
     plugin_factory(plugin_id).includeme(config)

rhodecode/authentication/plugins/auth_headers.py

@@ -1,233 +1,231 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 logging
 
 from rhodecode.translation import _
 from rhodecode.authentication.base import (
     RhodeCodeExternalAuthPlugin, 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.str_utils import safe_str
 from rhodecode.lib.utils2 import str2bool
 from rhodecode.model.db import User
 
 
 log = logging.getLogger(__name__)
 
 
 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 HeadersAuthnResource(AuthnPluginResourceBase):
     pass
 
 
 class HeadersSettingsSchema(AuthnPluginSettingsSchemaBase):
     header = colander.SchemaNode(
         colander.String(),
         default='REMOTE_USER',
         description=_('Header to extract the user from'),
         preparer=strip_whitespace,
         title=_('Header'),
         widget='string')
     fallback_header = colander.SchemaNode(
         colander.String(),
         default='HTTP_X_FORWARDED_USER',
         description=_('Header to extract the user from when main one fails'),
         preparer=strip_whitespace,
         title=_('Fallback header'),
         widget='string')
     clean_username = colander.SchemaNode(
         colander.Boolean(),
         default=True,
         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'),
         missing=False,
         title=_('Clean username'),
         widget='bool')
 
 
 class RhodeCodeAuthPlugin(RhodeCodeExternalAuthPlugin):
     uid = 'headers'
 
     def includeme(self, config):
         config.add_authn_plugin(self)
         config.add_authn_resource(self.get_id(), HeadersAuthnResource(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=HeadersAuthnResource)
         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=HeadersAuthnResource)
 
     def get_display_name(self, load_from_settings=False):
         return _('Headers')
 
     def get_settings_schema(self):
         return HeadersSettingsSchema()
 
     @hybrid_property
     def name(self):
-        return u"headers"
+        return "headers"
 
     @property
     def is_headers_auth(self):
         return True
 
     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 _clean_username(self, username):
         # Removing realm and domain from username
         username = username.split('@')[0]
         username = username.rsplit('\\')[-1]
         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 headers', 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. headers 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(RhodeCodeAuthPlugin, self).get_user(username)
+        return super().get_user(username)
 
     def auth(self, userobj, username, password, settings, **kwargs):
         """
         Gets the headers_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: RhodeCodeAuthPluginBase.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 headers auth plugins.
         # only way to log in is using environ
         username = None
         if userobj:
             username = 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 RhodeCode database
         admin = getattr(userobj, 'admin', False)
         active = getattr(userobj, 'active', True)
         email = getattr(userobj, 'email', '')
         firstname = getattr(userobj, 'firstname', '')
         lastname = getattr(userobj, 'lastname', '')
         extern_type = getattr(userobj, 'extern_type', '')
 
         user_attrs = {
             'username': username,
             'firstname': safe_str(firstname or username),
             'lastname': safe_str(lastname or ''),
             'groups': [],
             'user_group_sync': False,
             'email': email or '',
             'admin': admin or False,
             'active': active,
             'active_from_extern': True,
             'extern_name': username,
             'extern_type': extern_type,
         }
 
         log.info('user `%s` authenticated correctly', user_attrs['username'],
                  extra={"action": "user_auth_ok", "auth_module": "auth_headers", "username": user_attrs["username"]})
         return user_attrs
 
 
 def includeme(config):
-    plugin_id = 'egg:rhodecode-enterprise-ce#{}'.format(RhodeCodeAuthPlugin.uid)
+    plugin_id = f'egg:rhodecode-enterprise-ce#{RhodeCodeAuthPlugin.uid}'
     plugin_factory(plugin_id).includeme(config)

rhodecode/authentication/plugins/auth_jasig_cas.py

@@ -1,175 +1,173 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/
 
 """
 RhodeCode authentication plugin for Jasig CAS
 http://www.jasig.org/cas
 """
 
 
 import colander
 import logging
 import rhodecode
 import urllib.request
 import urllib.parse
 import urllib.error
 
 
 from rhodecode.translation import _
 from rhodecode.authentication.base import (
     RhodeCodeExternalAuthPlugin, hybrid_property)
 from rhodecode.authentication.schema import AuthnPluginSettingsSchemaBase
 from rhodecode.authentication.routes import AuthnPluginResourceBase
 from rhodecode.lib.colander_utils import strip_whitespace
 from rhodecode.model.db import User
 from rhodecode.lib.str_utils import safe_str
 
 log = logging.getLogger(__name__)
 
 
 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 JasigCasAuthnResource(AuthnPluginResourceBase):
     pass
 
 
 class JasigCasSettingsSchema(AuthnPluginSettingsSchemaBase):
     service_url = colander.SchemaNode(
         colander.String(),
         default='https://domain.com/cas/v1/tickets',
         description=_('The url of the Jasig CAS REST service'),
         preparer=strip_whitespace,
         title=_('URL'),
         widget='string')
 
 
 class RhodeCodeAuthPlugin(RhodeCodeExternalAuthPlugin):
     uid = 'jasig_cas'
 
     def includeme(self, config):
         config.add_authn_plugin(self)
         config.add_authn_resource(self.get_id(), JasigCasAuthnResource(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=JasigCasAuthnResource)
         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=JasigCasAuthnResource)
 
     def get_settings_schema(self):
         return JasigCasSettingsSchema()
 
     def get_display_name(self, load_from_settings=False):
         return _('Jasig-CAS')
 
     @hybrid_property
     def name(self):
-        return u"jasig-cas"
+        return "jasig-cas"
 
     @property
     def is_headers_auth(self):
         return True
 
     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 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
 
         log.debug("Jasig CAS settings: %s", settings)
         params = urllib.parse.urlencode({'username': username, 'password': password})
         headers = {"Content-type": "application/x-www-form-urlencoded",
                    "Accept": "text/plain",
                    "User-Agent": "RhodeCode-auth-%s" % rhodecode.__version__}
         url = settings["service_url"]
 
         log.debug("Sent Jasig CAS: \n%s",
                   {"url": url, "body": params, "headers": headers})
         request = urllib.request.Request(url, params, headers)
         try:
             urllib.request.urlopen(request)
         except urllib.error.HTTPError as e:
             log.debug("HTTPError when requesting Jasig CAS (status code: %d)", e.code)
             return None
         except urllib.error.URLError as e:
             log.debug("URLError when requesting Jasig CAS url: %s %s", url, e)
             return None
 
         # 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', '')
 
         user_attrs = {
             'username': username,
             'firstname': safe_str(firstname or username),
             'lastname': safe_str(lastname or ''),
             'groups': [],
             'user_group_sync': False,
             'email': email or '',
             'admin': admin or False,
             'active': active,
             'active_from_extern': True,
             'extern_name': username,
             'extern_type': extern_type,
         }
 
         log.info('user `%s` authenticated correctly', user_attrs['username'])
         return user_attrs
 
 
 def includeme(config):
-    plugin_id = 'egg:rhodecode-enterprise-ce#{}'.format(RhodeCodeAuthPlugin.uid)
+    plugin_id = f'egg:rhodecode-enterprise-ce#{RhodeCodeAuthPlugin.uid}'
     plugin_factory(plugin_id).includeme(config)

rhodecode/authentication/plugins/auth_ldap.py

@@ -1,551 +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))' % (
                 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(
-                    'Failed to authenticate user `{}` with given password'.format(username))
+                    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 u"ldap"
+        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:
                 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)
             log.debug('Got ldap DN response %s', user_dn)
 
             def get_ldap_attr(k):
                 return ldap_attrs.get(settings.get(k), [''])[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),
                 '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 = 'egg:rhodecode-enterprise-ce#{}'.format(RhodeCodeAuthPlugin.uid)
+    plugin_id = f'egg:rhodecode-enterprise-ce#{RhodeCodeAuthPlugin.uid}'
     plugin_factory(plugin_id).includeme(config)

rhodecode/authentication/plugins/auth_pam.py

@@ -1,172 +1,170 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/
 
 """
 RhodeCode authentication library for PAM
 """
 
 import colander
 import grp
 import logging
 import pam
 import pwd
 import re
 import socket
 
 from rhodecode.translation import _
 from rhodecode.authentication.base import (
     RhodeCodeExternalAuthPlugin, hybrid_property)
 from rhodecode.authentication.schema import AuthnPluginSettingsSchemaBase
 from rhodecode.authentication.routes import AuthnPluginResourceBase
 from rhodecode.lib.colander_utils import strip_whitespace
 
 log = logging.getLogger(__name__)
 
 
 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 PamAuthnResource(AuthnPluginResourceBase):
     pass
 
 
 class PamSettingsSchema(AuthnPluginSettingsSchemaBase):
     service = colander.SchemaNode(
         colander.String(),
         default='login',
         description=_('PAM service name to use for authentication.'),
         preparer=strip_whitespace,
         title=_('PAM service name'),
         widget='string')
     gecos = colander.SchemaNode(
         colander.String(),
         default=r'(?P<last_name>.+),\s*(?P<first_name>\w+)',
         description=_('Regular expression for extracting user name/email etc. '
                       'from Unix userinfo.'),
         preparer=strip_whitespace,
         title=_('Gecos Regex'),
         widget='string')
 
 
 class RhodeCodeAuthPlugin(RhodeCodeExternalAuthPlugin):
     uid = 'pam'
     # PAM authentication can be slow. Repository operations involve a lot of
     # auth calls. Little caching helps speedup push/pull operations significantly
     AUTH_CACHE_TTL = 4
 
     def includeme(self, config):
         config.add_authn_plugin(self)
         config.add_authn_resource(self.get_id(), PamAuthnResource(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=PamAuthnResource)
         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=PamAuthnResource)
 
     def get_display_name(self, load_from_settings=False):
         return _('PAM')
 
     @classmethod
     def docs(cls):
         return "https://docs.rhodecode.com/RhodeCode-Enterprise/auth/auth-pam.html"
 
     @hybrid_property
     def name(self):
-        return u"pam"
+        return "pam"
 
     def get_settings_schema(self):
         return PamSettingsSchema()
 
     def use_fake_password(self):
         return True
 
     def auth(self, userobj, username, password, settings, **kwargs):
         if not username or not password:
             log.debug('Empty username or password skipping...')
             return None
         _pam = pam.pam()
         auth_result = _pam.authenticate(username, password, settings["service"])
 
         if not auth_result:
             log.error("PAM was unable to authenticate user: %s", username)
             return None
 
         log.debug('Got PAM response %s', auth_result)
 
         # old attrs fetched from RhodeCode database
-        default_email = "%s@%s" % (username, socket.gethostname())
+        default_email = "{}@{}".format(username, socket.gethostname())
         admin = getattr(userobj, 'admin', False)
         active = getattr(userobj, 'active', True)
         email = getattr(userobj, 'email', '') or default_email
         username = getattr(userobj, 'username', username)
         firstname = getattr(userobj, 'firstname', '')
         lastname = getattr(userobj, 'lastname', '')
         extern_type = getattr(userobj, 'extern_type', '')
 
         user_attrs = {
             'username': username,
             'firstname': firstname,
             'lastname': lastname,
             'groups': [g.gr_name for g in grp.getgrall()
                        if username in g.gr_mem],
             'user_group_sync': True,
             'email': email,
             'admin': admin,
             'active': active,
             'active_from_extern': None,
             'extern_name': username,
             'extern_type': extern_type,
         }
 
         try:
             user_data = pwd.getpwnam(username)
             regex = settings["gecos"]
             match = re.search(regex, user_data.pw_gecos)
             if match:
                 user_attrs["firstname"] = match.group('first_name')
                 user_attrs["lastname"] = match.group('last_name')
         except Exception:
             log.warning("Cannot extract additional info for PAM user")
             pass
 
         log.debug("pamuser: %s", user_attrs)
         log.info('user `%s` authenticated correctly', user_attrs['username'],
                  extra={"action": "user_auth_ok", "auth_module": "auth_pam", "username": user_attrs["username"]})
         return user_attrs
 
 
 def includeme(config):
-    plugin_id = 'egg:rhodecode-enterprise-ce#{}'.format(RhodeCodeAuthPlugin.uid)
+    plugin_id = f'egg:rhodecode-enterprise-ce#{RhodeCodeAuthPlugin.uid}'
     plugin_factory(plugin_id).includeme(config)

rhodecode/authentication/plugins/auth_rhodecode.py

@@ -1,222 +1,220 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/
 
 """
 RhodeCode authentication plugin for built in internal auth
 """
 
 import logging
 
 import colander
 
 from rhodecode.translation import _
 from rhodecode.lib.utils2 import safe_bytes
 from rhodecode.model.db import User
 from rhodecode.authentication.schema import AuthnPluginSettingsSchemaBase
 from rhodecode.authentication.base import (
     RhodeCodeAuthPluginBase, hybrid_property, HTTP_TYPE, VCS_TYPE)
 from rhodecode.authentication.routes import AuthnPluginResourceBase
 
 log = logging.getLogger(__name__)
 
 
 def plugin_factory(plugin_id, *args, **kwargs):
     plugin = RhodeCodeAuthPlugin(plugin_id)
     return plugin
 
 
 class RhodecodeAuthnResource(AuthnPluginResourceBase):
     pass
 
 
 class RhodeCodeAuthPlugin(RhodeCodeAuthPluginBase):
     uid = 'rhodecode'
     AUTH_RESTRICTION_NONE = 'user_all'
     AUTH_RESTRICTION_SUPER_ADMIN = 'user_super_admin'
     AUTH_RESTRICTION_SCOPE_ALL = 'scope_all'
     AUTH_RESTRICTION_SCOPE_HTTP = 'scope_http'
     AUTH_RESTRICTION_SCOPE_VCS = 'scope_vcs'
 
     def includeme(self, config):
         config.add_authn_plugin(self)
         config.add_authn_resource(self.get_id(), RhodecodeAuthnResource(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=RhodecodeAuthnResource)
         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=RhodecodeAuthnResource)
 
     def get_settings_schema(self):
         return RhodeCodeSettingsSchema()
 
     def get_display_name(self, load_from_settings=False):
         return _('RhodeCode Internal')
 
     @classmethod
     def docs(cls):
         return "https://docs.rhodecode.com/RhodeCode-Enterprise/auth/auth.html"
 
     @hybrid_property
     def name(self):
-        return u"rhodecode"
+        return "rhodecode"
 
     def user_activation_state(self):
         def_user_perms = User.get_default_user().AuthUser().permissions['global']
         return 'hg.register.auto_activate' in def_user_perms
 
     def allows_authentication_from(
             self, user, allows_non_existing_user=True,
             allowed_auth_plugins=None, allowed_auth_sources=None):
         """
         Custom method for this auth that doesn't accept non existing users.
         We know that user exists in our database.
         """
         allows_non_existing_user = False
-        return super(RhodeCodeAuthPlugin, self).allows_authentication_from(
+        return super().allows_authentication_from(
             user, allows_non_existing_user=allows_non_existing_user)
 
     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
 
         # check scope of auth
         scope_restriction = settings.get('scope_restriction', '')
 
         if scope_restriction == self.AUTH_RESTRICTION_SCOPE_HTTP \
                 and self.auth_type != HTTP_TYPE:
             log.warning("userobj:%s tried scope type %s and scope restriction is set to %s",
                         userobj, self.auth_type, scope_restriction)
             return None
 
         if scope_restriction == self.AUTH_RESTRICTION_SCOPE_VCS \
                 and self.auth_type != VCS_TYPE:
             log.warning("userobj:%s tried scope type %s and scope restriction is set to %s",
                         userobj, self.auth_type, scope_restriction)
             return None
 
         # check super-admin restriction
         auth_restriction = settings.get('auth_restriction', '')
 
         if auth_restriction == self.AUTH_RESTRICTION_SUPER_ADMIN \
                 and userobj.admin is False:
             log.warning("userobj:%s is not super-admin and auth restriction is set to %s",
                         userobj, auth_restriction)
             return None
 
         user_attrs = {
             "username": userobj.username,
             "firstname": userobj.firstname,
             "lastname": userobj.lastname,
             "groups": [],
             'user_group_sync': False,
             "email": userobj.email,
             "admin": userobj.admin,
             "active": userobj.active,
             "active_from_extern": userobj.active,
             "extern_name": userobj.user_id,
             "extern_type": userobj.extern_type,
         }
 
         log.debug("User attributes:%s", user_attrs)
         if userobj.active:
             from rhodecode.lib import auth
             crypto_backend = auth.crypto_backend()
             password_encoded = safe_bytes(password)
             password_match, new_hash = crypto_backend.hash_check_with_upgrade(
                 password_encoded, userobj.password or '')
 
             if password_match and new_hash:
                 log.debug('user %s properly authenticated, but '
                           'requires hash change to bcrypt', userobj)
                 # if password match, and we use OLD deprecated hash,
                 # we should migrate this user hash password to the new hash
                 # we store the new returned by hash_check_with_upgrade function
                 user_attrs['_hash_migrate'] = new_hash
 
             if userobj.username == User.DEFAULT_USER and userobj.active:
                 log.info('user `%s` authenticated correctly as anonymous user',
                          userobj.username,
                          extra={"action": "user_auth_ok", "auth_module": "auth_rhodecode_anon", "username": userobj.username})
                 return user_attrs
 
             elif userobj.username == username and password_match:
                 log.info('user `%s` authenticated correctly', userobj.username,
                          extra={"action": "user_auth_ok", "auth_module": "auth_rhodecode", "username": userobj.username})
                 return user_attrs
             log.warning("user `%s` used a wrong password when "
                         "authenticating on this plugin", userobj.username)
             return None
         else:
             log.warning('user `%s` failed to authenticate via %s, reason: account not '
                         'active.', username, self.name)
             return None
 
 
 class RhodeCodeSettingsSchema(AuthnPluginSettingsSchemaBase):
 
     auth_restriction_choices = [
         (RhodeCodeAuthPlugin.AUTH_RESTRICTION_NONE, 'All users'),
         (RhodeCodeAuthPlugin.AUTH_RESTRICTION_SUPER_ADMIN, 'Super admins only'),
     ]
 
     auth_scope_choices = [
         (RhodeCodeAuthPlugin.AUTH_RESTRICTION_SCOPE_ALL, 'HTTP and VCS'),
         (RhodeCodeAuthPlugin.AUTH_RESTRICTION_SCOPE_HTTP, 'HTTP only'),
     ]
 
     auth_restriction = colander.SchemaNode(
         colander.String(),
         default=auth_restriction_choices[0],
         description=_('Allowed user types for authentication using this plugin.'),
         title=_('User restriction'),
         validator=colander.OneOf([x[0] for x in auth_restriction_choices]),
         widget='select_with_labels',
         choices=auth_restriction_choices
     )
     scope_restriction = colander.SchemaNode(
         colander.String(),
         default=auth_scope_choices[0],
         description=_('Allowed protocols for authentication using this plugin. '
                       'VCS means GIT/HG/SVN. HTTP is web based login.'),
         title=_('Scope restriction'),
         validator=colander.OneOf([x[0] for x in auth_scope_choices]),
         widget='select_with_labels',
         choices=auth_scope_choices
     )
 
 
 def includeme(config):
-    plugin_id = 'egg:rhodecode-enterprise-ce#{}'.format(RhodeCodeAuthPlugin.uid)
+    plugin_id = f'egg:rhodecode-enterprise-ce#{RhodeCodeAuthPlugin.uid}'
     plugin_factory(plugin_id).includeme(config)

rhodecode/authentication/plugins/auth_token.py

@@ -1,177 +1,175 b''
-
-
 # Copyright (C) 2016-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 token plugin for built in internal auth
 """
 
 import logging
 import colander
 
 from rhodecode.authentication.schema import AuthnPluginSettingsSchemaBase
 from rhodecode.translation import _
 from rhodecode.authentication.base import (
     RhodeCodeAuthPluginBase, VCS_TYPE, hybrid_property)
 from rhodecode.authentication.routes import AuthnPluginResourceBase
 from rhodecode.model.db import User, UserApiKeys, Repository
 
 
 log = logging.getLogger(__name__)
 
 
 def plugin_factory(plugin_id, *args, **kwargs):
     plugin = RhodeCodeAuthPlugin(plugin_id)
     return plugin
 
 
 class RhodecodeAuthnResource(AuthnPluginResourceBase):
     pass
 
 
 class RhodeCodeAuthPlugin(RhodeCodeAuthPluginBase):
     """
     Enables usage of authentication tokens for vcs operations.
     """
     uid = 'token'
     AUTH_RESTRICTION_SCOPE_VCS = 'scope_vcs'
 
     def includeme(self, config):
         config.add_authn_plugin(self)
         config.add_authn_resource(self.get_id(), RhodecodeAuthnResource(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=RhodecodeAuthnResource)
         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=RhodecodeAuthnResource)
 
     def get_settings_schema(self):
         return RhodeCodeSettingsSchema()
 
     def get_display_name(self, load_from_settings=False):
         return _('Rhodecode Token')
 
     @classmethod
     def docs(cls):
         return "https://docs.rhodecode.com/RhodeCode-Enterprise/auth/auth-token.html"
 
     @hybrid_property
     def name(self):
-        return u"authtoken"
+        return "authtoken"
 
     def user_activation_state(self):
         def_user_perms = User.get_default_user().AuthUser().permissions['global']
         return 'hg.register.auto_activate' in def_user_perms
 
     def allows_authentication_from(
             self, user, allows_non_existing_user=True,
             allowed_auth_plugins=None, allowed_auth_sources=None):
         """
         Custom method for this auth that doesn't accept empty users. And also
         allows users from all other active plugins to use it and also
         authenticate against it. But only via vcs mode
         """
         from rhodecode.authentication.base import get_authn_registry
         authn_registry = get_authn_registry()
 
         active_plugins = set(
             [x.name for x in authn_registry.get_plugins_for_authentication()])
         active_plugins.discard(self.name)
 
         allowed_auth_plugins = [self.name] + list(active_plugins)
         # only for vcs operations
         allowed_auth_sources = [VCS_TYPE]
 
-        return super(RhodeCodeAuthPlugin, self).allows_authentication_from(
+        return super().allows_authentication_from(
             user, allows_non_existing_user=False,
             allowed_auth_plugins=allowed_auth_plugins,
             allowed_auth_sources=allowed_auth_sources)
 
     def auth(self, userobj, username, password, settings, **kwargs):
         if not userobj:
             log.debug('userobj was:%s skipping', userobj)
             return None
 
         user_attrs = {
             "username": userobj.username,
             "firstname": userobj.firstname,
             "lastname": userobj.lastname,
             "groups": [],
             'user_group_sync': False,
             "email": userobj.email,
             "admin": userobj.admin,
             "active": userobj.active,
             "active_from_extern": userobj.active,
             "extern_name": userobj.user_id,
             "extern_type": userobj.extern_type,
         }
 
         log.debug('Authenticating user with args %s', user_attrs)
         if userobj.active:
             # calling context repo for token scopes
             scope_repo_id = None
             if self.acl_repo_name:
                 repo = Repository.get_by_repo_name(self.acl_repo_name)
                 scope_repo_id = repo.repo_id if repo else None
 
             token_match = userobj.authenticate_by_token(
                 password, roles=[UserApiKeys.ROLE_VCS],
                 scope_repo_id=scope_repo_id)
 
             if userobj.username == username and token_match:
                 log.info(
                     'user `%s` successfully authenticated via %s',
                     user_attrs['username'], self.name)
                 return user_attrs
             log.warning('user `%s` failed to authenticate via %s, reason: bad or '
                         'inactive token.', username, self.name)
         else:
             log.warning('user `%s` failed to authenticate via %s, reason: account not '
                         'active.', username, self.name)
         return None
 
 
 def includeme(config):
-    plugin_id = 'egg:rhodecode-enterprise-ce#{}'.format(RhodeCodeAuthPlugin.uid)
+    plugin_id = f'egg:rhodecode-enterprise-ce#{RhodeCodeAuthPlugin.uid}'
     plugin_factory(plugin_id).includeme(config)
 
 
 class RhodeCodeSettingsSchema(AuthnPluginSettingsSchemaBase):
     auth_scope_choices = [
         (RhodeCodeAuthPlugin.AUTH_RESTRICTION_SCOPE_VCS, 'VCS only'),
     ]
 
     scope_restriction = colander.SchemaNode(
         colander.String(),
         default=auth_scope_choices[0],
         description=_('Choose operation scope restriction when authenticating.'),
         title=_('Scope restriction'),
         validator=colander.OneOf([x[0] for x in auth_scope_choices]),
         widget='select_with_labels',
         choices=auth_scope_choices
     )

rhodecode/authentication/registry.py

@@ -1,141 +1,139 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 time
 import logging
 
 from pyramid.exceptions import ConfigurationError
 from zope.interface import implementer
 
 from rhodecode.authentication.interface import IAuthnPluginRegistry
 from rhodecode.model.settings import SettingsModel
 from rhodecode.lib.utils2 import safe_str
 from rhodecode.lib.statsd_client import StatsdClient
 from rhodecode.lib import rc_cache
 
 log = logging.getLogger(__name__)
 
 
 @implementer(IAuthnPluginRegistry)
 class AuthenticationPluginRegistry(object):
 
     # INI settings key to set a fallback authentication plugin.
     fallback_plugin_key = 'rhodecode.auth_plugin_fallback'
 
     def __init__(self, settings):
         self._plugins = {}
         self._fallback_plugin = settings.get(self.fallback_plugin_key, None)
 
     def add_authn_plugin(self, config, plugin):
         plugin_id = plugin.get_id()
         if plugin_id in self._plugins.keys():
             raise ConfigurationError(
                 'Cannot register authentication plugin twice: "%s"', plugin_id)
         else:
             log.debug('Register authentication plugin: "%s"', plugin_id)
             self._plugins[plugin_id] = plugin
 
     def get_plugins(self):
         def sort_key(plugin):
             return str.lower(safe_str(plugin.get_display_name()))
 
         return sorted(self._plugins.values(), key=sort_key)
 
     def get_plugin(self, plugin_id):
         return self._plugins.get(plugin_id, None)
 
     def get_plugin_by_uid(self, plugin_uid):
         for plugin in self._plugins.values():
             if plugin.uid == plugin_uid:
                 return plugin
 
     def get_cache_call_method(self, cache=True):
         region, _ns = self.get_cache_region()
 
         @region.conditional_cache_on_arguments(condition=cache)
         def _get_auth_plugins(name: str, key: str, fallback_plugin):
             log.debug('auth-plugins: calculating plugins available for authentication')
 
             _plugins = []
             # Add all enabled and active plugins to the list. We iterate over the
             # auth_plugins setting from DB because it also represents the ordering.
             enabled_plugins = SettingsModel().get_auth_plugins()
             raw_settings = SettingsModel().get_all_settings(cache=False)
 
             for plugin_id in enabled_plugins:
                 plugin = self.get_plugin(plugin_id)
                 if plugin is not None and plugin.is_active(
                         plugin_cached_settings=raw_settings):
 
                     # inject settings into plugin, we can re-use the DB fetched settings here
                     plugin._settings = plugin._propagate_settings(raw_settings)
                     _plugins.append(plugin)
 
             # Add the fallback plugin from ini file.
             if fallback_plugin:
                 log.warning(
                     'Using fallback authentication plugin from INI file: "%s"',
                     fallback_plugin)
                 plugin = self.get_plugin(fallback_plugin)
                 if plugin is not None and plugin not in _plugins:
                     plugin._settings = plugin._propagate_settings(raw_settings)
                     _plugins.append(plugin)
             return _plugins
 
         return _get_auth_plugins
 
     def get_plugins_for_authentication(self, cache=True):
         """
         Returns a list of plugins which should be consulted when authenticating
         a user. It only returns plugins which are enabled and active.
         Additionally, it includes the fallback plugin from the INI file, if
         `rhodecode.auth_plugin_fallback` is set to a plugin ID.
         """
 
         _get_auth_plugins = self.get_cache_call_method(cache=cache)
 
         start = time.time()
         plugins = _get_auth_plugins('rhodecode_auth_plugins', 'v1', self._fallback_plugin)
 
         compute_time = time.time() - start
         log.debug('cached method:%s took %.4fs', _get_auth_plugins.__name__, compute_time)
 
         statsd = StatsdClient.statsd
         if statsd:
             elapsed_time_ms = round(1000.0 * compute_time)  # use ms only
             statsd.timing("rhodecode_auth_plugins_timing.histogram", elapsed_time_ms,
                           use_decimals=False)
 
         return plugins
 
     @classmethod
     def get_cache_region(cls):
         cache_namespace_uid = 'auth_plugins'
         region = rc_cache.get_or_create_region('cache_general', cache_namespace_uid)
         return region, cache_namespace_uid
 
     @classmethod
     def invalidate_auth_plugins_cache(cls, hard=True):
         region, namespace_key = cls.get_cache_region()
         log.debug('Invalidation cache [%s] region %s for cache_key: %s',
                   'invalidate_auth_plugins_cache', region, namespace_key)
 
         # we use hard cleanup if invalidation is sent
         rc_cache.clear_cache_namespace(region, namespace_key, method=rc_cache.CLEAR_DELETE)

rhodecode/authentication/routes.py

@@ -1,155 +1,153 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 logging
 import collections
 
 from pyramid.exceptions import ConfigurationError
 
 from rhodecode.lib.utils2 import safe_str
 from rhodecode.model.settings import SettingsModel
 from rhodecode.translation import _
 
 
 log = logging.getLogger(__name__)
 
 
 class AuthnResourceBase(object):
     __name__ = None
     __parent__ = None
 
     def get_root(self):
         current = self
         while current.__parent__ is not None:
             current = current.__parent__
         return current
 
 
 class AuthnPluginResourceBase(AuthnResourceBase):
 
     def __init__(self, plugin):
         self.plugin = plugin
         self.__name__ = plugin.get_url_slug()
         self.display_name = plugin.get_display_name()
 
 
 class AuthnRootResource(AuthnResourceBase):
     """
     This is the root traversal resource object for the authentication settings.
     """
 
     def __init__(self):
         self._store = collections.OrderedDict()
         self._resource_name_map = {}
         self.display_name = _('Authentication Plugins')
 
     def __getitem__(self, key):
         """
         Customized get item function to return only items (plugins) that are
         activated.
         """
         if self._is_item_active(key):
             return self._store[key]
         else:
             raise KeyError('Authentication plugin "{}" is not active.'.format(
                 key))
 
     def __iter__(self):
         for key in self._store.keys():
             if self._is_item_active(key):
                 yield self._store[key]
 
     def _is_item_active(self, key):
         activated_plugins = SettingsModel().get_auth_plugins()
         plugin_id = self.get_plugin_id(key)
         return plugin_id in activated_plugins
 
     def get_plugin_id(self, resource_name):
         """
         Return the plugin id for the given traversal resource name.
         """
         # TODO: Store this info in the resource element.
         return self._resource_name_map[resource_name]
 
     def get_sorted_list(self, sort_key=None):
         """
         Returns a sorted list of sub resources for displaying purposes.
         """
         def default_sort_key(resource):
             return str.lower(safe_str(resource.display_name))
 
         active = [item for item in self]
         return sorted(active, key=sort_key or default_sort_key)
 
     def get_nav_list(self, sort=True):
         """
         Returns a sorted list of resources for displaying the navigation.
         """
         if sort:
             nav_list = self.get_sorted_list()
         else:
             nav_list = [item for item in self]
 
         nav_list.insert(0, self)
         return nav_list
 
     def add_authn_resource(self, config, plugin_id, resource):
         """
         Register a traversal resource as a sub element to the authentication
         settings. This method is registered as a directive on the pyramid
         configurator object and called by plugins.
         """
 
         def _ensure_unique_name(name, limit=100):
             counter = 1
             current = name
             while current in self._store.keys():
-                current = '{}{}'.format(name, counter)
+                current = f'{name}{counter}'
                 counter += 1
                 if counter > limit:
                     raise ConfigurationError(
                         'Cannot build unique name for traversal resource "%s" '
                         'registered by plugin "%s"', name, plugin_id)
             return current
 
         # Allow plugin resources with identical names by rename duplicates.
         unique_name = _ensure_unique_name(resource.__name__)
         if unique_name != resource.__name__:
             log.warning('Name collision for traversal resource "%s" registered '
                         'by authentication plugin "%s"', resource.__name__,
                         plugin_id)
             resource.__name__ = unique_name
 
         log.debug('Register traversal resource "%s" for plugin "%s"',
                   unique_name, plugin_id)
         self._resource_name_map[unique_name] = plugin_id
         resource.__parent__ = self
         self._store[unique_name] = resource
 
 
 root = AuthnRootResource()
 
 
 def root_factory(request=None):
     """
     Returns the root traversal resource instance used for the authentication
     settings route.
     """
     return root

rhodecode/authentication/schema.py

@@ -1,52 +1,50 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
 
 from rhodecode.authentication import plugin_default_auth_ttl
 from rhodecode.translation import _
 
 
 class AuthnPluginSettingsSchemaBase(colander.MappingSchema):
     """
     This base schema is intended for use in authentication plugins.
     It adds a few default settings (e.g., "enabled"), so that plugin
     authors don't have to maintain a bunch of boilerplate.
     """
     enabled = colander.SchemaNode(
         colander.Bool(),
         default=False,
         description=_('Enable or disable this authentication plugin.'),
         missing=False,
         title=_('Enabled'),
         widget='bool',
     )
     cache_ttl = colander.SchemaNode(
         colander.Int(),
         default=plugin_default_auth_ttl,
         description=_('Amount of seconds to cache the authentication and '
                       'permissions check response call for this plugin. \n'
                       'Useful for expensive calls like LDAP to improve the '
                       'performance of the system (0 means disabled).'),
         missing=0,
         title=_('Auth Cache TTL'),
         validator=colander.Range(min=0, max=None),
         widget='int',
     )

rhodecode/authentication/tests/conftest.py

@@ -1,89 +1,87 b''
-
-
 # Copyright (C) 2016-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 pytest
 
 
 class EnabledAuthPlugin(object):
     """
     Context manager that updates the 'auth_plugins' setting in DB to enable
     a plugin. Previous setting is restored on exit. The rhodecode auth plugin
     is also enabled because it is needed to login the test users.
     """
 
     def __init__(self, plugin):
         self.new_value = {'egg:rhodecode-enterprise-ce#rhodecode', plugin.get_id()}
 
     def __enter__(self):
         from rhodecode.model.settings import SettingsModel
         self._old_value = SettingsModel().get_auth_plugins()
         SettingsModel().create_or_update_setting(
             'auth_plugins', ','.join(self.new_value))
 
     def __exit__(self, type, value, traceback):
         from rhodecode.model.settings import SettingsModel
         SettingsModel().create_or_update_setting(
             'auth_plugins', ','.join(self._old_value))
 
 
 class DisabledAuthPlugin(object):
     """
     Context manager that updates the 'auth_plugins' setting in DB to disable
     a plugin. Previous setting is restored on exit.
     """
 
     def __init__(self, plugin):
         self.plugin_id = plugin.get_id()
 
     def __enter__(self):
         from rhodecode.model.settings import SettingsModel
         self._old_value = SettingsModel().get_auth_plugins()
         new_value = [id_ for id_ in self._old_value if id_ != self.plugin_id]
         SettingsModel().create_or_update_setting(
             'auth_plugins', ','.join(new_value))
 
     def __exit__(self, type, value, traceback):
         from rhodecode.model.settings import SettingsModel
         SettingsModel().create_or_update_setting(
             'auth_plugins', ','.join(self._old_value))
 
 
 @pytest.fixture(params=[
     ('rhodecode.authentication.plugins.auth_crowd', 'egg:rhodecode-enterprise-ce#crowd'),
     ('rhodecode.authentication.plugins.auth_headers', 'egg:rhodecode-enterprise-ce#headers'),
     ('rhodecode.authentication.plugins.auth_jasig_cas', 'egg:rhodecode-enterprise-ce#jasig_cas'),
     ('rhodecode.authentication.plugins.auth_ldap', 'egg:rhodecode-enterprise-ce#ldap'),
     ('rhodecode.authentication.plugins.auth_pam', 'egg:rhodecode-enterprise-ce#pam'),
     ('rhodecode.authentication.plugins.auth_rhodecode', 'egg:rhodecode-enterprise-ce#rhodecode'),
     ('rhodecode.authentication.plugins.auth_token', 'egg:rhodecode-enterprise-ce#token'),
 ])
 def auth_plugin(request):
     """
     Fixture that provides instance for each authentication plugin. These
     instances are NOT the instances which are registered to the authentication
     registry.
     """
     from importlib import import_module
 
     # Create plugin instance.
     module, plugin_id = request.param
     plugin_module = import_module(module)
     return plugin_module.plugin_factory(plugin_id)

rhodecode/authentication/tests/functional/test_settings.py

@@ -1,77 +1,75 b''
-
-
 # Copyright (C) 2016-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 pytest
 
 from rhodecode.authentication.tests.conftest import (
     EnabledAuthPlugin, DisabledAuthPlugin)
 from rhodecode.apps._base import ADMIN_PREFIX
 
 
 @pytest.mark.usefixtures('autologin_user', 'app')
 class TestAuthenticationSettings:
 
     def test_auth_settings_global_view_get(self, app):
-        url = '{prefix}/auth/'.format(prefix=ADMIN_PREFIX)
+        url = f'{ADMIN_PREFIX}/auth/'
         response = app.get(url)
         assert response.status_code == 200
 
     def test_plugin_settings_view_get(self, app, auth_plugin):
         url = '{prefix}/auth/{name}'.format(
             prefix=ADMIN_PREFIX,
             name=auth_plugin.name)
         with EnabledAuthPlugin(auth_plugin):
             response = app.get(url)
             assert response.status_code == 200
 
     def test_plugin_settings_view_post(self, app, auth_plugin, csrf_token):
         url = '{prefix}/auth/{name}'.format(
             prefix=ADMIN_PREFIX,
             name=auth_plugin.name)
         params = {
             'enabled': True,
             'cache_ttl': 0,
             'csrf_token': csrf_token,
         }
         with EnabledAuthPlugin(auth_plugin):
             response = app.post(url, params=params)
             assert response.status_code in [200, 302]
 
     def test_plugin_settings_view_get_404(self, app, auth_plugin):
         url = '{prefix}/auth/{name}'.format(
             prefix=ADMIN_PREFIX,
             name=auth_plugin.name)
         with DisabledAuthPlugin(auth_plugin):
             response = app.get(url, status=404)
             assert response.status_code == 404
 
     def test_plugin_settings_view_post_404(self, app, auth_plugin, csrf_token):
         url = '{prefix}/auth/{name}'.format(
             prefix=ADMIN_PREFIX,
             name=auth_plugin.name)
         params = {
             'enabled': True,
             'cache_ttl': 0,
             'csrf_token': csrf_token,
         }
         with DisabledAuthPlugin(auth_plugin):
             response = app.post(url, params=params, status=404)
             assert response.status_code == 404

rhodecode/authentication/views.py

@@ -1,180 +1,178 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()
 
         # 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')
 
         redirect_to = self.request.resource_path(self.context, route_name='auth_home')
 
         return HTTPFound(redirect_to)