PYTHONPATH=./ ./tests/tests.py --test=testnameid
PYTHONPATH=./ ./tests/tests.py --test=testrest
PYTHONPATH=./ ./tests/tests.py --test=testmapping
+ PYTHONPATH=./ ./tests/tests.py --test=testgssapi
PYTHONPATH=./ ./tests/tests.py --test=attrs
PYTHONPATH=./ ./tests/tests.py --test=trans
PYTHONPATH=./ ./tests/tests.py --test=pgdb
import subprocess
+WRAP_HOSTNAME = 'idp.ipsilon.dev'
+TESTREALM = 'IPSILON.DEV'
+TESTDOMAIN = 'ipsilon.dev'
+KDC_DBNAME = 'db.file'
+KDC_STASH = 'stash.file'
+KDC_PASSWORD = 'ipsilon'
+KRB5_CONF_TEMPLATE = '''
+[libdefaults]
+ default_realm = ${TESTREALM}
+ dns_lookup_realm = false
+ dns_lookup_kdc = false
+ rdns = false
+ ticket_lifetime = 24h
+ forwardable = yes
+ default_ccache_name = FILE://${TESTDIR}/ccaches/krb5_ccache_XXXXXX
+ udp_preference_limit = 0
+
+[realms]
+ ${TESTREALM} = {
+ kdc =${WRAP_HOSTNAME}
+ }
+
+[domain_realm]
+ .${TESTDOMAIN} = ${TESTREALM}
+ ${TESTDOMAIN} = ${TESTREALM}
+
+[dbmodules]
+ ${TESTREALM} = {
+ database_name = ${KDCDIR}/${KDC_DBNAME}
+ }
+'''
+
+KDC_CONF_TEMPLATE = '''
+[kdcdefaults]
+ kdc_ports = 88
+ kdc_tcp_ports = 88
+ restrict_anonymous_to_tgt = true
+
+[realms]
+ ${TESTREALM} = {
+ master_key_type = aes256-cts
+ max_life = 7d
+ max_renewable_life = 14d
+ acl_file = ${KDCDIR}/kadm5.acl
+ dict_file = /usr/share/dict/words
+ default_principal_flags = +preauth
+ admin_keytab = ${TESTREALM}/kadm5.keytab
+ key_stash_file = ${KDCDIR}/${KDC_STASH}
+ }
+[logging]
+ kdc = FILE:${KDCLOG}
+'''
+
+USER_KTNAME = "user.keytab"
+HTTP_KTNAME = "http.keytab"
+KEY_TYPE = "aes256-cts-hmac-sha1-96:normal"
+
+
class IpsilonTestBase(object):
def __init__(self, name, execname):
'TESTDIR': self.testdir,
'ROOTDIR': self.rootdir,
'NAMEID': nameid,
+ 'HTTP_KTNAME': HTTP_KTNAME,
'TEST_USER': self.testuser})
filename = os.path.join(self.testdir, '%s_profile.cfg' % name)
env=env, preexec_fn=os.setsid)
self.processes.append(p)
+ def setup_kdc(self, env):
+
+ # setup kerberos environment
+ testlog = os.path.join(self.testdir, 'kerb.log')
+ krb5conf = os.path.join(self.testdir, 'krb5.conf')
+ kdcconf = os.path.join(self.testdir, 'kdc.conf')
+ kdcdir = os.path.join(self.testdir, 'kdc')
+ if os.path.exists(kdcdir):
+ shutil.rmtree(kdcdir)
+ os.makedirs(kdcdir)
+
+ t = Template(KRB5_CONF_TEMPLATE)
+ text = t.substitute({'TESTREALM': TESTREALM,
+ 'TESTDOMAIN': TESTDOMAIN,
+ 'TESTDIR': self.testdir,
+ 'KDCDIR': kdcdir,
+ 'KDC_DBNAME': KDC_DBNAME,
+ 'WRAP_HOSTNAME': WRAP_HOSTNAME})
+ with open(krb5conf, 'w+') as f:
+ f.write(text)
+
+ t = Template(KDC_CONF_TEMPLATE)
+ text = t.substitute({'TESTREALM': TESTREALM,
+ 'KDCDIR': kdcdir,
+ 'KDCLOG': testlog,
+ 'KDC_STASH': KDC_STASH})
+ with open(kdcconf, 'w+') as f:
+ f.write(text)
+
+ kdcenv = {'PATH': '/sbin:/bin:/usr/sbin:/usr/bin',
+ 'KRB5_CONFIG': krb5conf,
+ 'KRB5_KDC_PROFILE': kdcconf}
+ kdcenv.update(env)
+
+ with (open(testlog, 'a')) as logfile:
+ ksetup = subprocess.Popen(["kdb5_util", "create", "-s",
+ "-r", TESTREALM, "-P", KDC_PASSWORD],
+ stdout=logfile, stderr=logfile,
+ env=kdcenv, preexec_fn=os.setsid)
+ ksetup.wait()
+ if ksetup.returncode != 0:
+ raise ValueError('KDC Setup failed')
+
+ kdcproc = subprocess.Popen(['krb5kdc', '-n'],
+ env=kdcenv, preexec_fn=os.setsid)
+ self.processes.append(kdcproc)
+
+ return kdcenv
+
+ def kadmin_local(self, cmd, env, logfile):
+ ksetup = subprocess.Popen(["kadmin.local", "-q", cmd],
+ stdout=logfile, stderr=logfile,
+ env=env, preexec_fn=os.setsid)
+ ksetup.wait()
+ if ksetup.returncode != 0:
+ raise ValueError('Kadmin local [%s] failed' % cmd)
+
+ def setup_keys(self, env):
+
+ testlog = os.path.join(self.testdir, 'kerb.log')
+
+ svc_name = "HTTP/%s" % WRAP_HOSTNAME
+ svc_keytab = os.path.join(self.testdir, HTTP_KTNAME)
+ cmd = "addprinc -randkey -e %s %s" % (KEY_TYPE, svc_name)
+ with (open(testlog, 'a')) as logfile:
+ self.kadmin_local(cmd, env, logfile)
+ cmd = "ktadd -k %s -e %s %s" % (svc_keytab, KEY_TYPE, svc_name)
+ with (open(testlog, 'a')) as logfile:
+ self.kadmin_local(cmd, env, logfile)
+
+ usr_keytab = os.path.join(self.testdir, USER_KTNAME)
+ cmd = "addprinc -randkey -e %s %s" % (KEY_TYPE, self.testuser)
+ with (open(testlog, 'a')) as logfile:
+ self.kadmin_local(cmd, env, logfile)
+ cmd = "ktadd -k %s -e %s %s" % (usr_keytab, KEY_TYPE, self.testuser)
+ with (open(testlog, 'a')) as logfile:
+ self.kadmin_local(cmd, env, logfile)
+
+ keys_env = {"KRB5_KTNAME": svc_keytab}
+ keys_env.update(env)
+
+ return keys_env
+
+ def kinit_keytab(self, kdcenv):
+ testlog = os.path.join(self.testdir, 'kinit.log')
+ usr_keytab = os.path.join(self.testdir, USER_KTNAME)
+ kdcenv['KRB5CCNAME'] = 'FILE:' + os.path.join(
+ self.testdir, 'ccaches/user')
+ with (open(testlog, 'a')) as logfile:
+ logfile.write("\n%s\n" % kdcenv)
+ ksetup = subprocess.Popen(["kinit", "-kt", usr_keytab,
+ self.testuser],
+ stdout=logfile, stderr=logfile,
+ env=kdcenv, preexec_fn=os.setsid)
+ ksetup.wait()
+ if ksetup.returncode != 0:
+ raise ValueError('kinit %s failed' % self.testuser)
+
def wait(self):
for p in self.processes:
os.killpg(p.pid, signal.SIGTERM)
import urlparse
import json
from urllib import urlencode
+from requests_kerberos import HTTPKerberosAuth, OPTIONAL
class WrongPage(Exception):
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:
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)
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))
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'])
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}
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))
ServerRoot "${HTTPROOT}"
-ServerName localhost
+ServerName idp.ipsilon.dev
Listen ${HTTPADDR}:${HTTPPORT}
LoadModule access_compat_module modules/mod_access_compat.so
LoadModule mpm_prefork_module modules/mod_mpm_prefork.so
LoadModule wsgi_module modules/mod_wsgi.so
+LoadModule auth_gssapi_module modules/mod_auth_gssapi.so
LoadModule auth_mellon_module modules/mod_auth_mellon.so
+
<Directory />
AllowOverride none
Require all denied
--- /dev/null
+#!/usr/bin/python
+# Copyright (C) 2015 Ipsilon Project Contributors
+
+
+from helpers.common import IpsilonTestBase # pylint: disable=relative-import
+from helpers.common import WRAP_HOSTNAME # pylint: disable=relative-import
+from helpers.http import HttpSessions # pylint: disable=relative-import
+import os
+import pwd
+import sys
+from string import Template
+
+idp_g = {'TEMPLATES': '${TESTDIR}/templates/install',
+ 'CONFDIR': '${TESTDIR}/etc',
+ 'DATADIR': '${TESTDIR}/lib',
+ 'HTTPDCONFD': '${TESTDIR}/${NAME}/conf.d',
+ 'STATICDIR': '${ROOTDIR}',
+ 'BINDIR': '${ROOTDIR}/ipsilon',
+ 'WSGI_SOCKET_PREFIX': '${TESTDIR}/${NAME}/logs/wsgi'}
+
+
+idp_a = {'hostname': '${ADDRESS}:${PORT}',
+ 'admin_user': '${TEST_USER}',
+ 'system_user': '${TEST_USER}',
+ 'instance': '${NAME}',
+ 'secure': 'no',
+ 'testauth': 'yes',
+ 'pam': 'no',
+ 'gssapi': 'yes',
+ 'ipa': 'no',
+ 'gssapi_httpd_keytab': '${TESTDIR}/${HTTP_KTNAME}',
+ 'server_debugging': 'True'}
+
+
+sp_g = {'HTTPDCONFD': '${TESTDIR}/${NAME}/conf.d',
+ 'SAML2_TEMPLATE': '${TESTDIR}/templates/install/saml2/sp.conf',
+ 'SAML2_CONFFILE': '${TESTDIR}/${NAME}/conf.d/ipsilon-saml.conf',
+ 'SAML2_HTTPDIR': '${TESTDIR}/${NAME}/saml2'}
+
+
+sp_a = {'hostname': '${ADDRESS}:${PORT}',
+ 'saml_idp_metadata':
+ 'http://%s:45080/idp1/saml2/metadata' % WRAP_HOSTNAME,
+ 'saml_secure_setup': 'False',
+ 'saml_auth': '/sp',
+ 'httpd_user': '${TEST_USER}'}
+
+sp2_g = {'HTTPDCONFD': '${TESTDIR}/${NAME}/conf.d',
+ 'SAML2_TEMPLATE': '${TESTDIR}/templates/install/saml2/sp.conf',
+ 'SAML2_CONFFILE': '${TESTDIR}/${NAME}/conf.d/ipsilon-saml.conf',
+ 'SAML2_HTTPDIR': '${TESTDIR}/${NAME}/saml2'}
+
+sp2_a = {'hostname': '${ADDRESS}:${PORT}',
+ 'saml_idp_url': 'http://idp.ipsilon.dev:45080/idp1',
+ 'admin_user': '${TEST_USER}',
+ 'admin_password': '${TESTDIR}/pw.txt',
+ 'saml_sp_name': 'sp2',
+ 'saml_secure_setup': 'False',
+ 'saml_auth': '/sp',
+ 'httpd_user': '${TEST_USER}'}
+
+
+def fixup_sp_httpd(httpdir):
+ location = """
+
+Alias /sp ${HTTPDIR}/sp
+
+<Directory ${HTTPDIR}/sp>
+ Require all granted
+</Directory>
+"""
+ index = """WORKS!"""
+
+ t = Template(location)
+ text = t.substitute({'HTTPDIR': httpdir})
+ with open(httpdir + '/conf.d/ipsilon-saml.conf', 'a') as f:
+ f.write(text)
+
+ os.mkdir(httpdir + '/sp')
+ with open(httpdir + '/sp/index.html', 'w') as f:
+ f.write(index)
+
+
+class IpsilonTest(IpsilonTestBase):
+
+ def __init__(self):
+ super(IpsilonTest, self).__init__('testgssapi', __file__)
+
+ def setup_servers(self, env=None):
+ os.mkdir("%s/ccaches" % self.testdir)
+
+ print "Installing KDC server"
+ kdcenv = self.setup_kdc(env)
+
+ print "Creating principals and keytabs"
+ self.setup_keys(kdcenv)
+
+ print "Getting a TGT"
+ self.kinit_keytab(kdcenv)
+
+ print "Installing IDP server"
+ name = 'idp1'
+ addr = 'idp.ipsilon.dev'
+ port = '45080'
+ env.update(kdcenv)
+ idp = self.generate_profile(idp_g, idp_a, name, addr, port)
+ conf = self.setup_idp_server(idp, name, addr, port, env)
+
+ print "Starting IDP's httpd server"
+ self.start_http_server(conf, env)
+
+ print "Installing first SP server"
+ name = 'sp1'
+ addr = '127.0.0.11'
+ port = '45081'
+ sp = self.generate_profile(sp_g, sp_a, name, addr, port)
+ conf = self.setup_sp_server(sp, name, addr, port, env)
+ fixup_sp_httpd(os.path.dirname(conf))
+
+ print "Starting first SP's httpd server"
+ self.start_http_server(conf, env)
+
+ print "Installing second SP server"
+ name = 'sp2'
+ addr = '127.0.0.11'
+ port = '45082'
+ sp = self.generate_profile(sp2_g, sp2_a, name, addr, port)
+ with open(os.path.dirname(sp) + '/pw.txt', 'a') as f:
+ f.write('ipsilon')
+ conf = self.setup_sp_server(sp, name, addr, port, env)
+ os.remove(os.path.dirname(sp) + '/pw.txt')
+ fixup_sp_httpd(os.path.dirname(conf))
+
+ print "Starting second SP's httpd server"
+ self.start_http_server(conf, env)
+
+if __name__ == '__main__':
+
+ idpname = 'idp1'
+ sp1name = 'sp1'
+ sp2name = 'sp2'
+ user = pwd.getpwuid(os.getuid())[0]
+
+ testdir = os.environ['TESTDIR']
+
+ krb5conf = os.path.join(testdir, 'krb5.conf')
+ kenv = {'PATH': '/sbin:/bin:/usr/sbin:/usr/bin',
+ 'KRB5_CONFIG': krb5conf,
+ 'KRB5CCNAME': 'FILE:' + os.path.join(testdir, 'ccaches/user')}
+
+ for key in kenv:
+ os.environ[key] = kenv[key]
+
+ sess = HttpSessions()
+ sess.add_server(idpname, 'http://%s:45080' % WRAP_HOSTNAME, user,
+ 'ipsilon')
+ sess.add_server(sp1name, 'http://127.0.0.11:45081')
+ sess.add_server(sp2name, 'http://127.0.0.11:45082')
+
+ print "testgssapi: Authenticate to IDP ...",
+ try:
+ sess.auth_to_idp(idpname, krb=True)
+ except Exception, e: # pylint: disable=broad-except
+ print >> sys.stderr, " ERROR: %s" % repr(e)
+ sys.exit(1)
+ print " SUCCESS"
+
+ print "testgssapi: Add first SP Metadata to IDP ...",
+ try:
+ sess.add_sp_metadata(idpname, sp1name)
+ except Exception, e: # pylint: disable=broad-except
+ print >> sys.stderr, " ERROR: %s" % repr(e)
+ sys.exit(1)
+ print " SUCCESS"
+
+ print "testgssapi: Access first SP Protected Area ...",
+ try:
+ 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)
+ print " SUCCESS"
+
+ print "testgssapi: Access second SP Protected Area ...",
+ try:
+ page = sess.fetch_page(idpname, 'http://127.0.0.11:45082/sp/')
+ page.expected_value('text()', 'WORKS!')
+ except ValueError, e:
+ print >> sys.stderr, " ERROR: %s" % repr(e)
+ sys.exit(1)
+ print " SUCCESS"
import subprocess
import time
import traceback
+from helpers.common import WRAP_HOSTNAME # pylint: disable=relative-import
logger = None
else:
raise ValueError('Socket Wrappers not available')
+ pkgcfg = subprocess.Popen(['pkg-config', '--exists', 'nss_wrapper'])
+ pkgcfg.wait()
+ if pkgcfg.returncode != 0:
+ if wrappers == 'auto':
+ return {}
+ else:
+ raise ValueError('Nss Wrappers not available')
+
wrapdir = os.path.join(base, 'wrapdir')
os.mkdir(wrapdir)
- wenv = {'LD_PRELOAD': 'libsocket_wrapper.so',
+ hosts_file = os.path.join(base, 'hosts')
+ with open(hosts_file, 'w+') as f:
+ f.write('127.0.0.9 %s\n' % WRAP_HOSTNAME)
+
+ wenv = {'LD_PRELOAD': 'libsocket_wrapper.so libnss_wrapper.so',
'SOCKET_WRAPPER_DIR': wrapdir,
- 'SOCKET_WRAPPER_DEFAULT_IFACE': '9'}
+ 'SOCKET_WRAPPER_DEFAULT_IFACE': '9',
+ 'SOCKET_WRAPPER_DEBUGLEVEL': '1',
+ 'NSS_WRAPPER_HOSTNAME': WRAP_HOSTNAME,
+ 'NSS_WRAPPER_HOSTS': hosts_file}
return wenv
env = try_wrappers(test.testdir, args['wrappers'])
env['PYTHONPATH'] = test.rootdir
+ env['TESTDIR'] = test.testdir
try:
test.setup_servers(env)
projects / cascardo / ipsilon.git / commitdiff