@@ -86,104 +86,114 @@ class CrowdServer(object):
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
@@ -184,75 +184,76 @@ class TestAuthSettingsController(TestCon
auth_container_email_header='',
auth_container_firstname_header='',
auth_container_lastname_header='',
auth_container_fallback_header='',
auth_container_clean_username='True',
)
self._container_auth_verify_login(
extra_environ={'REMOTE_USER': 'john@example.org'},
resulting_username='john',
def test_container_auth_clean_username_backslash(self):
self._container_auth_setup(
auth_container_header='REMOTE_USER',
extra_environ={'REMOTE_USER': r'example\jane'},
resulting_username=r'jane',
def test_container_auth_no_logout(self):
response = self.app.get(
url=url(controller='admin/my_account', action='my_account'),
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: