23 """A module for krb5 test scripts
25 To run test scripts during "make check" (if Python 2.5 or later is
28     check-pytests::
31 A sample test script:
35     # Run a test program under a variety of configurations:
39     # Run a test server and client under just the default configuration:
79     - $realm:    The realm name
80     - $testdir:  The realm storage directory (absolute path)
81     - $buildtop: The root of the build directory
82     - $srctop:   The root of the source directory
83     - $plugins:  The plugin directory in the build tree
84     - $certs:    The PKINIT certificate directory in the source tree
85     - $hostname: The FQDN of the host
86     - $port0:    The first listener port (portbase)
87     - ...
88     - $port9:    The tenth listener port (portbase + 9)
92     - port0 is used in the default krb5.conf for the KDC
93     - port1 is used in the default krb5.conf for kadmind
94     - port2 is used in the default krb5.conf for kpasswd
95     - port3 is used in the default krb5.conf for kpropd
96     - port4 is used in the default krb5.conf for iprop (in kadmind)
97     - port5 is the return value of realm.server_port()
132 * success(message): Indicate that the test script has completed
134   messages in the on-exit handler.  message should briefly summarize
150 * mark(message): Place a divider message in the test output, to make
151   it easier to determine what part of the test script a command
171   a realm for each of the standard test passes, each of which alters
176   they will be merged with the default and per-pass specifications.
180   included, but setting up a test LDAP server currently requires a
181   one-second delay, so all LDAP tests are currently confined to
189   the xtgts parameter, where each tuple is a pair of zero-based realm
206 * pkinit_certs: The directory containing test PKINIT certificates.
210   turned off in the test environment to make this value easy to
225 * testpass: The command-line test pass argument.  The script does not
230   - krb5kdc
231   - kadmind
232   - kadmin
233   - kadminl (kadmin.local)
234   - kdb5_ldap_util
235   - kdb5_util
236   - ktutil
237   - kinit
238   - klist
239   - kswitch
240   - kvno
241   - kdestroy
242   - kpasswd
243   - t_inetd
244   - kproplog
245   - kpropd
246   - kprop
252   command-line debugging options.  Fail if the command does not return
254   multi-line string.  Keyword arguments can contain input='string' to
304   a keytab for princname in the filename keytab.  Uses the -norandkey
305   option to avoid re-randomizing princname's key.
314   supplying a PKINIT identity of the basic user test certificate
319   self.ccache if not specified) and verify that the output shows
320   credentials for client_princ and service_princ (self.krbtgt_princ if
324   (must be a filename; self.keytab if not specified) and verify that
349 * realm.run_kpropd_once(env, args=[]): Run kpropd once, using the -t
388 When the test script is run, its behavior can be modified with
389 command-line flags.  These are documented in the --help output.
408 # without help from the test script.  The on-exit handler will display
417         print("*** Last command (#%d): %s" % (_cmd_index - 1, _last_cmd))
425         print("*** Failed in test pass:", _current_pass)
518             print('Or re-run this test script with the -v flag:')
520             print('    PYTHONPATH=%s %s %s -v' %
523         print('Use --debug=NUM to run a command under a debugger.  Use')
524         print('--stop-after=NUM to stop after a daemon is started in order to')
525         print('attach to it with a debugger.  Use --help to see other')
576     f = which('llvm-symbolizer')
580     # Debian-derived systems have versioned symbolizer names.  If any
582     l = glob.glob('/usr/bin/llvm-symbolizer-*')
595 # be used by the test script.
600     parser.add_option('-v', '--verbose', action='store_true', dest='verbose',
602     parser.add_option('-p', '--pass', dest='testpass', metavar='PASS',
603                       help='If a multi-pass test, run only PASS')
604     parser.add_option('--debug', dest='debug', metavar='NUM',
606     parser.add_option('--debugger', dest='debugger', metavar='COMMAND',
607                       help='Debugger command (default is gdb --args)')
608     parser.add_option('--stop-before', dest='stopb', metavar='NUM',
610     parser.add_option('--stop-after', dest='stopa', metavar='NUM',
612     parser.add_option('--shell-before', dest='shellb', metavar='NUM',
614     parser.add_option('--shell-after', dest='shella', metavar='NUM',
619     _debug = _parse_cmdnum('--debug', options.debug)
620     _stop_before = _parse_cmdnum('--stop-before', options.stopb)
621     _stop_after = _parse_cmdnum('--stop-after', options.stopa)
622     _shell_before = _parse_cmdnum('--shell-before', options.shellb)
623     _shell_after = _parse_cmdnum('--shell-after', options.shella)
628         _debugger_command = ['gdb', '--args']
630         _debugger_command = ['lldb', '--']
632         print('Cannot find a debugger; use --debugger=COMMAND')
636 # Translate a command number spec.  -1 means all, None means none.
641         return -1
648 # Test if a command index matches a translated command number spec.
652     elif cmdnum == -1:
820         output( 'test script cannot proceed after debugging a daemon\n', True)
853     #   - Output to a file and poll the file for the sentinel
854     #     (undesirable because it slows down the test suite by the
856     #   - Create an intermediate subprocess which discards output
870     # If a daemon doesn't catch SIGTERM (like gss-server), treat it as
872     if code == -signal.SIGTERM:
917     """An object representing a functional krb5 test realm."""
919     def __init__(self, realm='KRBTEST.COM', portbase=61000, testdir='testdir',  argument
927         self.realm = realm
928         self.testdir = os.path.join(os.getcwd(), testdir)
929         self.portbase = portbase
930         self.user_princ = 'user@' + self.realm
931         self.admin_princ = 'user/admin@' + self.realm
932         self.host_princ = 'host/%s@%s' % (hostname, self.realm)
933         self.nfs_princ = 'nfs/%s@%s' % (hostname, self.realm)
934         self.krbtgt_princ = 'krbtgt/%s@%s' % (self.realm, self.realm)
935         self.keytab = os.path.join(self.testdir, 'keytab')
936         self.client_keytab = os.path.join(self.testdir, 'client_keytab')
937         self.ccache = os.path.join(self.testdir, 'ccache')
938         self.gss_mech_config = os.path.join(self.testdir, 'mech.conf')
939         self.kadmin_ccache = os.path.join(self.testdir, 'kadmin_ccache')
948         self._krb5_conf = _cfg_merge(base_krb5_conf, krb5_conf)
949         self._kdc_conf = _cfg_merge(base_kdc_conf, kdc_conf)
950         self._kdc_proc = None
951         self._kadmind_proc = None
952         self._kpropd_procs = []
953         krb5_conf_path = os.path.join(self.testdir, 'krb5.conf')
954         kdc_conf_path = os.path.join(self.testdir, 'kdc.conf')
955         self.env = self._make_env(krb5_conf_path, kdc_conf_path)
957         self._create_empty_dir()
958         self._create_conf(self._krb5_conf, krb5_conf_path)
959         self._create_conf(self._kdc_conf, kdc_conf_path)
960         self._create_acl()
961         self._create_dictfile()
964             self.create_kdb()
966             self.run([kadminl, 'cpw', '-randkey', '-e', krbtgt_keysalt,
967                       self.krbtgt_princ])
969             self.addprinc(self.user_princ, password('user'))
970             self.addprinc(self.admin_princ, password('admin'))
972             self.addprinc(self.host_princ)
973             self.extract_keytab(self.host_princ, self.keytab)
975             self.start_kdc()
977             self.start_kadmind()
979             self.kinit(self.user_princ, password('user'))
980             self.klist(self.user_princ)
981         self._setup_symbolizer()
983     def _create_empty_dir(self):  argument
984         dir = self.testdir
987             fail('Cannot remove %s to create test realm.' % dir)
990     def _create_conf(self, profile, filename):  argument
994             self._write_cfg_section(file, contents, 1)
997     def _write_cfg_section(self, file, contents, indent_level):  argument
1000             name = self._subst_cfg_value(name)
1004                 self._write_cfg_section(file, value, indent_level + 1)
1009                     item = self._subst_cfg_value(item)
1013                 value = self._subst_cfg_value(value)
1018     def _subst_cfg_value(self, value):  argument
1021         subst = template.substitute(realm=self.realm,
1022                                     testdir=self.testdir,
1028                                     port0=self.portbase,
1029                                     port1=self.portbase + 1,
1030                                     port2=self.portbase + 2,
1031                                     port3=self.portbase + 3,
1032                                     port4=self.portbase + 4,
1033                                     port5=self.portbase + 5,
1034                                     port6=self.portbase + 6,
1035                                     port7=self.portbase + 7,
1036                                     port8=self.portbase + 8,
1037                                     port9=self.portbase + 9)
1041     def _create_acl(self):  argument
1043         filename = os.path.join(self.testdir, 'acl')
1045         file.write('%s *e\n' % self.admin_princ)
1046         file.write('kiprop/%s@%s p\n' % (hostname, self.realm))
1049     def _create_dictfile(self):  argument
1050         filename = os.path.join(self.testdir, 'dictfile')
1055     def _make_env(self, krb5_conf_path, kdc_conf_path):  argument
1059         env['KRB5CCNAME'] = self.ccache
1060         env['KRB5_KTNAME'] = self.keytab
1061         env['KRB5_CLIENT_KTNAME'] = self.client_keytab
1062         env['KRB5RCACHEDIR'] = self.testdir
1063         env['KPROPD_PORT'] = str(self.kprop_port())
1064         env['KPROP_PORT'] = str(self.kprop_port())
1065         env['GSS_MECH_CONFIG'] = self.gss_mech_config
1069     # llvm-symbolizer, which is invoked by asan when displaying stack
1070     # traces.  If they are, asan-compiled krb5 libraries in
1075     def _setup_symbolizer(self):  argument
1078         if 'ASAN_SYMBOLIZER_PATH' in self.env:
1083         wrapper_path = os.path.join(self.testdir, 'llvm-symbolizer')
1090         self.env['ASAN_SYMBOLIZER_PATH'] = wrapper_path
1092     def run(self, args, env=None, **keywords):  argument
1094             env = self.env
1097     def kprop_port(self):  argument
1098         return self.portbase + 3
1100     def server_port(self):  argument
1101         return self.portbase + 5
1103     def start_server(self, args, sentinel, env=None):  argument
1105             env = self.env
1108     def start_in_inetd(self, args, port=None, env=None):  argument
1110             port = self.server_port()
1112             env = self.env
1116     def create_kdb(self):  argument
1118         self.run([kdb5_util, 'create', '-s', '-P', 'master'])
1120     def start_kdc(self, args=[], env=None):  argument
1123             env = self.env
1124         assert(self._kdc_proc is None)
1125         self._kdc_proc = _start_daemon([krb5kdc, '-n'] + args, env,
1128     def stop_kdc(self):  argument
1129         assert(self._kdc_proc is not None)
1130         stop_daemon(self._kdc_proc)
1131         self._kdc_proc = None
1133     def start_kadmind(self, env=None):  argument
1136             env = self.env
1137         assert(self._kadmind_proc is None)
1138         dump_path = os.path.join(self.testdir, 'dump')
1139         self._kadmind_proc = _start_daemon([kadmind, '-nofork',
1140                                             '-p', kdb5_util, '-K', kprop,
1141                                             '-F', dump_path], env,
1144     def stop_kadmind(self):  argument
1145         assert(self._kadmind_proc is not None)
1146         stop_daemon(self._kadmind_proc)
1147         self._kadmind_proc = None
1149     def _kpropd_args(self):  argument
1150         datatrans_path = os.path.join(self.testdir, 'incoming-datatrans')
1151         kpropdacl_path = os.path.join(self.testdir, 'kpropd-acl')
1152         return [kpropd, '-D', '-P', str(self.kprop_port()),
1153                 '-f', datatrans_path, '-p', kdb5_util, '-a', kpropdacl_path]
1155     def start_kpropd(self, env, args=[]):  argument
1156         proc = _start_daemon(self._kpropd_args() + args, env, 'ready')
1157         self._kpropd_procs.append(proc)
1160     def stop_kpropd(self, proc):  argument
1162         self._kpropd_procs.remove(proc)
1164     def run_kpropd_once(self, env, args=[]):  argument
1165         return self.run(self._kpropd_args() + ['-t'] + args, env=env)
1167     def stop(self):  argument
1168         if self._kdc_proc:
1169             self.stop_kdc()
1170         if self._kadmind_proc:
1171             self.stop_kadmind()
1172         for p in self._kpropd_procs:
1174         self._kpropd_procs = []
1176     def addprinc(self, princname, password=None):  argument
1178             self.run([kadminl, 'addprinc', '-pw', password, princname])
1180             self.run([kadminl, 'addprinc', '-randkey', princname])
1182     def extract_keytab(self, princname, keytab):  argument
1183         self.run([kadminl, 'ktadd', '-k', keytab, '-norandkey', princname])
1185     def kinit(self, princname, password=None, flags=[], **keywords):  argument
1190         return self.run([kinit] + flags + [princname], input=input, **keywords)
1192     def pkinit(self, princ, flags=[], **kw):  argument
1195         flags = flags + ['-X', 'X509_user_identity=%s' % id]
1196         self.kinit(princ, flags=flags, **kw)
1198     def klist(self, client_princ, service_princ=None, ccache=None, **keywords):  argument
1200             service_princ = self.krbtgt_princ
1202             ccache = self.ccache
1206         output = self.run([klist, ccache], **keywords)
1212     def klist_keytab(self, princ, keytab=None, **keywords):  argument
1214             keytab = self.keytab
1215         output = self.run([klist, '-k', keytab], **keywords)
1217             'KVNO Principal\n----' not in output or
1221     def prep_kadmin(self, princname=None, pw=None, flags=[]):  argument
1223             princname = self.admin_princ
1225         return self.kinit(princname, pw,
1226                           flags=['-S', 'kadmin/admin',
1227                                  '-c', self.kadmin_ccache] + flags)
1229     def run_kadmin(self, args, **keywords):  argument
1230         return self.run([kadmin, '-c', self.kadmin_ccache] + args, **keywords)
1232     def special_env(self, name, has_kdc_conf, krb5_conf=None, kdc_conf=None):  argument
1233         krb5_conf_path = os.path.join(self.testdir, 'krb5.conf.%s' % name)
1234         krb5_conf = _cfg_merge(self._krb5_conf, krb5_conf)
1235         self._create_conf(krb5_conf, krb5_conf_path)
1237             kdc_conf_path = os.path.join(self.testdir, 'kdc.conf.%s' % name)
1238             kdc_conf = _cfg_merge(self._kdc_conf, kdc_conf)
1239             self._create_conf(kdc_conf, kdc_conf_path)
1242         return self._make_env(krb5_conf_path, kdc_conf_path)
1334                 r.addprinc('krbtgt/%s' % jname, password('cr-%d-%d-' % (i, j)))
1338                            password('cr-%d-%d-' % (j, i)))
1395                     'supported_enctypes': 'des3-cbc-sha1:normal',
1396                     'master_key_type': 'des3-cbc-sha1'}}}),
1402                     'supported_enctypes': 'arcfour-hmac:normal',
1403                     'master_key_type': 'arcfour-hmac'}}}),
1407       {'libdefaults': {'permitted_enctypes': 'aes128-cts'}},
1409                     'supported_enctypes': 'aes128-cts:normal',
1410                     'master_key_type': 'aes128-cts'}}}),
1412     # Exercise the camellia256-cts enctype.
1414       {'libdefaults': {'permitted_enctypes': 'camellia256-cts'}},
1416                     'supported_enctypes': 'camellia256-cts:normal',
1417                     'master_key_type': 'camellia256-cts'}}}),
1419     # Exercise the aes128-sha2 enctype.
1420     ('aes128-sha2', None,
1421       {'libdefaults': {'permitted_enctypes': 'aes128-sha2'}},
1423                     'supported_enctypes': 'aes128-sha2:normal',
1424                     'master_key_type': 'aes128-sha2'}}}),
1426     # Exercise the aes256-sha2 enctype.
1427     ('aes256-sha2', None,
1428       {'libdefaults': {'permitted_enctypes': 'aes256-sha2'}},
1430                     'supported_enctypes': 'aes256-sha2:normal',
1431                     'master_key_type': 'aes256-sha2'}}}),
1433     # Test a setup with modern principal keys but an old TGT key.
1434     ('aes256.destgt', 'arcfour-hmac:normal',
1456 pkinit_certs = os.path.join(srctop, 'tests', 'pkinit-certs')