s4-possibleinferiors.py: Fix usage of 'paged_search' module for remote LDB connections
[mat/samba.git] / source4 / dsdb / samdb / ldb_modules / tests / possibleinferiors.py
1 #!/usr/bin/env python
2
3 # Unix SMB/CIFS implementation.
4 # Copyright (C) Andrew Tridgell 2009
5 #
6 # This program is free software; you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation; either version 3 of the License, or
9 # (at your option) any later version.
10 #   
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14 # GNU General Public License for more details.
15 #   
16 # You should have received a copy of the GNU General Public License
17 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
18 #
19
20 """Tests the possibleInferiors generation in the schema_fsmo ldb module"""
21
22 import optparse
23 import sys
24
25
26 # Find right directory when running from source tree
27 sys.path.insert(0, "bin/python")
28
29 import samba
30 from samba import getopt as options, Ldb
31 import ldb
32
33 parser = optparse.OptionParser("possibleinferiors.py <URL> [<CLASS>]")
34 sambaopts = options.SambaOptions(parser)
35 parser.add_option_group(sambaopts)
36 credopts = options.CredentialsOptions(parser)
37 parser.add_option_group(credopts)
38 parser.add_option_group(options.VersionOptions(parser))
39 parser.add_option("--wspp", action="store_true")
40
41 opts, args = parser.parse_args()
42
43 if len(args) < 1:
44     parser.print_usage()
45     sys.exit(1)
46
47 url = args[0]
48 if (len(args) > 1):
49     objectclass = args[1]
50 else:
51     objectclass = None
52
53 def uniq_list(alist):
54     """return a unique list"""
55     set = {}
56     return [set.setdefault(e,e) for e in alist if e not in set]
57
58
59 lp_ctx = sambaopts.get_loadparm()
60
61 creds = credopts.get_credentials(lp_ctx)
62
63 ldb_options = []
64 # use 'paged_search' module when connecting remotely
65 if url.lower().startswith("ldap://"):
66     ldb_options = ["modules:paged_searches"]
67
68 db = Ldb(url, credentials=creds, lp=lp_ctx, options=ldb_options)
69
70 # get the rootDSE
71 res = db.search(base="", expression="",
72                 scope=ldb.SCOPE_BASE,
73                 attrs=["schemaNamingContext"])
74 rootDse = res[0]
75
76 schema_base = rootDse["schemaNamingContext"][0]
77
78 def possible_inferiors_search(db, oc):
79     """return the possible inferiors via a search for the possibleInferiors attribute"""
80     res = db.search(base=schema_base,
81                     expression=("ldapDisplayName=%s" % oc),
82                     attrs=["possibleInferiors"])
83
84     poss=[]
85     if len(res) == 0 or res[0].get("possibleInferiors") is None:
86         return poss
87     for item in res[0]["possibleInferiors"]:
88         poss.append(str(item))
89     poss = uniq_list(poss)
90     poss.sort()
91     return poss;
92
93
94
95 # see [MS-ADTS] section 3.1.1.4.5.21
96 # and section 3.1.1.4.2 for this algorithm
97
98 # !systemOnly=TRUE
99 # !objectClassCategory=2
100 # !objectClassCategory=3
101
102 def SUPCLASSES(classinfo, oc):
103     list = []
104     if oc == "top":
105         return list
106     if classinfo[oc].get("SUPCLASSES") is not None:
107         return classinfo[oc]["SUPCLASSES"]
108     res = classinfo[oc]["subClassOf"];
109     for r in res:
110         list.append(r)
111         list.extend(SUPCLASSES(classinfo,r))
112     classinfo[oc]["SUPCLASSES"] = list
113     return list
114
115 def AUXCLASSES(classinfo, oclist):
116     list = []
117     if oclist == []:
118         return list
119     for oc in oclist:
120         if classinfo[oc].get("AUXCLASSES") is not None:
121             list.extend(classinfo[oc]["AUXCLASSES"])
122         else:
123             list2 = []
124             list2.extend(classinfo[oc]["systemAuxiliaryClass"])
125             list2.extend(AUXCLASSES(classinfo, classinfo[oc]["systemAuxiliaryClass"]))
126             list2.extend(classinfo[oc]["auxiliaryClass"])
127             list2.extend(AUXCLASSES(classinfo, classinfo[oc]["auxiliaryClass"]))
128             list2.extend(AUXCLASSES(classinfo, SUPCLASSES(classinfo, oc)))
129             classinfo[oc]["AUXCLASSES"] = list2
130             list.extend(list2)
131     return list
132
133 def SUBCLASSES(classinfo, oclist):
134     list = []
135     for oc in oclist:
136         list.extend(classinfo[oc]["SUBCLASSES"])
137     return list
138
139 def POSSSUPERIORS(classinfo, oclist):
140     list = []
141     for oc in oclist:
142         if classinfo[oc].get("POSSSUPERIORS") is not None:
143             list.extend(classinfo[oc]["POSSSUPERIORS"])
144         else:
145             list2 = []
146             list2.extend(classinfo[oc]["systemPossSuperiors"])
147             list2.extend(classinfo[oc]["possSuperiors"])
148             list2.extend(POSSSUPERIORS(classinfo, SUPCLASSES(classinfo, oc)))
149             if opts.wspp:
150                 # the WSPP docs suggest we should do this:
151                 list2.extend(POSSSUPERIORS(classinfo, AUXCLASSES(classinfo, [oc])))
152             else:
153                 # but testing against w2k3 and w2k8 shows that we need to do this instead
154                 list2.extend(SUBCLASSES(classinfo, list2))
155             classinfo[oc]["POSSSUPERIORS"] = list2
156             list.extend(list2)
157     return list
158
159 def pull_classinfo(db):
160     """At startup we build a classinfo[] dictionary that holds all the information needed to construct the possible inferiors"""
161     classinfo = {}
162     res = db.search(base=schema_base,
163                     expression="objectclass=classSchema",
164                     attrs=["ldapDisplayName", "systemOnly", "objectClassCategory",
165                            "possSuperiors", "systemPossSuperiors",
166                            "auxiliaryClass", "systemAuxiliaryClass", "subClassOf"])
167     for r in res:
168         name = str(r["ldapDisplayName"][0])
169         classinfo[name] = {}
170         if str(r["systemOnly"]) == "TRUE":
171             classinfo[name]["systemOnly"] = True
172         else:
173             classinfo[name]["systemOnly"] = False
174         if r.get("objectClassCategory"):
175             classinfo[name]["objectClassCategory"] = int(r["objectClassCategory"][0])
176         else:
177             classinfo[name]["objectClassCategory"] = 0
178         for a in [ "possSuperiors", "systemPossSuperiors",
179                    "auxiliaryClass", "systemAuxiliaryClass",
180                    "subClassOf" ]:
181             classinfo[name][a] = []
182             if r.get(a):
183                 for i in r[a]:
184                     classinfo[name][a].append(str(i))
185
186     # build a list of subclasses for each class
187     def subclasses_recurse(subclasses, oc):
188         list = subclasses[oc]
189         for c in list:
190             list.extend(subclasses_recurse(subclasses, c))
191         return list
192
193     subclasses = {}
194     for oc in classinfo:
195         subclasses[oc] = []
196     for oc in classinfo:
197         for c in classinfo[oc]["subClassOf"]:
198             if not c == oc:
199                 subclasses[c].append(oc)
200     for oc in classinfo:
201         classinfo[oc]["SUBCLASSES"] = uniq_list(subclasses_recurse(subclasses, oc))
202
203     return classinfo
204
205 def is_in_list(list, c):
206     for a in list:
207         if c == a:
208             return True
209     return False
210
211 def possible_inferiors_constructed(db, classinfo, c):
212     list = []
213     for oc in classinfo:
214         superiors = POSSSUPERIORS(classinfo, [oc])
215         if (is_in_list(superiors, c) and
216             classinfo[oc]["systemOnly"] == False and
217             classinfo[oc]["objectClassCategory"] != 2 and
218             classinfo[oc]["objectClassCategory"] != 3):
219             list.append(oc)
220     list = uniq_list(list)
221     list.sort()
222     return list
223
224 def test_class(db, classinfo, oc):
225     """test to see if one objectclass returns the correct possibleInferiors"""
226     print "Testing objectClass %s" % oc
227     poss1 = possible_inferiors_search(db, oc)
228     poss2 = possible_inferiors_constructed(db, classinfo, oc)
229     if poss1 != poss2:
230         print "Returned incorrect list for objectclass %s" % oc
231         print "search:      %s" % poss1
232         print "constructed: %s" % poss2
233         for i in range(0,min(len(poss1),len(poss2))):
234             print "%30s %30s" % (poss1[i], poss2[i])
235         exit(1)
236
237 def get_object_classes(db):
238     """return a list of all object classes"""
239     list=[]
240     for item in classinfo:
241         list.append(item)
242     return list
243
244 classinfo = pull_classinfo(db)
245
246 if objectclass is None:
247     for oc in get_object_classes(db):
248         test_class(db,classinfo,oc)
249 else:
250     test_class(db,classinfo,objectclass)
251
252 print "Lists match OK"