From 9cbeccde3151029d70bedb0579482e10a5cbab7b Mon Sep 17 00:00:00 2001 From: Rob Crittenden Date: Tue, 28 Apr 2015 10:52:55 -0400 Subject: Configure a KDC, add test for GSSAPI/Kerberos Using nss_wrappers so we can control host names we can setup a KDC and test GSSAPI, including fallback to forms-based auth. This also means that fetch_page() needs to handle 401 a bit better, so it can re-try a failed authentication or fall back to forms-based auth. Note for posterity: if gss_localname() fails this is likely due to using the wrong krb5.conf in Apache, so pass in all environment variables. The KDC setup code was based heavily on the tests in the gssproxy project. https://fedorahosted.org/ipsilon/ticket/116 Signed-off-by: Rob Crittenden --- tests/helpers/http.py | 54 ++++++++++++++++++++++++++++++++++++++------------- 1 file changed, 40 insertions(+), 14 deletions(-) (limited to 'tests/helpers/http.py') diff --git a/tests/helpers/http.py b/tests/helpers/http.py index ff696d4..0da7ee2 100755 --- a/tests/helpers/http.py +++ b/tests/helpers/http.py @@ -24,6 +24,7 @@ import string import urlparse import json from urllib import urlencode +from requests_kerberos import HTTPKerberosAuth, OPTIONAL class WrongPage(Exception): @@ -89,18 +90,25 @@ class HttpSessions(object): raise ValueError("Unknown URL: %s" % url) - def get(self, url, **kwargs): + def get(self, url, krb=False, **kwargs): session = self.get_session(url) - return session.get(url, allow_redirects=False, **kwargs) + allow_redirects = False + if krb: + # In at least the test instance we don't get back a negotiate + # blob to do mutual authentication against. + kerberos_auth = HTTPKerberosAuth(mutual_authentication=OPTIONAL) + kwargs['auth'] = kerberos_auth + allow_redirects = True + return session.get(url, allow_redirects=allow_redirects, **kwargs) def post(self, url, **kwargs): session = self.get_session(url) return session.post(url, allow_redirects=False, **kwargs) - def access(self, action, url, **kwargs): + def access(self, action, url, krb=False, **kwargs): action = string.lower(action) if action == 'get': - return self.get(url, **kwargs) + return self.get(url, krb, **kwargs) elif action == 'post': return self.post(url, **kwargs) else: @@ -242,19 +250,39 @@ class HttpSessions(object): return [method, self.new_url(referer, action_url), {'headers': headers, 'data': payload}] - def fetch_page(self, idp, target_url, follow_redirect=True): + def fetch_page(self, idp, target_url, follow_redirect=True, krb=False): + """ + Fetch a page and parse the response code to determine what to do + next. + + The login process consists of redirections (302/303) and + potentially an unauthorized (401). For the case of unauthorized + try the page returned in case of fallback authentication. + """ url = target_url action = 'get' args = {} while True: - r = self.access(action, url, **args) # pylint: disable=star-args + # pylint: disable=star-args + r = self.access(action, url, krb=krb, **args) if r.status_code == 303 or r.status_code == 302: if not follow_redirect: return PageTree(r) url = r.headers['location'] action = 'get' args = {} + elif r.status_code == 401: + page = PageTree(r) + if r.headers.get('WWW-Authenticate', None) is None: + return page + + # Fall back, hopefully to testauth authentication. + try: + (action, url, args) = self.handle_login_form(idp, page) + continue + except WrongPage: + pass elif r.status_code == 200: page = PageTree(r) @@ -288,12 +316,12 @@ class HttpSessions(object): raise ValueError("Unhandled status (%d) on url %s" % ( r.status_code, url)) - def auth_to_idp(self, idp): + def auth_to_idp(self, idp, krb=False): srv = self.servers[idp] target_url = '%s/%s/' % (srv['baseuri'], idp) - r = self.access('get', target_url) + r = self.access('get', target_url, krb=krb) if r.status_code != 200: raise ValueError("Access to idp failed: %s" % repr(r)) @@ -302,7 +330,8 @@ class HttpSessions(object): href = page.first_value('//div[@id="content"]/p/a/@href') url = self.new_url(target_url, href) - page = self.fetch_page(idp, url) + page = self.fetch_page(idp, url, krb=krb) + page.expected_value('//div[@id="welcome"]/p/text()', 'Welcome %s!' % srv['user']) @@ -325,7 +354,6 @@ class HttpSessions(object): def add_sp_metadata(self, idp, sp, rest=False): expected_status = 200 - idpsrv = self.servers[idp] (idpuri, m) = self.get_sp_metadata(idp, sp) url = '%s/%s/admin/providers/saml2/admin/new' % (idpuri, idp) headers = {'referer': url} @@ -334,13 +362,11 @@ class HttpSessions(object): payload = {'metadata': m.content} headers['content-type'] = 'application/x-www-form-urlencoded' url = '%s/%s/rest/providers/saml2/SPS/%s' % (idpuri, idp, sp) - r = idpsrv['session'].post(url, headers=headers, - data=urlencode(payload)) + r = self.post(url, headers=headers, data=urlencode(payload)) else: metafile = {'metafile': m.content} payload = {'name': sp} - r = idpsrv['session'].post(url, headers=headers, - data=payload, files=metafile) + r = self.post(url, headers=headers, data=payload, files=metafile) if r.status_code != expected_status: raise ValueError('Failed to post SP data [%s]' % repr(r)) -- cgit