tests/helpers/common.py

   1 #!/usr/bin/python
   2 #
   3 # Copyright (C) 2014 Ipsilon project Contributors, for license see COPYING
   4
   5 import ConfigParser
   6 import io
   7 import os
   8 import pwd
   9 import shutil
  10 import signal
  11 import random
  12 from string import Template
  13 import subprocess
  14
  15
  16 WRAP_HOSTNAME = 'idp.ipsilon.dev'
  17 TESTREALM = 'IPSILON.DEV'
  18 TESTDOMAIN = 'ipsilon.dev'
  19 KDC_DBNAME = 'db.file'
  20 KDC_STASH = 'stash.file'
  21 KDC_PASSWORD = 'ipsilon'
  22 KRB5_CONF_TEMPLATE = '''
  23 [libdefaults]
  24   default_realm = ${TESTREALM}
  25   dns_lookup_realm = false
  26   dns_lookup_kdc = false
  27   rdns = false
  28   ticket_lifetime = 24h
  29   forwardable = yes
  30   default_ccache_name = FILE://${TESTDIR}/ccaches/krb5_ccache_XXXXXX
  31   udp_preference_limit = 0
  32
  33 [realms]
  34   ${TESTREALM} = {
  35     kdc =${WRAP_HOSTNAME}
  36   }
  37
  38 [domain_realm]
  39   .${TESTDOMAIN} = ${TESTREALM}
  40   ${TESTDOMAIN} = ${TESTREALM}
  41
  42 [dbmodules]
  43   ${TESTREALM} = {
  44     database_name = ${KDCDIR}/${KDC_DBNAME}
  45   }
  46 '''
  47
  48 KDC_CONF_TEMPLATE = '''
  49 [kdcdefaults]
  50  kdc_ports = 88
  51  kdc_tcp_ports = 88
  52  restrict_anonymous_to_tgt = true
  53
  54 [realms]
  55  ${TESTREALM} = {
  56   master_key_type = aes256-cts
  57   max_life = 7d
  58   max_renewable_life = 14d
  59   acl_file = ${KDCDIR}/kadm5.acl
  60   dict_file = /usr/share/dict/words
  61   default_principal_flags = +preauth
  62   admin_keytab = ${TESTREALM}/kadm5.keytab
  63   key_stash_file = ${KDCDIR}/${KDC_STASH}
  64  }
  65 [logging]
  66   kdc = FILE:${KDCLOG}
  67 '''
  68
  69 USER_KTNAME = "user.keytab"
  70 HTTP_KTNAME = "http.keytab"
  71 KEY_TYPE = "aes256-cts-hmac-sha1-96:normal"
  72
  73
  74 class IpsilonTestBase(object):
  75
  76     def __init__(self, name, execname):
  77         self.name = name
  78         self.execname = execname
  79         self.rootdir = os.getcwd()
  80         self.testdir = None
  81         self.testuser = pwd.getpwuid(os.getuid())[0]
  82         self.processes = []
  83
  84     def force_remove(self, op, name, info):
  85         os.chmod(name, 0700)
  86         os.remove(name)
  87
  88     def setup_base(self, path, test):
  89         self.testdir = os.path.join(path, test.name)
  90         if os.path.exists(self.testdir):
  91             shutil.rmtree(self.testdir, onerror=self.force_remove)
  92         os.makedirs(self.testdir)
  93         shutil.copytree(os.path.join(self.rootdir, 'templates'),
  94                         os.path.join(self.testdir, 'templates'))
  95         os.mkdir(os.path.join(self.testdir, 'etc'))
  96         os.mkdir(os.path.join(self.testdir, 'lib'))
  97         os.mkdir(os.path.join(self.testdir, 'lib', test.name))
  98         os.mkdir(os.path.join(self.testdir, 'log'))
  99
 100     def generate_profile(self, global_opts, args_opts, name, addr, port,
 101                          nameid='unspecified'):
 102         newconf = ConfigParser.ConfigParser()
 103         newconf.add_section('globals')
 104         for k in global_opts:
 105             newconf.set('globals', k, global_opts[k])
 106         newconf.add_section('arguments')
 107         for k in args_opts:
 108             newconf.set('arguments', k, args_opts[k])
 109
 110         profile = io.BytesIO()
 111         newconf.write(profile)
 112
 113         t = Template(profile.getvalue())
 114         text = t.substitute({'NAME': name, 'ADDRESS': addr, 'PORT': port,
 115                              'TESTDIR': self.testdir,
 116                              'ROOTDIR': self.rootdir,
 117                              'NAMEID': nameid,
 118                              'HTTP_KTNAME': HTTP_KTNAME,
 119                              'TEST_USER': self.testuser})
 120
 121         filename = os.path.join(self.testdir, '%s_profile.cfg' % name)
 122         with open(filename, 'wb') as f:
 123             f.write(text)
 124
 125         return filename
 126
 127     def setup_http(self, name, addr, port):
 128         httpdir = os.path.join(self.testdir, name)
 129         os.mkdir(httpdir)
 130         os.mkdir(os.path.join(httpdir, 'conf.d'))
 131         os.mkdir(os.path.join(httpdir, 'html'))
 132         os.mkdir(os.path.join(httpdir, 'logs'))
 133         os.symlink('/etc/httpd/modules', os.path.join(httpdir, 'modules'))
 134
 135         with open(os.path.join(self.rootdir, 'tests/httpd.conf')) as f:
 136             t = Template(f.read())
 137             text = t.substitute({'HTTPROOT': httpdir,
 138                                  'HTTPADDR': addr,
 139                                  'HTTPPORT': port})
 140         filename = os.path.join(httpdir, 'httpd.conf')
 141         with open(filename, 'w+') as f:
 142             f.write(text)
 143
 144         return filename
 145
 146     def setup_idp_server(self, profile, name, addr, port, env):
 147         http_conf_file = self.setup_http(name, addr, port)
 148         cmd = [os.path.join(self.rootdir,
 149                             'ipsilon/install/ipsilon-server-install'),
 150                '--config-profile=%s' % profile]
 151         subprocess.check_call(cmd, env=env)
 152         os.symlink(os.path.join(self.rootdir, 'ipsilon'),
 153                    os.path.join(self.testdir, 'lib', name, 'ipsilon'))
 154
 155         return http_conf_file
 156
 157     def setup_sp_server(self, profile, name, addr, port, env):
 158         http_conf_file = self.setup_http(name, addr, port)
 159         cmd = [os.path.join(self.rootdir,
 160                             'ipsilon/install/ipsilon-client-install'),
 161                '--config-profile=%s' % profile]
 162         subprocess.check_call(cmd, env=env)
 163
 164         return http_conf_file
 165
 166     def setup_pgdb(self, datadir, env):
 167         cmd = ['/usr/bin/pg_ctl', 'initdb', '-D', datadir]
 168         subprocess.check_call(cmd, env=env)
 169         auth = 'host all all 127.0.0.1/24 trust\n'
 170         filename = os.path.join(datadir, 'pg_hba.conf')
 171         with open(filename, 'a') as f:
 172             f.write(auth)
 173
 174     def start_http_server(self, conf, env):
 175         env['MALLOC_CHECK_'] = '3'
 176         env['MALLOC_PERTURB_'] = str(random.randint(0, 32767) % 255 + 1)
 177         p = subprocess.Popen(['/usr/sbin/httpd', '-DFOREGROUND', '-f', conf],
 178                              env=env, preexec_fn=os.setsid)
 179         self.processes.append(p)
 180         return p
 181
 182     def start_pgdb_server(self, datadir, rundir, log, addr, port, env):
 183         p = subprocess.Popen(['/usr/bin/pg_ctl', 'start', '-D', datadir, '-o',
 184                               '-c unix_socket_directories=%s -c port=%s -c \
 185                                listen_addresses=%s' % (rundir, port, addr),
 186                               '-l', log, '-w'],
 187                              env=env, preexec_fn=os.setsid)
 188         self.processes.append(p)
 189         p.wait()
 190         for d in ['adminconfig', 'users', 'transactions', 'sessions',
 191                   'saml2.sessions.db']:
 192             cmd = ['/usr/bin/createdb', '-h', addr, '-p', port, d]
 193             subprocess.check_call(cmd, env=env)
 194
 195     def setup_ldap(self, env):
 196         ldapdir = os.path.join(self.testdir, 'ldap')
 197         os.mkdir(ldapdir)
 198         with open(os.path.join(self.rootdir, 'tests/slapd.conf')) as f:
 199             t = Template(f.read())
 200             text = t.substitute({'ldapdir': ldapdir})
 201         filename = os.path.join(ldapdir, 'slapd.conf')
 202         with open(filename, 'w+') as f:
 203             f.write(text)
 204         subprocess.check_call(['/usr/sbin/slapadd', '-f', filename, '-l',
 205                                'tests/ldapdata.ldif'], env=env)
 206
 207         return filename
 208
 209     def start_ldap_server(self, conf, addr, port, env):
 210         p = subprocess.Popen(['/usr/sbin/slapd', '-d', '0', '-f', conf,
 211                              '-h', 'ldap://%s:%s' % (addr, port)],
 212                              env=env, preexec_fn=os.setsid)
 213         self.processes.append(p)
 214
 215     def setup_kdc(self, env):
 216
 217         # setup kerberos environment
 218         testlog = os.path.join(self.testdir, 'kerb.log')
 219         krb5conf = os.path.join(self.testdir, 'krb5.conf')
 220         kdcconf = os.path.join(self.testdir, 'kdc.conf')
 221         kdcdir = os.path.join(self.testdir, 'kdc')
 222         if os.path.exists(kdcdir):
 223             shutil.rmtree(kdcdir)
 224         os.makedirs(kdcdir)
 225
 226         t = Template(KRB5_CONF_TEMPLATE)
 227         text = t.substitute({'TESTREALM': TESTREALM,
 228                              'TESTDOMAIN': TESTDOMAIN,
 229                              'TESTDIR': self.testdir,
 230                              'KDCDIR': kdcdir,
 231                              'KDC_DBNAME': KDC_DBNAME,
 232                              'WRAP_HOSTNAME': WRAP_HOSTNAME})
 233         with open(krb5conf, 'w+') as f:
 234             f.write(text)
 235
 236         t = Template(KDC_CONF_TEMPLATE)
 237         text = t.substitute({'TESTREALM': TESTREALM,
 238                              'KDCDIR': kdcdir,
 239                              'KDCLOG': testlog,
 240                              'KDC_STASH': KDC_STASH})
 241         with open(kdcconf, 'w+') as f:
 242             f.write(text)
 243
 244         kdcenv = {'PATH': '/sbin:/bin:/usr/sbin:/usr/bin',
 245                   'KRB5_CONFIG': krb5conf,
 246                   'KRB5_KDC_PROFILE': kdcconf}
 247         kdcenv.update(env)
 248
 249         with (open(testlog, 'a')) as logfile:
 250             ksetup = subprocess.Popen(["kdb5_util", "create", "-s",
 251                                        "-r", TESTREALM, "-P", KDC_PASSWORD],
 252                                       stdout=logfile, stderr=logfile,
 253                                       env=kdcenv, preexec_fn=os.setsid)
 254         ksetup.wait()
 255         if ksetup.returncode != 0:
 256             raise ValueError('KDC Setup failed')
 257
 258         kdcproc = subprocess.Popen(['krb5kdc', '-n'],
 259                                    env=kdcenv, preexec_fn=os.setsid)
 260         self.processes.append(kdcproc)
 261
 262         return kdcenv
 263
 264     def kadmin_local(self, cmd, env, logfile):
 265         ksetup = subprocess.Popen(["kadmin.local", "-q", cmd],
 266                                   stdout=logfile, stderr=logfile,
 267                                   env=env, preexec_fn=os.setsid)
 268         ksetup.wait()
 269         if ksetup.returncode != 0:
 270             raise ValueError('Kadmin local [%s] failed' % cmd)
 271
 272     def setup_keys(self, env):
 273
 274         testlog = os.path.join(self.testdir, 'kerb.log')
 275
 276         svc_name = "HTTP/%s" % WRAP_HOSTNAME
 277         svc_keytab = os.path.join(self.testdir, HTTP_KTNAME)
 278         cmd = "addprinc -randkey -e %s %s" % (KEY_TYPE, svc_name)
 279         with (open(testlog, 'a')) as logfile:
 280             self.kadmin_local(cmd, env, logfile)
 281         cmd = "ktadd -k %s -e %s %s" % (svc_keytab, KEY_TYPE, svc_name)
 282         with (open(testlog, 'a')) as logfile:
 283             self.kadmin_local(cmd, env, logfile)
 284
 285         usr_keytab = os.path.join(self.testdir, USER_KTNAME)
 286         cmd = "addprinc -randkey -e %s %s" % (KEY_TYPE, self.testuser)
 287         with (open(testlog, 'a')) as logfile:
 288             self.kadmin_local(cmd, env, logfile)
 289         cmd = "ktadd -k %s -e %s %s" % (usr_keytab, KEY_TYPE, self.testuser)
 290         with (open(testlog, 'a')) as logfile:
 291             self.kadmin_local(cmd, env, logfile)
 292
 293         keys_env = {"KRB5_KTNAME": svc_keytab}
 294         keys_env.update(env)
 295
 296         return keys_env
 297
 298     def kinit_keytab(self, kdcenv):
 299         testlog = os.path.join(self.testdir, 'kinit.log')
 300         usr_keytab = os.path.join(self.testdir, USER_KTNAME)
 301         kdcenv['KRB5CCNAME'] = 'FILE:' + os.path.join(
 302             self.testdir, 'ccaches/user')
 303         with (open(testlog, 'a')) as logfile:
 304             logfile.write("\n%s\n" % kdcenv)
 305             ksetup = subprocess.Popen(["kinit", "-kt", usr_keytab,
 306                                        self.testuser],
 307                                       stdout=logfile, stderr=logfile,
 308                                       env=kdcenv, preexec_fn=os.setsid)
 309             ksetup.wait()
 310             if ksetup.returncode != 0:
 311                 raise ValueError('kinit %s failed' % self.testuser)
 312
 313     def wait(self):
 314         for p in self.processes:
 315             os.killpg(p.pid, signal.SIGTERM)
 316
 317     def setup_servers(self, env=None):
 318         raise NotImplementedError()
 319
 320     def run(self, env):
 321         exe = self.execname
 322         if exe.endswith('c'):
 323             exe = exe[:-1]
 324         return subprocess.call([exe], env=env)