kallithea Changeset - 861ef58bc36e

Changeset - 861ef58bc36e

Parent rev.

Child rev.

[Not reviewed]

default

0 2 0

Chris Rule - 8 years ago 2018-05-01 17:08:36
crule@aegistg.com

auth: add https ability to the crowd auth module (issue #315)

[Thomas De Schampheleire:
- use select iso checkbox to remove need for bool->string conversion
- update tests]

2 files changed with 11 insertions and 0 deletions:

kallithea/lib/auth_modules/auth_crowd.py

kallithea/tests/functional/test_admin_auth_settings.py

0 comments (0 inline, 0 general)

kallithea/lib/auth_modules/auth_crowd.py

➞

Show inline comments

@@ @@ -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)
                 rval["status"] = True
             else:
                 rval = "".join(rdoc.readlines())
         except Exception as e:
             if not noformat:
                 rval = {"status": False,
                         "body": body,
                         "error": str(e) + "\n" + msg}
             else:
                 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"
                % (self._uri, self._version, urllib2.quote(username)))
         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",
                 "validator": self.validators.UnicodeString(strip=True),
                 "type": "string",
                 "description": "The Application Name to authenticate to CROWD",
                 "default": "",
                 "formname": "Application Name"
             },
+            {
                 "name": "app_password",
                 "validator": self.validators.UnicodeString(strip=True),
                 "type": "string",
                 "description": "The password to authenticate to CROWD",
                 "default": "",
                 "formname": "Application Password"
             },
+            {
                 "name": "admin_groups",
                 "validator": self.validators.UnicodeString(strip=True),
                 "type": "string",
                 "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

kallithea/tests/functional/test_admin_auth_settings.py

➞

Show inline comments

@@ @@ -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',
             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': r'example\jane'},
             resulting_username=r'jane',
+        )
     def test_container_auth_no_logout(self):
         self._container_auth_setup(
             auth_container_header='REMOTE_USER',
             auth_container_email_header='',
             auth_container_firstname_header='',
             auth_container_lastname_header='',
             auth_container_fallback_header='',
             auth_container_clean_username='True',
+        )
         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):
         self.log_user()
         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-.*'})
         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_pam_service'] == u'kallithea', 'fail db write compare'

0 comments (0 inline, 0 general)