@@ -38,200 +38,210 @@ log = logging.getLogger(__name__)
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 = urllib2.HTTPPasswordMgrWithDefaultRealm()
mgr.add_password(None, self._uri, self.user, self.passwd)
handler = urllib2.HTTPBasicAuthHandler(mgr)
self.opener = urllib2.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))
_headers["Authorization"] = "Basic %s" % authstring
if headers:
_headers.update(headers)
log.debug("Sent crowd: \n%s",
formatted_json({"url": url, "body": body,
"headers": _headers}))
req = urllib2.Request(url, body, _headers)
if method:
req.get_method = lambda: method
global msg
msg = ""
try:
rdoc = self.opener.open(req)
msg = "".join(rdoc.readlines())
if not msg and empty_response_ok:
rval = {}
rval["status"] = True
rval["error"] = "Response body was empty"
elif not noformat:
rval = json.loads(msg)
else:
rval = "".join(rdoc.readlines())
except Exception as e:
if not noformat:
rval = {"status": False,
"body": body,
"error": str(e) + "\n" + msg}
rval = None
return rval
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, urllib2.quote(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"
return self._request(url)
class KallitheaAuthPlugin(auth_modules.KallitheaExternalAuthPlugin):
def __init__(self):
self._protocol_values = ["http", "https"]
@hybrid_property
def name(self):
return "crowd"
def settings(self):
settings = [
{
"name": "method",
"validator": self.validators.OneOf(self._protocol_values),
"type": "select",
"values": self._protocol_values,
"description": "The protocol used to connect to the Atlassian CROWD server.",
"formname": "Protocol"
},
"name": "host",
"validator": self.validators.UnicodeString(strip=True),
"type": "string",
"description": "The FQDN or IP of the Atlassian CROWD Server",
"default": "127.0.0.1",
"formname": "Host"
"name": "port",
"validator": self.validators.Number(strip=True),
"type": "int",
"description": "The Port in use by the Atlassian CROWD Server",
"default": 8095,
"formname": "Port"
"name": "app_name",
"description": "The Application Name to authenticate to CROWD",
"default": "",
"formname": "Application Name"
"name": "app_password",
"description": "The password to authenticate to CROWD",
"formname": "Application Password"
"name": "admin_groups",
"description": "A comma separated list of group names that identify users as Kallithea Administrators",
"formname": "Admin Groups"
}
]
return settings
def use_fake_password(self):
return True
def 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: KallitheaAuthPluginBase.auth_func_attrs
This is later validated for correctness
if not username or not password:
log.debug('Empty username or password skipping...')
return None
log.debug("Crowd settings: \n%s", formatted_json(settings))
server = CrowdServer(**settings)
server.set_credentials(settings["app_name"], settings["app_password"])
crowd_user = server.user_auth(username, password)
log.debug("Crowd returned: \n%s", formatted_json(crowd_user))
if not crowd_user["status"]:
res = server.user_groups(crowd_user["name"])
log.debug("Crowd groups: \n%s", formatted_json(res))
crowd_user["groups"] = [x["name"] for x in res["groups"]]
# old attrs fetched from Kallithea database
admin = getattr(userobj, 'admin', False)
active = getattr(userobj, 'active', True)
email = getattr(userobj, 'email', '')
firstname = getattr(userobj, 'firstname', '')
lastname = getattr(userobj, 'lastname', '')
user_data = {
'username': crowd_user["name"] or username,
'firstname': crowd_user["first-name"] or firstname,
'lastname': crowd_user["last-name"] or lastname,
'groups': crowd_user["groups"],
'email': crowd_user["email"] or email,
'admin': admin,
'active': active,
'active_from_extern': crowd_user.get('active'), # ???
'extern_name': crowd_user["name"],
# set an admin if we're in admin_groups of crowd
for group in settings["admin_groups"].split(","):
@@ -136,123 +136,124 @@ class TestAuthSettingsController(TestCon
auth_container_firstname_header='',
auth_container_lastname_header='',
auth_container_fallback_header='',
auth_container_clean_username='False',
)
self._container_auth_verify_login(
extra_environ={'THE_USER_NAME': 'john@example.org'},
resulting_username='john@example.org',
def test_container_auth_login_header_attr(self):
self._container_auth_setup(
auth_container_header='THE_USER_NAME',
auth_container_email_header='THE_USER_EMAIL',
auth_container_firstname_header='THE_USER_FIRSTNAME',
auth_container_lastname_header='THE_USER_LASTNAME',
response = self.app.get(
url=url(controller='admin/my_account', action='my_account'),
extra_environ={'THE_USER_NAME': 'johnd',
'THE_USER_EMAIL': 'john@example.org',
'THE_USER_FIRSTNAME': 'John',
'THE_USER_LASTNAME': 'Doe',
assert response.form['email'].value == 'john@example.org'
assert response.form['firstname'].value == 'John'
assert response.form['lastname'].value == 'Doe'
def test_container_auth_login_fallback_header(self):
auth_container_email_header='',
auth_container_fallback_header='HTTP_X_YZZY',
headers={'X-Yzzy': r'foo\bar'},
resulting_username=r'foo\bar',
def test_container_auth_clean_username_at(self):
auth_container_header='REMOTE_USER',
auth_container_clean_username='True',
extra_environ={'REMOTE_USER': 'john@example.org'},
resulting_username='john',
def test_container_auth_clean_username_backslash(self):
extra_environ={'REMOTE_USER': r'example\jane'},
resulting_username=r'jane',
def test_container_auth_no_logout(self):
extra_environ={'REMOTE_USER': 'john'},
assert 'Log Out' not in response.normal_body
def test_crowd_save_settings(self):
self.log_user()
params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_crowd')
params.update({'auth_crowd_host': ' hostname ',
'auth_crowd_app_password': 'secret',
'auth_crowd_admin_groups': 'mygroup',
'auth_crowd_port': '123',
'auth_crowd_method': 'https',
'auth_crowd_app_name': 'xyzzy'})
test_url = url(controller='admin/auth_settings',
action='auth_settings')
response = self.app.post(url=test_url, params=params)
self.checkSessionFlash(response, 'Auth settings updated successfully')
new_settings = Setting.get_auth_settings()
assert new_settings['auth_crowd_host'] == u'hostname', 'fail db write compare'
@skipif(not pam_lib_installed, reason='skipping due to missing pam lib')
def test_pam_save_settings(self):
params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_pam')
params.update({'auth_pam_service': 'kallithea',
'auth_pam_gecos': '^foo-.*'})
assert new_settings['auth_pam_service'] == u'kallithea', 'fail db write compare'
Status change: