source4/dsdb/tests/python/token_group.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3 # test tokengroups attribute against internal token calculation
   4
   5 import optparse
   6 import sys
   7 import os
   8
   9 sys.path.insert(0, "bin/python")
  10 import samba
  11 samba.ensure_external_module("testtools", "testtools")
  12 samba.ensure_external_module("subunit", "subunit/python")
  13
  14 import samba.getopt as options
  15
  16 from samba.auth import system_session
  17 from samba import ldb
  18 from samba.samdb import SamDB
  19 from samba.auth import AuthContext
  20 from samba.ndr import ndr_unpack
  21 from samba import gensec
  22 from samba.credentials import Credentials
  23
  24 from subunit.run import SubunitTestRunner
  25 import unittest
  26 import samba.tests
  27
  28 from samba.auth import AUTH_SESSION_INFO_DEFAULT_GROUPS, AUTH_SESSION_INFO_AUTHENTICATED, AUTH_SESSION_INFO_SIMPLE_PRIVILEGES
  29
  30
  31 parser = optparse.OptionParser("ldap.py [options] <host>")
  32 sambaopts = options.SambaOptions(parser)
  33 parser.add_option_group(sambaopts)
  34 parser.add_option_group(options.VersionOptions(parser))
  35 # use command line creds if available
  36 credopts = options.CredentialsOptions(parser)
  37 parser.add_option_group(credopts)
  38 opts, args = parser.parse_args()
  39
  40 if len(args) < 1:
  41     parser.print_usage()
  42     sys.exit(1)
  43
  44 url = args[0]
  45
  46 lp = sambaopts.get_loadparm()
  47 creds = credopts.get_credentials(lp)
  48
  49 class TokenTest(samba.tests.TestCase):
  50
  51     def setUp(self):
  52         super(TokenTest, self).setUp()
  53         self.ldb = samdb
  54         self.base_dn = samdb.domain_dn()
  55
  56         res = self.ldb.search("", scope=ldb.SCOPE_BASE, attrs=["tokenGroups"])
  57         self.assertEquals(len(res), 1)
  58
  59         self.user_sid_dn = "<SID=%s>" % str(ndr_unpack(samba.dcerpc.security.dom_sid, res[0]["tokenGroups"][0]))
  60
  61         session_info_flags = ( AUTH_SESSION_INFO_DEFAULT_GROUPS |
  62                                AUTH_SESSION_INFO_AUTHENTICATED |
  63                                AUTH_SESSION_INFO_SIMPLE_PRIVILEGES)
  64         session = samba.auth.user_session(self.ldb, lp_ctx=lp, dn=self.user_sid_dn,
  65                                           session_info_flags=session_info_flags)
  66
  67         token = session.security_token
  68         self.user_sids = []
  69         for s in token.sids:
  70             self.user_sids.append(str(s))
  71
  72     def test_rootDSE_tokenGroups(self):
  73         """Testing rootDSE tokengroups against internal calculation"""
  74         if not url.startswith("ldap"):
  75             self.fail(msg="This test is only valid on ldap")
  76
  77         res = self.ldb.search("", scope=ldb.SCOPE_BASE, attrs=["tokenGroups"])
  78         self.assertEquals(len(res), 1)
  79
  80         print("Getting tokenGroups from rootDSE")
  81         tokengroups = []
  82         for sid in res[0]['tokenGroups']:
  83             tokengroups.append(str(ndr_unpack(samba.dcerpc.security.dom_sid, sid)))
  84
  85         sidset1 = set(tokengroups)
  86         sidset2 = set(self.user_sids)
  87         if len(sidset1.difference(sidset2)):
  88             print("token sids don't match")
  89             print("tokengroups: %s" % tokengroups)
  90             print("calculated : %s" % self.user_sids)
  91             print("difference : %s" % sidset1.difference(sidset2))
  92             self.fail(msg="calculated groups don't match against rootDSE tokenGroups")
  93
  94     def test_dn_tokenGroups(self):
  95         print("Getting tokenGroups from user DN")
  96         res = self.ldb.search(self.user_sid_dn, scope=ldb.SCOPE_BASE, attrs=["tokenGroups"])
  97         self.assertEquals(len(res), 1)
  98
  99         dn_tokengroups = []
 100         for sid in res[0]['tokenGroups']:
 101             dn_tokengroups.append(str(ndr_unpack(samba.dcerpc.security.dom_sid, sid)))
 102
 103         sidset1 = set(dn_tokengroups)
 104         sidset2 = set(self.user_sids)
 105         if len(sidset1.difference(sidset2)):
 106             print("token sids don't match")
 107             print("difference : %s" % sidset1.difference(sidset2))
 108             self.fail(msg="calculated groups don't match against user DN tokenGroups")
 109
 110     def test_pac_groups(self):
 111         settings = {}
 112         settings["lp_ctx"] = lp
 113         settings["target_hostname"] = lp.get("netbios name")
 114
 115         gensec_client = gensec.Security.start_client(settings)
 116         gensec_client.set_credentials(creds)
 117         gensec_client.want_feature(gensec.FEATURE_SEAL)
 118         gensec_client.start_mech_by_sasl_name("GSSAPI")
 119
 120         auth_context = AuthContext(lp_ctx=lp, ldb=self.ldb, methods=[])
 121
 122         gensec_server = gensec.Security.start_server(settings, auth_context)
 123         machine_creds = Credentials()
 124         machine_creds.guess(lp)
 125         machine_creds.set_machine_account(lp)
 126         gensec_server.set_credentials(machine_creds)
 127
 128         gensec_server.want_feature(gensec.FEATURE_SEAL)
 129         gensec_server.start_mech_by_sasl_name("GSSAPI")
 130
 131         client_finished = False
 132         server_finished = False
 133         server_to_client = ""
 134
 135         """Run the actual call loop"""
 136         while client_finished == False and server_finished == False:
 137             if not client_finished:
 138                 print "running client gensec_update"
 139                 (client_finished, client_to_server) = gensec_client.update(server_to_client)
 140             if not server_finished:
 141                 print "running server gensec_update"
 142                 (server_finished, server_to_client) = gensec_server.update(client_to_server)
 143
 144         session = gensec_server.session_info()
 145
 146         token = session.security_token
 147         pac_sids = []
 148         for s in token.sids:
 149             pac_sids.append(str(s))
 150
 151         sidset1 = set(pac_sids)
 152         sidset2 = set(self.user_sids)
 153         if len(sidset1.difference(sidset2)):
 154             print("token sids don't match")
 155             print("difference : %s" % sidset1.difference(sidset2))
 156             self.fail(msg="calculated groups don't match against user PAC tokenGroups")
 157
 158
 159 if not "://" in url:
 160     if os.path.isfile(url):
 161         url = "tdb://%s" % url
 162     else:
 163         url = "ldap://%s" % url
 164
 165 samdb = SamDB(url, credentials=creds, session_info=system_session(lp), lp=lp)
 166
 167 runner = SubunitTestRunner()
 168 rc = 0
 169 if not runner.run(unittest.makeSuite(TokenTest)).wasSuccessful():
 170     rc = 1
 171 sys.exit(rc)