# along with this program. If not, see <http://www.gnu.org/licenses/>.
-from lxml import html
+from helpers import http # pylint: disable=relative-import
import os
import pwd
-import requests
import sys
-import urlparse
-def get_session(srvs, url):
- for srv in srvs:
- if url.startswith(srv['baseuri']):
- return srv['session']
-
- raise ValueError("Unknown URL: %s" % url)
-
-
-def get_url(srvs, url, **kwargs):
- session = get_session(srvs, url)
- return session.get(url, allow_redirects=False, **kwargs)
-
-
-def post_url(srvs, url, **kwargs):
- session = get_session(srvs, url)
- return session.post(url, allow_redirects=False, **kwargs)
-
-
-def access_url(action, srvs, url, **kwargs):
- if action == 'get':
- return get_url(srvs, url, **kwargs)
- elif action == 'post':
- return post_url(srvs, url, **kwargs)
- else:
- raise ValueError("Unknown action type: [%s]" % action)
-
-
-def get_new_url(referer, action):
- if action.startswith('/'):
- u = urlparse.urlparse(referer)
- return '%s://%s%s' % (u.scheme, u.netloc, action)
- return action
-
-
-def parse_first(tree, rule):
- result = tree.xpath(rule)
- if type(result) is list:
- if len(result) > 0:
- result = result[0]
- else:
- result = None
- return result
-
-
-def parse_list(tree, rule):
- result = tree.xpath(rule)
- if type(result) is list:
- return result
- return [result]
-
-
-def handle_login_form(idp, r):
- tree = html.fromstring(r.text)
- try:
- action_url = parse_first(tree, '//form[@id="login_form"]/@action')
- method = parse_first(tree, '//form[@id="login_form"]/@method')
- except Exception: # pylint: disable=broad-except
- return []
-
- if action_url is None:
- return []
-
- headers = {'referer': r.url}
- payload = {'login_name': idp['user'],
- 'login_password': idp['pass']}
-
- return [method,
- get_new_url(r.url, action_url),
- {'headers': headers, 'data': payload}]
-
-
-def handle_return_form(r):
- tree = html.fromstring(r.text)
- try:
- action_url = parse_first(tree, '//form[@id="saml-response"]/@action')
- method = parse_first(tree, '//form[@id="saml-response"]/@method')
- names = parse_list(tree, '//form[@id="saml-response"]/input/@name')
- values = parse_list(tree, '//form[@id="saml-response"]/input/@value')
- except Exception: # pylint: disable=broad-except
- return []
-
- if action_url is None:
- return []
-
- headers = {'referer': r.url}
- payload = {}
- for i in range(0, len(names)):
- payload[names[i]] = values[i]
-
- return [method,
- get_new_url(r.url, action_url),
- {'headers': headers, 'data': payload}]
-
-
-def go_to_url(srvs, idp, start_url, target_url):
-
- url = start_url
- action = 'get'
- args = {}
-
- good = True
- while good:
- r = access_url(action, srvs, url, **args) # pylint: disable=star-args
- if r.status_code == 303:
- url = r.headers['location']
- action = 'get'
- args = {}
- elif r.status_code == 200:
- if url == target_url:
- return r.text
-
- result = handle_login_form(idp, r)
- if result:
- action = result[0]
- url = result[1]
- args = result[2]
- continue
-
- result = handle_return_form(r)
- if result:
- action = result[0]
- url = result[1]
- args = result[2]
- continue
-
- raise ValueError("Unhandled Success code at url %s" % url)
-
- else:
- good = False
-
- raise ValueError("Unhandled status (%d) on url %s" % (r.status_code, url))
-
-
-def auth_to_idp(idp):
-
- target_url = '%s/%s/' % (idp['baseuri'], idp['name'])
- srvs = [idp]
+if __name__ == '__main__':
+ basedir = sys.argv[1]
- r = access_url('get', srvs, target_url)
- if r.status_code != 200:
- print >> sys.stderr, " ERROR: Access to idp failed: %s" % repr(r)
- return False
+ idpname = 'idp1'
+ spname = 'sp1'
+ user = pwd.getpwuid(os.getuid())[0]
- tree = html.fromstring(r.text)
- try:
- expected = 'Log In'
- login = parse_first(tree, '//div[@id="content"]/p/a/text()')
- if login != expected:
- print >> sys.stderr, " ERROR: Expected [%s] got [%s]" % (expected,
- login)
- href = parse_first(tree, '//div[@id="content"]/p/a/@href')
- start_url = get_new_url(target_url, href)
- except Exception, e: # pylint: disable=broad-except
- print >> sys.stderr, " ERROR: Unexpected reply [%s]" % repr(e)
- return False
+ sess = http.HttpSessions()
+ sess.add_server(idpname, 'http://127.0.0.10:45080', user, 'ipsilon')
+ sess.add_server(spname, 'http://127.0.0.11:45081')
+ print "test1: Authenticate to IDP ...",
try:
- page = go_to_url(srvs, idp, start_url, target_url)
+ sess.auth_to_idp(idpname)
except Exception, e: # pylint: disable=broad-except
print >> sys.stderr, " ERROR: %s" % repr(e)
- return False
-
- tree = html.fromstring(page)
- try:
- welcome = parse_first(tree, '//div[@id="welcome"]/p/text()')
- except Exception, e: # pylint: disable=broad-except
- print >> sys.stderr, " ERROR: Unexpected reply [%s]" % repr(e)
- return False
-
- expected = 'Welcome %s!' % idp['user']
- if welcome != expected:
- print >> sys.stderr, " ERROR: Expected [%s] got [%s]" % (expected,
- welcome)
- return False
-
- return True
-
-
-def add_sp_metadata(idp, sp):
- url = '%s/%s/admin/providers/saml2/admin/new' % (idp['baseuri'],
- idp['name'])
- headers = {'referer': url}
- payload = {'name': sp['name']}
- m = requests.get('%s/saml2/metadata' % sp['baseuri'])
- metafile = {'metafile': m.content}
- r = idp['session'].post(url, headers=headers,
- data=payload, files=metafile)
- if r.status_code != 200:
- print >> sys.stderr, " ERROR: %s" % repr(r)
- return False
-
- tree = html.fromstring(r.text)
- try:
- alert = parse_first(tree,
- '//div[@class="alert alert-success"]/p/text()')
- except Exception, e: # pylint: disable=broad-except
- print >> sys.stderr, " ERROR: Unexpected reply [%s]" % repr(e)
- return False
-
- expected = 'SP Successfully added'
- if alert != expected:
- print >> sys.stderr, " ERROR: Expected [%s] got [%s]" % (expected,
- alert)
- return False
-
- return True
-
-
-if __name__ == '__main__':
- basedir = sys.argv[1]
-
- idpsrv = {'name': 'idp1',
- 'baseuri': 'http://127.0.0.10:45080',
- 'session': requests.Session(),
- 'user': pwd.getpwuid(os.getuid())[0],
- 'pass': 'ipsilon'}
- spsrv = {'name': 'sp1',
- 'baseuri': 'http://127.0.0.11:45081',
- 'session': requests.Session()}
-
- print "test1: Authenticate to IDP ...",
- if not auth_to_idp(idpsrv):
sys.exit(1)
print " SUCCESS"
print "test1: Add SP Metadata to IDP ...",
- if not add_sp_metadata(idpsrv, spsrv):
+ try:
+ sess.add_sp_metadata(idpname, spname)
+ except Exception, e: # pylint: disable=broad-except
+ print >> sys.stderr, " ERROR: %s" % repr(e)
sys.exit(1)
print " SUCCESS"
print "test1: Access SP Protected Area ...",
- servers = [idpsrv, spsrv]
- spurl = '%s/sp/' % (spsrv['baseuri'])
try:
- text = go_to_url(servers, idpsrv, spurl, spurl)
+ page = sess.fetch_page(idpname, 'http://127.0.0.11:45081/sp/')
+ page.expected_value('text()', 'WORKS!')
except ValueError, e:
print >> sys.stderr, " ERROR: %s" % repr(e)
sys.exit(1)
- if text != "WORKS!":
- print >> sys.stderr, "ERROR: Expected [WORKS!], got [%s]" % text
- sys.exit(1)
print " SUCCESS"
projects / cascardo / ipsilon.git / blobdiff