|
|
# Copyright (C) 2012-2024 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 "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().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 = f'egg:rhodecode-enterprise-ce#{RhodeCodeAuthPlugin.uid}'
|
|
|
plugin_factory(plugin_id).includeme(config)
|
|
|
|
