TESTS: Add a cwrap-enabled test for krb5_child

Adds two Python scripts -- one that runs a KDC on non-privileged port
and one that talks to this KDC using the krb5_child test program we
developed earlier.

5 files changed, 258 insertions, 4 deletions

diff --git a/Makefile.am b/Makefile.am
index f9e2de2e0..ed14fb95d 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -1616,7 +1616,7 @@ krb5_child_test_SOURCES = \
     $(NULL)
 krb5_child_test_CFLAGS = \
     $(AM_CFLAGS) \
-    -DKRB5_CHILD_DIR=\"$(builddir)\" \
+    -DKRB5_CHILD_DIR=\"$(abs_top_builddir)\" \
     $(KRB5_CFLAGS) \
     $(CHECK_CFLAGS)
 krb5_child_test_LDADD = \
diff --git a/src/providers/krb5/krb5_child_handler.c b/src/providers/krb5/krb5_child_handler.c
index 93961172c..d8a774c37 100644
--- a/src/providers/krb5/krb5_child_handler.c
+++ b/src/providers/krb5/krb5_child_handler.c
@@ -301,8 +301,9 @@ static errno_t fork_child(struct tevent_req *req)
                          pipefd_to_child, pipefd_from_child,
                          KRB5_CHILD, state->kr->krb5_ctx->child_debug_fd);
         if (err != EOK) {
-            DEBUG(SSSDBG_CRIT_FAILURE, "Could not exec KRB5 child: [%d][%s].\n",
-                      err, strerror(err));
+            DEBUG(SSSDBG_CRIT_FAILURE,
+                  "Could not exec KRB5 child [%s]: [%d][%s].\n",
+                  KRB5_CHILD, err, strerror(err));
             return err;
         }
     } else if (pid > 0) { /* parent */
diff --git a/src/tests/cwrap/Makefile.am b/src/tests/cwrap/Makefile.am
index d98bb3379..5a014ae79 100644
--- a/src/tests/cwrap/Makefile.am
+++ b/src/tests/cwrap/Makefile.am
@@ -17,6 +17,7 @@ TESTS_ENVIRONMENT = \
 
 dist_noinst_SCRIPTS = \
     cwrap_test_setup.sh \
+    test_krb5_child.py \
     $(NULL)
 
 SSSD_LIBS = \
@@ -39,6 +40,7 @@ dist_noinst_DATA = \
     $(NULL)
 
 check_PROGRAMS =
+PYTHON_TESTS =
 if HAVE_CMOCKA
 if HAVE_NSS_WRAPPER
 if HAVE_UID_WRAPPER
@@ -48,11 +50,23 @@ check_PROGRAMS += \
     usertools-tests \
     responder_common-tests \
     $(NULL)
+
+EXTRA_DIST = \
+	fake_kdc.py \
+	$(NULL)
+
+PYTHON_TESTS += \
+	test_krb5_child.py \
+	$(NULL)
+
 endif # HAVE_UID_WRAPPER
 endif # HAVE_NSS_WRAPPER
 endif # HAVE_CMOCKA
 
-TESTS = $(check_PROGRAMS)
+TESTS = \
+	$(check_PROGRAMS) \
+	$(PYTHON_TESTS) \
+	$(NULL)
 
 become_user_tests_SOURCES = \
     test_become_user.c \
diff --git a/src/tests/cwrap/fake_kdc.py b/src/tests/cwrap/fake_kdc.py
new file mode 100644
index 000000000..896d1c8cb
--- /dev/null
+++ b/src/tests/cwrap/fake_kdc.py
@@ -0,0 +1,137 @@
+#!/usr/bin/python2
+
+import tempfile
+import subprocess
+import shutil
+import time
+import os
+import os.path
+import signal
+from string import Template
+
+mock_kdc_conf = \
+"""
+[kdcdefaults]
+ kdc_ports = $KDC_PORT, $KADMIN_PORT
+ kdc_tcp_ports = $KDC_PORT, $KADMIN_PORT
+
+[logging]
+ default = FILE:$DIR/krb5.log
+ kdc = FILE:$DIR/kdc.log
+ admin_server = FILE:$DIR/kadmind.log
+
+[realms]
+ $REALM = {
+  key_stash_file = $DIR/key_stash
+  acl_file = $KADM_ACL_FILE
+  admin_keytab = $DIR/kadm5.keytab
+  supported_enctypes = aes256-cts:normal aes128-cts:normal des3-hmac-sha1:normal arcfour-hmac:normal camellia256-cts:normal camellia128-cts:normal des-hmac-sha1:normal des-cbc-md5:normal des-cbc-crc:normal
+  database_name = $DIR/database
+ }
+"""
+
+mock_krb5_conf = \
+"""
+[logging]
+ default = FILE:$DIR/krb5.log
+ kdc = FILE:$DIR/kdc.log
+ admin_server = FILE:$DIR/kadmind.log
+
+[libdefaults]
+ dns_lookup_realm = false
+ default_realm = $REALM
+ rdns = false
+
+[realms]
+ $REALM = {
+   kdc = $KDC_HOSTNAME:$KDC_PORT
+   master_kdc = $KDC_HOSTNAME:$KDC_PORT
+   admin_server = $KADMIN_HOSTNAME:$KADMIN_PORT
+   kpasswd_server = $KADMIN_HOSTNAME:$KADMIN_PORT
+ }
+
+[domain_realm]
+ $KDC_HOSTNAME = $REALM
+"""
+
+mock_kadm_acl = "*/admin@$REALM      *"
+
+def tmpfile_from_template(template, basename, dir, **subst_dict):
+    tmpl = Template(template)
+    content = tmpl.substitute(**subst_dict)
+    full_path = os.path.join(dir, basename)
+
+    with open(full_path, 'w') as f:
+        f.write(content)
+        f.flush()
+
+    return full_path
+
+def mock_kdc(wdir, users):
+    subst_dict = {
+            'REALM' : 'SSSD.MOCK',
+            'KDC_HOSTNAME' : 'localhost',
+            'KADMIN_HOSTNAME' : 'localhost',
+            'KDC_PORT' : '20088',
+            'KADMIN_PORT' : '20750',
+            'DIR' : wdir,
+    }
+
+    env = dict(os.environ)
+
+    kadm_acl = tmpfile_from_template(mock_kadm_acl, 'kadm5.acl', wdir, **subst_dict)
+
+    subst_dict['KADM_ACL_FILE'] = kadm_acl
+    kdc_conf = tmpfile_from_template(mock_kdc_conf, 'kdc.conf', wdir, **subst_dict)
+    env['KRB5_KDC_PROFILE'] = kdc_conf
+
+    krb5_conf = tmpfile_from_template(mock_krb5_conf, 'krb5.conf', wdir, **subst_dict)
+    env['KRB5_CONFIG'] = krb5_conf
+
+    # Generate the KDC database, undocumented -W argument: no strong random
+    kdb_util = subprocess.Popen(['kdb5_util', 'create', '-r',
+                                 subst_dict['REALM'], '-s', '-W',
+                                 '-P', 'foobar'],
+                                 env = env,
+                                 cwd = wdir)
+    kdb_util.communicate()
+
+    addprinc_cmd = [ 'kadmin.local', '-r', subst_dict['REALM'], '-q' ]
+
+    for username, password in users.iteritems():
+        addprinc_arg = 'addprinc -pw %s -clearpolicy %s' % (password, username)
+        kadmin = subprocess.Popen(addprinc_cmd + [addprinc_arg],
+                                  env = env, cwd = wdir)
+        kadmin.communicate()
+
+    pidfile = os.path.join(wdir, "kdc.pid")
+    kdc = subprocess.Popen(['krb5kdc', '-P', pidfile, '-r', subst_dict['REALM']],
+                           env = env, cwd = wdir)
+    kdc.communicate()
+
+    # Wait for the KDC to come up
+    for i in range(1, 10):
+        try:
+            with open(pidfile) as pf:
+                pid = int(pf.read())
+        except IOError:
+            time.sleep(1)
+
+    return (krb5_conf, pid)
+
+if __name__ == "__main__":
+    users = { 'root/admin' : 'TurboGoesToRocket',
+              'foobar' : 'Secret123' }
+    realm = 'SSSD.MOCK'
+
+    wdir = tempfile.mkdtemp(prefix='sssd_mock_kdc')
+    kdc_pid = None
+    try:
+        krb5_conf, kdc_pid = mock_kdc(wdir, users)
+        print "KRB5_CONFIG=%s" % krb5_conf
+        print "PID=%d" % kdc_pid
+        raw_input()
+    finally:
+        shutil.rmtree(wdir)
+        if kdc_pid:
+            os.kill(kdc_pid, signal.SIGTERM)
diff --git a/src/tests/cwrap/test_krb5_child.py b/src/tests/cwrap/test_krb5_child.py
new file mode 100755
index 000000000..50f273483
--- /dev/null
+++ b/src/tests/cwrap/test_krb5_child.py
@@ -0,0 +1,102 @@
+#!/usr/bin/python2
+
+import unittest
+import os
+import os.path
+import sys
+import tempfile
+import subprocess
+import shutil
+import signal
+
+from fake_kdc import mock_kdc
+
+test_bin = '../../../krb5-child-test'
+
+class Krb5ChildTest(unittest.TestCase):
+    def setUp(self):
+        # No point in running the tests without wrappers in place
+        self.assertWrappers()
+
+        # It would be nice to not pollute /tmp with testing data, but
+        # it's not really possible to chown a directory to the test user
+        # either
+        self.ccache_dir = '/tmp'
+        self.realm = 'SSSD.MOCK'
+        self.users = { 'root/admin' : 'TurboGoesToRocket',
+                       'foobar' : 'Secret123' }
+
+        self.wdir = tempfile.mkdtemp(prefix='sssd_mock_kdc')
+
+        self.krb5_conf, self.kdc_pid = mock_kdc(self.wdir, self.users)
+        self.env = dict(os.environ)
+        self.env['KRB5_CONFIG'] = self.krb5_conf
+
+    def tearDown(self):
+        os.kill(self.kdc_pid, signal.SIGTERM)
+        shutil.rmtree(self.wdir)
+
+    def testKinit(self):
+        username = 'foobar'
+
+        child_test = subprocess.Popen([test_bin, '-u', username,
+                                     '-w', self.users[username],
+                                     '-r', self.realm,
+                                     '--debug', '10',
+                                     '-c', 'FILE:%s' % self.ccache_path(username),
+                                     '-k'],
+                                     env = self.env)
+        child_test.communicate()
+        self.assertEqual(child_test.returncode, 0)
+        self.assertPrincipalInCcache(self.principal(username, self.realm),
+                                     self.ccache_path(username))
+
+    def testKinitBadPassword(self):
+        username = 'foobar'
+
+        child_test = subprocess.Popen([test_bin, '-u', username,
+                                     '-w', 'NotTheRightOne',
+                                     '-r', self.realm,
+                                     '--debug', '10',
+                                     '-c', 'FILE:%s' % self.ccache_path(username)],
+                                     env = self.env)
+        child_test.communicate()
+        self.assertEqual(child_test.returncode, 6)
+
+    def assertPrincipalInCcache(self, principal, ccache):
+        klist = subprocess.Popen(['klist', ccache], stdout=subprocess.PIPE)
+        klist.communicate()
+        # FIXME - open the ccache with python-kerberos and check the contents
+        self.assertEqual(klist.returncode, 0)
+
+    def assertWrappers(self):
+        required_vars = [ 'UID_WRAPPER', 'UID_WRAPPER_ROOT',
+                          'NSS_WRAPPER_PASSWD', 'NSS_WRAPPER_GROUP' ]
+        for v in required_vars:
+            assert v in os.environ
+
+    def principal(self, username, realm):
+        return '%s@%s' % (username, realm)
+
+    def ccache_path(self, username):
+        return os.path.join(self.ccache_dir, "%s_ccache" % username)
+
+if __name__ == "__main__":
+    error = 0
+
+    try:
+        subprocess.call(["krb5kdc"])
+    except OSError as e:
+        if e.errno == os.errno.ENOENT:
+            print "KRB5KDC not found, cannot run tests!\n"
+            sys.exit(error)
+        else:
+            # Something else went wrong while trying to run `wget`
+            raise
+
+    suite = unittest.TestLoader().loadTestsFromTestCase(Krb5ChildTest)
+    res = unittest.TextTestRunner().run(suite)
+    if not res.wasSuccessful():
+        error |= 0x1
+
+    sys.exit(error)