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.keys():
 105             newconf.set('globals', k, global_opts[k])
 106         newconf.add_section('arguments')
 107         for k in args_opts.keys():
 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
 181     def start_pgdb_server(self, datadir, rundir, log, addr, port, env):
 182         p = subprocess.Popen(['/usr/bin/pg_ctl', 'start', '-D', datadir, '-o',
 183                               '-c unix_socket_directories=%s -c port=%s -c \
 184                                listen_addresses=%s' % (rundir, port, addr),
 185                               '-l', log, '-w'],
 186                              env=env, preexec_fn=os.setsid)
 187         self.processes.append(p)
 188         p.wait()
 189         for d in ['adminconfig', 'users', 'transactions', 'sessions']:
 190             cmd = ['/usr/bin/createdb', '-h', addr, '-p', port, d]
 191             subprocess.check_call(cmd, env=env)
 192
 193     def setup_ldap(self, env):
 194         ldapdir = os.path.join(self.testdir, 'ldap')
 195         os.mkdir(ldapdir)
 196         with open(os.path.join(self.rootdir, 'tests/slapd.conf')) as f:
 197             t = Template(f.read())
 198             text = t.substitute({'ldapdir': ldapdir})
 199         filename = os.path.join(ldapdir, 'slapd.conf')
 200         with open(filename, 'w+') as f:
 201             f.write(text)
 202         subprocess.check_call(['/usr/sbin/slapadd', '-f', filename, '-l',
 203                                'tests/ldapdata.ldif'], env=env)
 204
 205         return filename
 206
 207     def start_ldap_server(self, conf, addr, port, env):
 208         p = subprocess.Popen(['/usr/sbin/slapd', '-d', '0', '-f', conf,
 209                              '-h', 'ldap://%s:%s' % (addr, port)],
 210                              env=env, preexec_fn=os.setsid)
 211         self.processes.append(p)
 212
 213     def setup_kdc(self, env):
 214
 215         # setup kerberos environment
 216         testlog = os.path.join(self.testdir, 'kerb.log')
 217         krb5conf = os.path.join(self.testdir, 'krb5.conf')
 218         kdcconf = os.path.join(self.testdir, 'kdc.conf')
 219         kdcdir = os.path.join(self.testdir, 'kdc')
 220         if os.path.exists(kdcdir):
 221             shutil.rmtree(kdcdir)
 222         os.makedirs(kdcdir)
 223
 224         t = Template(KRB5_CONF_TEMPLATE)
 225         text = t.substitute({'TESTREALM': TESTREALM,
 226                              'TESTDOMAIN': TESTDOMAIN,
 227                              'TESTDIR': self.testdir,
 228                              'KDCDIR': kdcdir,
 229                              'KDC_DBNAME': KDC_DBNAME,
 230                              'WRAP_HOSTNAME': WRAP_HOSTNAME})
 231         with open(krb5conf, 'w+') as f:
 232             f.write(text)
 233
 234         t = Template(KDC_CONF_TEMPLATE)
 235         text = t.substitute({'TESTREALM': TESTREALM,
 236                              'KDCDIR': kdcdir,
 237                              'KDCLOG': testlog,
 238                              'KDC_STASH': KDC_STASH})
 239         with open(kdcconf, 'w+') as f:
 240             f.write(text)
 241
 242         kdcenv = {'PATH': '/sbin:/bin:/usr/sbin:/usr/bin',
 243                   'KRB5_CONFIG': krb5conf,
 244                   'KRB5_KDC_PROFILE': kdcconf}
 245         kdcenv.update(env)
 246
 247         with (open(testlog, 'a')) as logfile:
 248             ksetup = subprocess.Popen(["kdb5_util", "create", "-s",
 249                                        "-r", TESTREALM, "-P", KDC_PASSWORD],
 250                                       stdout=logfile, stderr=logfile,
 251                                       env=kdcenv, preexec_fn=os.setsid)
 252         ksetup.wait()
 253         if ksetup.returncode != 0:
 254             raise ValueError('KDC Setup failed')
 255
 256         kdcproc = subprocess.Popen(['krb5kdc', '-n'],
 257                                    env=kdcenv, preexec_fn=os.setsid)
 258         self.processes.append(kdcproc)
 259
 260         return kdcenv
 261
 262     def kadmin_local(self, cmd, env, logfile):
 263         ksetup = subprocess.Popen(["kadmin.local", "-q", cmd],
 264                                   stdout=logfile, stderr=logfile,
 265                                   env=env, preexec_fn=os.setsid)
 266         ksetup.wait()
 267         if ksetup.returncode != 0:
 268             raise ValueError('Kadmin local [%s] failed' % cmd)
 269
 270     def setup_keys(self, env):
 271
 272         testlog = os.path.join(self.testdir, 'kerb.log')
 273
 274         svc_name = "HTTP/%s" % WRAP_HOSTNAME
 275         svc_keytab = os.path.join(self.testdir, HTTP_KTNAME)
 276         cmd = "addprinc -randkey -e %s %s" % (KEY_TYPE, svc_name)
 277         with (open(testlog, 'a')) as logfile:
 278             self.kadmin_local(cmd, env, logfile)
 279         cmd = "ktadd -k %s -e %s %s" % (svc_keytab, KEY_TYPE, svc_name)
 280         with (open(testlog, 'a')) as logfile:
 281             self.kadmin_local(cmd, env, logfile)
 282
 283         usr_keytab = os.path.join(self.testdir, USER_KTNAME)
 284         cmd = "addprinc -randkey -e %s %s" % (KEY_TYPE, self.testuser)
 285         with (open(testlog, 'a')) as logfile:
 286             self.kadmin_local(cmd, env, logfile)
 287         cmd = "ktadd -k %s -e %s %s" % (usr_keytab, KEY_TYPE, self.testuser)
 288         with (open(testlog, 'a')) as logfile:
 289             self.kadmin_local(cmd, env, logfile)
 290
 291         keys_env = {"KRB5_KTNAME": svc_keytab}
 292         keys_env.update(env)
 293
 294         return keys_env
 295
 296     def kinit_keytab(self, kdcenv):
 297         testlog = os.path.join(self.testdir, 'kinit.log')
 298         usr_keytab = os.path.join(self.testdir, USER_KTNAME)
 299         kdcenv['KRB5CCNAME'] = 'FILE:' + os.path.join(
 300             self.testdir, 'ccaches/user')
 301         with (open(testlog, 'a')) as logfile:
 302             logfile.write("\n%s\n" % kdcenv)
 303             ksetup = subprocess.Popen(["kinit", "-kt", usr_keytab,
 304                                        self.testuser],
 305                                       stdout=logfile, stderr=logfile,
 306                                       env=kdcenv, preexec_fn=os.setsid)
 307             ksetup.wait()
 308             if ksetup.returncode != 0:
 309                 raise ValueError('kinit %s failed' % self.testuser)
 310
 311     def wait(self):
 312         for p in self.processes:
 313             os.killpg(p.pid, signal.SIGTERM)
 314
 315     def setup_servers(self, env=None):
 316         raise NotImplementedError()
 317
 318     def run(self, env):
 319         exe = self.execname
 320         if exe.endswith('c'):
 321             exe = exe[:-1]
 322         return subprocess.call([exe], env=env)