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