token_group: Use samba.tests.subunitrun.
[Samba.git] / source4 / dsdb / tests / python / token_group.py
blob94a45b1810fd24993264263c2bf4eccfa1daad9e
1 #!/usr/bin/env python
2 # -*- coding: utf-8 -*-
3 # test tokengroups attribute against internal token calculation
5 import optparse
6 import sys
7 import os
9 sys.path.insert(0, "bin/python")
10 import samba
12 from samba.tests.subunitrun import SubunitOptions, TestProgram
14 import samba.getopt as options
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
24 import samba.tests
26 from samba.auth import AUTH_SESSION_INFO_DEFAULT_GROUPS, AUTH_SESSION_INFO_AUTHENTICATED, AUTH_SESSION_INFO_SIMPLE_PRIVILEGES
29 parser = optparse.OptionParser("ldap.py [options] <host>")
30 sambaopts = options.SambaOptions(parser)
31 parser.add_option_group(sambaopts)
32 parser.add_option_group(options.VersionOptions(parser))
33 # use command line creds if available
34 credopts = options.CredentialsOptions(parser)
35 parser.add_option_group(credopts)
36 subunitopts = SubunitOptions(parser)
37 parser.add_option_group(subunitopts)
38 opts, args = parser.parse_args()
40 if len(args) < 1:
41 parser.print_usage()
42 sys.exit(1)
44 url = args[0]
46 lp = sambaopts.get_loadparm()
47 creds = credopts.get_credentials(lp)
49 class TokenTest(samba.tests.TestCase):
51 def setUp(self):
52 super(TokenTest, self).setUp()
53 self.ldb = samdb
54 self.base_dn = samdb.domain_dn()
56 res = self.ldb.search("", scope=ldb.SCOPE_BASE, attrs=["tokenGroups"])
57 self.assertEquals(len(res), 1)
59 self.user_sid_dn = "<SID=%s>" % str(ndr_unpack(samba.dcerpc.security.dom_sid, res[0]["tokenGroups"][0]))
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)
67 token = session.security_token
68 self.user_sids = []
69 for s in token.sids:
70 self.user_sids.append(str(s))
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")
77 res = self.ldb.search("", scope=ldb.SCOPE_BASE, attrs=["tokenGroups"])
78 self.assertEquals(len(res), 1)
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)))
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")
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)
99 dn_tokengroups = []
100 for sid in res[0]['tokenGroups']:
101 dn_tokengroups.append(str(ndr_unpack(samba.dcerpc.security.dom_sid, sid)))
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")
110 def test_pac_groups(self):
111 settings = {}
112 settings["lp_ctx"] = lp
113 settings["target_hostname"] = lp.get("netbios name")
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")
120 auth_context = AuthContext(lp_ctx=lp, ldb=self.ldb, methods=[])
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)
128 gensec_server.want_feature(gensec.FEATURE_SEAL)
129 gensec_server.start_mech_by_sasl_name("GSSAPI")
131 client_finished = False
132 server_finished = False
133 server_to_client = ""
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)
144 session = gensec_server.session_info()
146 token = session.security_token
147 pac_sids = []
148 for s in token.sids:
149 pac_sids.append(str(s))
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")
159 if not "://" in url:
160 if os.path.isfile(url):
161 url = "tdb://%s" % url
162 else:
163 url = "ldap://%s" % url
165 samdb = SamDB(url, credentials=creds, session_info=system_session(lp), lp=lp)
167 TestProgram(module=__name__, opts=subunitopts)