s4-dsdb Add PAC validation test to tokengroups test.

This confirms that the groups obtained from a Kerberos PAC match those
that a manual search of a target LDAP server would reveal.

This should allow mixing of a KDC specified by krb5.conf to test Samba
or Windows alternatly.

Andrew Bartlett

Autobuild-User: Andrew Bartlett <abartlet@samba.org>
Autobuild-Date: Wed Jan 19 13:13:48 CET 2011 on sn-devel-104

2 files changed, 79 insertions, 21 deletions

diff --git a/source4/dsdb/tests/python/token_group.py b/source4/dsdb/tests/python/token_group.py
index 0314cd3332..a35f1836e2 100755
--- a/source4/dsdb/tests/python/token_group.py
+++ b/source4/dsdb/tests/python/token_group.py
@@ -16,10 +16,14 @@ import samba.getopt as options
 from samba.auth import system_session
 from samba import ldb
 from samba.samdb import SamDB
+from samba.auth import AuthContext
 from samba.ndr import ndr_pack, ndr_unpack
+from samba import gensec
+from samba.credentials import Credentials
 
 from subunit.run import SubunitTestRunner
 import unittest
+import samba.tests
 
 from samba.dcerpc import security
 from samba.auth import AUTH_SESSION_INFO_DEFAULT_GROUPS, AUTH_SESSION_INFO_AUTHENTICATED, AUTH_SESSION_INFO_SIMPLE_PRIVILEGES
@@ -43,14 +47,30 @@ url = args[0]
 lp = sambaopts.get_loadparm()
 creds = credopts.get_credentials(lp)
 
-class TokenTest(unittest.TestCase):
+class TokenTest(samba.tests.TestCase):
 
     def setUp(self):
         super(TokenTest, self).setUp()
         self.ldb = samdb
         self.base_dn = samdb.domain_dn()
 
-    def test_TokenGroups(self):
+        res = self.ldb.search("", scope=ldb.SCOPE_BASE, attrs=["tokenGroups"])
+        self.assertEquals(len(res), 1)
+
+        self.user_sid_dn = "<SID=%s>" % str(ndr_unpack(samba.dcerpc.security.dom_sid, res[0]["tokenGroups"][0]))
+
+        session_info_flags = ( AUTH_SESSION_INFO_DEFAULT_GROUPS |
+                               AUTH_SESSION_INFO_AUTHENTICATED |
+                               AUTH_SESSION_INFO_SIMPLE_PRIVILEGES)
+        session = samba.auth.user_session(self.ldb, lp_ctx=lp, dn=self.user_sid_dn,
+                                          session_info_flags=session_info_flags)
+
+        token = session.security_token
+        self.user_sids = []
+        for s in token.sids:
+            self.user_sids.append(str(s))
+
+    def test_rootDSE_tokenGroups(self):
         """Testing rootDSE tokengroups against internal calculation"""
         if not url.startswith("ldap"):
             self.fail(msg="This test is only valid on ldap")
@@ -63,38 +83,26 @@ class TokenTest(unittest.TestCase):
         for sid in res[0]['tokenGroups']:
             tokengroups.append(str(ndr_unpack(samba.dcerpc.security.dom_sid, sid)))
 
-        user_sid_dn = "<SID=%s>" % tokengroups[0]
-
-        print("Geting token from user session")
-        session_info_flags = ( AUTH_SESSION_INFO_DEFAULT_GROUPS |
-                               AUTH_SESSION_INFO_AUTHENTICATED |
-                               AUTH_SESSION_INFO_SIMPLE_PRIVILEGES)
-        session = samba.auth.user_session(self.ldb, lp_ctx=lp, dn=user_sid_dn,
-                                          session_info_flags=session_info_flags)
-
-        token = session.security_token
-        sids = []
-        for s in token.sids:
-            sids.append(str(s))
         sidset1 = set(tokengroups)
-        sidset2 = set(sids)
+        sidset2 = set(self.user_sids)
         if len(sidset1.difference(sidset2)):
             print("token sids don't match")
             print("tokengroups: %s" % tokengroups)
-            print("calculated : %s" % sids);
+            print("calculated : %s" % self.user_sids);
             print("difference : %s" % sidset1.difference(sidset2))
             self.fail(msg="calculated groups don't match against rootDSE tokenGroups")
 
-        res = self.ldb.search(user_sid_dn, scope=ldb.SCOPE_BASE, attrs=["tokenGroups"])
+    def test_dn_tokenGroups(self):
+        print("Geting tokenGroups from user DN")
+        res = self.ldb.search(self.user_sid_dn, scope=ldb.SCOPE_BASE, attrs=["tokenGroups"])
         self.assertEquals(len(res), 1)
 
-        print("Geting tokenGroups from user DN")
         dn_tokengroups = []
         for sid in res[0]['tokenGroups']:
             dn_tokengroups.append(str(ndr_unpack(samba.dcerpc.security.dom_sid, sid)))
 
         sidset1 = set(dn_tokengroups)
-        sidset2 = set(sids)
+        sidset2 = set(self.user_sids)
         if len(sidset1.difference(sidset2)):
             print("token sids don't match")
             print("tokengroups: %s" % tokengroups)
@@ -102,6 +110,56 @@ class TokenTest(unittest.TestCase):
             print("difference : %s" % sidset1.difference(sidset2))
             self.fail(msg="calculated groups don't match against user DN tokenGroups")
         
+    def test_pac_groups(self):
+        settings = {}
+        settings["lp_ctx"] = lp
+        settings["target_hostname"] = lp.get("netbios name")
+
+        gensec_client = gensec.Security.start_client(settings)
+        gensec_client.set_credentials(creds)
+        gensec_client.want_feature(gensec.FEATURE_SEAL)
+        gensec_client.start_mech_by_sasl_name("GSSAPI")
+
+        auth_context = AuthContext(lp_ctx=lp, ldb=self.ldb, methods=[])
+
+        gensec_server = gensec.Security.start_server(settings, auth_context)
+        machine_creds = Credentials()
+        machine_creds.guess(lp)
+        machine_creds.set_machine_account(lp)
+        gensec_server.set_credentials(machine_creds)
+
+        gensec_server.want_feature(gensec.FEATURE_SEAL)
+        gensec_server.start_mech_by_sasl_name("GSSAPI")
+
+        client_finished = False
+        server_finished = False
+        server_to_client = None
+        
+        """Run the actual call loop"""
+        while client_finished == False and server_finished == False:
+            if not client_finished:
+                print "running client gensec_update"
+                (client_finished, client_to_server) = gensec_client.update(server_to_client)
+            if not server_finished:
+                print "running server gensec_update"
+                (server_finished, server_to_client) = gensec_server.update(client_to_server)
+
+        session = gensec_server.session_info()
+
+        token = session.security_token
+        pac_sids = []
+        for s in token.sids:
+            pac_sids.append(str(s))
+
+        sidset1 = set(pac_sids)
+        sidset2 = set(self.user_sids)
+        if len(sidset1.difference(sidset2)):
+            print("token sids don't match")
+            print("tokengroups: %s" % tokengroups)
+            print("calculated : %s" % sids);
+            print("difference : %s" % sidset1.difference(sidset2))
+            self.fail(msg="calculated groups don't match against user PAC tokenGroups")
+
 
 if not "://" in url:
     if os.path.isfile(url):
diff --git a/source4/selftest/tests.py b/source4/selftest/tests.py
index 413d86d739..6e2ade145e 100755
--- a/source4/selftest/tests.py
+++ b/source4/selftest/tests.py
@@ -499,7 +499,7 @@ planpythontestsuite("none", "subunit")
 planpythontestsuite("dc:local", "samba.tests.dcerpc.rpcecho")
 plantestsuite_idlist("samba.tests.dcerpc.registry", "dc:local", [subunitrun, "$LISTOPT", '-U"$USERNAME%$PASSWORD"', "samba.tests.dcerpc.registry"])
 plantestsuite("samba4.ldap.python(dc)", "dc", [python, os.path.join(samba4srcdir, "dsdb/tests/python/ldap.py"), '$SERVER', '-U"$USERNAME%$PASSWORD"', '-W', '$DOMAIN'])
-plantestsuite("samba4.tokengroups.python(dc)", "dc", [python, os.path.join(samba4srcdir, "dsdb/tests/python/token_group.py"), '$SERVER', '-U"$USERNAME%$PASSWORD"', '-W', '$DOMAIN'])
+plantestsuite("samba4.tokengroups.python(dc)", "dc:local", [python, os.path.join(samba4srcdir, "dsdb/tests/python/token_group.py"), '$SERVER', '-U"$USERNAME%$PASSWORD"', '-W', '$DOMAIN'])
 plantestsuite("samba4.sam.python(dc)", "dc", [python, os.path.join(samba4srcdir, "dsdb/tests/python/sam.py"), '$SERVER', '-U"$USERNAME%$PASSWORD"', '-W', '$DOMAIN'])
 plansambapythontestsuite("samba4.schemaInfo.python(dc)", "dc", os.path.join(samba4srcdir, 'dsdb/tests/python'), 'dsdb_schema_info', extra_args=['-U"$DOMAIN/$DC_USERNAME%$DC_PASSWORD"'])
 plantestsuite("samba4.urgent_replication.python(dc)", "dc", [python, os.path.join(samba4srcdir, "dsdb/tests/python/urgent_replication.py"), '$PREFIX_ABS/dc/private/sam.ldb'], allow_empty_output=True)