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