JezK
Edit File: complex_expressions.py
# -*- coding: utf-8 -*- # Copyright Andrew Bartlett 2018 # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. # import optparse import samba import samba.getopt as options import sys import os import time from samba.auth import system_session from samba.tests import TestCase import ldb ERRCODE_ENTRY_EXISTS = 68 ERRCODE_OPERATIONS_ERROR = 1 ERRCODE_INVALID_VALUE = 21 ERRCODE_CLASS_VIOLATION = 65 parser = optparse.OptionParser("{0} <host>".format(sys.argv[0])) sambaopts = options.SambaOptions(parser) # use command line creds if available credopts = options.CredentialsOptions(parser) parser.add_option_group(credopts) parser.add_option("-v", action="store_true", dest="verbose", help="print successful expression outputs") opts, args = parser.parse_args() if len(args) < 1: parser.print_usage() sys.exit(1) lp = sambaopts.get_loadparm() creds = credopts.get_credentials(lp) # Set properly at end of file. host = None global ou_count ou_count = 0 class ComplexExpressionTests(TestCase): # Using setUpClass instead of setup because we're not modifying any # records in the tests @classmethod def setUpClass(cls): super(ComplexExpressionTests, cls).setUpClass() cls.samdb = samba.samdb.SamDB(host, lp=lp, session_info=system_session(), credentials=creds) ou_name = "ComplexExprTest" cls.base_dn = "OU={0},{1}".format(ou_name, cls.samdb.domain_dn()) try: cls.samdb.delete(cls.base_dn, ["tree_delete:1"]) except: pass try: cls.samdb.create_ou(cls.base_dn) except ldb.LdbError as e: if e.args[0] == ERRCODE_ENTRY_EXISTS: print(('test ou {ou} already exists. Delete with ' '"samba-tool group delete OU={ou} ' '--force-subtree-delete"').format(ou=ou_name)) raise e cls.name_template = "testuser{0}" cls.default_n = 10 # These fields are carefully hand-picked from the schema. They have # syntax and handling appropriate for our test structure. cls.largeint_f = "accountExpires" cls.str_f = "accountNameHistory" cls.int_f = "flags" cls.enum_f = "preferredDeliveryMethod" cls.time_f = "msTSExpireDate" cls.ranged_int_f = "countryCode" @classmethod def tearDownClass(cls): cls.samdb.delete(cls.base_dn, ["tree_delete:1"]) # Make test OU containing users with field=val for each val def make_test_objects(self, field, vals): global ou_count ou_count += 1 ou_dn = "OU=testou{0},{1}".format(ou_count, self.base_dn) self.samdb.create_ou(ou_dn) ldap_objects = [{"dn": "CN=testuser{0},{1}".format(n, ou_dn), "name": self.name_template.format(n), "objectClass": "user", field: n} for n in vals] for ldap_object in ldap_objects: # It's useful to keep appropriate python types in the ldap_object # dict but samdb's 'add' function expects strings. stringed_ldap_object = {k: str(v) for (k, v) in ldap_object.items()} try: self.samdb.add(stringed_ldap_object) except ldb.LdbError as e: print("failed to add %s" % (stringed_ldap_object)) raise e return ou_dn, ldap_objects # Run search expr and print out time. This function should be used for # almost all searching. def time_ldap_search(self, expr, dn): time_taken = 0 try: start_time = time.time() res = self.samdb.search(base=dn, scope=ldb.SCOPE_SUBTREE, expression=expr) time_taken = time.time() - start_time except Exception as e: print("failed expr " + expr) raise e print("{0} took {1}s".format(expr, time_taken)) return res, time_taken # Take an ldap expression and an equivalent python expression. # Run and time the ldap expression and compare the result to the python # expression run over the a list of ldap_object dicts. def assertLDAPQuery(self, ldap_expr, ou_dn, py_expr, ldap_objects): # run (and time) the LDAP search expression over the DB res, time_taken = self.time_ldap_search(ldap_expr, ou_dn) results = {str(row.get('name')[0]) for row in res} # build the set of expected results by evaluating the python-equivalent # of the search expression over the same set of objects expected_results = set() for ldap_object in ldap_objects: try: final_expr = py_expr.format(**ldap_object) except KeyError: # If the format on the py_expr hits a key error, then # ldap_object doesn't have the field, so it shouldn't match. continue if eval(final_expr): expected_results.add(str(ldap_object['name'])) self.assertEqual(results, expected_results) if opts.verbose: ldap_object_names = {l['name'] for l in ldap_objects} excluded = ldap_object_names - results excluded = "\n ".join(excluded) or "[NOTHING]" returned = "\n ".join(expected_results) or "[NOTHING]" print("PASS: Expression {0} took {1}s and returned:" "\n {2}\n" "Excluded:\n {3}\n".format(ldap_expr, time_taken, returned, excluded)) # Basic integer range test def test_int_range(self, field=None): n = self.default_n field = field or self.int_f ou_dn, ldap_objects = self.make_test_objects(field, range(n)) expr = "(&(%s>=%s)(%s<=%s))" % (field, n-1, field, n+1) py_expr = "%d <= {%s} <= %d" % (n-1, field, n+1) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) half_n = int(n/2) expr = "(%s<=%s)" % (field, half_n) py_expr = "{%s} <= %d" % (field, half_n) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) expr = "(%s>=%s)" % (field, half_n) py_expr = "{%s} >= %d" % (field, half_n) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) # Same test again for largeint and enum def test_largeint_range(self): self.test_int_range(self.largeint_f) def test_enum_range(self): self.test_int_range(self.enum_f) # Special range test for integer field with upper and lower bounds defined. # The bounds are checked on insertion, not search, so we should be able # to compare to a constant that is outside bounds. def test_ranged_int_range(self): field = self.ranged_int_f ubound = 2**16 width = 8 vals = list(range(ubound-width, ubound)) ou_dn, ldap_objects = self.make_test_objects(field, vals) # Check <= value above overflow returns all vals expr = "(%s<=%d)" % (field, ubound+5) py_expr = "{%s} <= %d" % (field, ubound+5) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) # Test range also works for time fields def test_time_range(self): n = self.default_n field = self.time_f n = self.default_n width = int(n/2) base_time = 20050116175514 time_range = [base_time + t for t in range(-width, width)] time_range = [str(t) + ".0Z" for t in time_range] ou_dn, ldap_objects = self.make_test_objects(field, time_range) expr = "(%s<=%s)" % (field, str(base_time) + ".0Z") py_expr = 'int("{%s}"[:-3]) <= %d' % (field, base_time) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) expr = "(&(%s>=%s)(%s<=%s))" % (field, str(base_time-1) + ".0Z", field, str(base_time+1) + ".0Z") py_expr = '%d <= int("{%s}"[:-3]) <= %d' % (base_time-1, field, base_time+1) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) # Run each comparison op on a simple test set. Time taken will be printed. def test_int_single_cmp_op_speeds(self, field=None): n = self.default_n field = field or self.int_f ou_dn, ldap_objects = self.make_test_objects(field, range(n)) comp_ops = ['=', '<=', '>='] py_comp_ops = ['==', '<=', '>='] exprs = ["(%s%s%d)" % (field, c, n) for c in comp_ops] py_exprs = ["{%s}%s%d" % (field, c, n) for c in py_comp_ops] for expr, py_expr in zip(exprs, py_exprs): self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) def test_largeint_single_cmp_op_speeds(self): self.test_int_single_cmp_op_speeds(self.largeint_f) def test_enum_single_cmp_op_speeds(self): self.test_int_single_cmp_op_speeds(self.enum_f) # Check strings are ordered using a naive ordering. def test_str_ordering(self): field = self.str_f a_ord = ord('A') n = 10 str_range = ['abc{0}d'.format(chr(c)) for c in range(a_ord, a_ord+n)] ou_dn, ldap_objects = self.make_test_objects(field, str_range) half_n = int(a_ord + n/2) # Basic <= and >= statements expr = "(%s>=abc%s)" % (field, chr(half_n)) py_expr = "'{%s}' >= 'abc%s'" % (field, chr(half_n)) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) expr = "(%s<=abc%s)" % (field, chr(half_n)) py_expr = "'{%s}' <= 'abc%s'" % (field, chr(half_n)) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) # String range expr = "(&(%s>=abc%s)(%s<=abc%s))" % (field, chr(half_n-2), field, chr(half_n+2)) py_expr = "'abc%s' <= '{%s}' <= 'abc%s'" % (chr(half_n-2), field, chr(half_n+2)) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) # Integers treated as string expr = "(%s>=1)" % (field) py_expr = "'{%s}' >= '1'" % (field) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) # Windows returns nothing for invalid expressions. Expected fail on samba. def test_invalid_expressions(self, field=None): field = field or self.int_f n = self.default_n ou_dn, ldap_objects = self.make_test_objects(field, list(range(n))) int_expressions = ["(%s>=abc)", "(%s<=abc)", "(%s=abc)"] for expr in int_expressions: expr = expr % (field) self.assertLDAPQuery(expr, ou_dn, "False", ldap_objects) def test_largeint_invalid_expressions(self): self.test_invalid_expressions(self.largeint_f) def test_enum_invalid_expressions(self): self.test_invalid_expressions(self.enum_f) def test_case_insensitive(self): str_range = ["äbc"+str(n) for n in range(10)] ou_dn, ldap_objects = self.make_test_objects(self.str_f, str_range) expr = "(%s=äbc1)" % (self.str_f) pyexpr = '"{%s}"=="äbc1"' % (self.str_f) self.assertLDAPQuery(expr, ou_dn, pyexpr, ldap_objects) expr = "(%s=ÄbC1)" % (self.str_f) self.assertLDAPQuery(expr, ou_dn, pyexpr, ldap_objects) # Check negative numbers can be entered and compared def test_negative_cmp(self, field=None): field = field or self.int_f width = 6 around_zero = list(range(-width, width)) ou_dn, ldap_objects = self.make_test_objects(field, around_zero) expr = "(%s>=-3)" % (field) py_expr = "{%s} >= -3" % (field) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) def test_negative_cmp_largeint(self): self.test_negative_cmp(self.largeint_f) def test_negative_cmp_enum(self): self.test_negative_cmp(self.enum_f) # Check behaviour on insertion and comparison of zero-prefixed numbers. # Samba errors on insertion, Windows strips the leading zeroes. def test_zero_prefix(self, field=None): field = field or self.int_f # Test comparison with 0-prefixed constants. n = self.default_n ou_dn, ldap_objects = self.make_test_objects(field, list(range(n))) expr = "(%s>=00%d)" % (field, n/2) py_expr = "{%s} >= %d" % (field, n/2) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) # Delete the test OU so we don't mix it up with the next one. self.samdb.delete(ou_dn, ["tree_delete:1"]) # Try inserting 0-prefixed numbers, check it fails. zero_pref_nums = ['00'+str(num) for num in range(n)] try: ou_dn, ldap_objects = self.make_test_objects(field, zero_pref_nums) except ldb.LdbError as e: if e.args[0] != ERRCODE_INVALID_VALUE: raise e return # Samba doesn't get this far - the exception is raised. Windows allows # the insertion and removes the leading 0s as tested below. # Either behaviour is fine. print("LDAP allowed insertion of 0-prefixed nums for field " + field) res = self.samdb.search(base=ou_dn, scope=ldb.SCOPE_SUBTREE, expression="(objectClass=user)") returned_nums = [str(r.get(field)[0]) for r in res] expect = [str(n) for n in range(n)] self.assertEqual(set(returned_nums), set(expect)) expr = "(%s>=%d)" % (field, n/2) py_expr = "{%s} >= %d" % (field, n/2) for ldap_object in ldap_objects: ldap_object[field] = int(ldap_object[field]) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) def test_zero_prefix_largeint(self): self.test_zero_prefix(self.largeint_f) def test_zero_prefix_enum(self): self.test_zero_prefix(self.enum_f) # Check integer overflow is handled as best it can be. def test_int_overflow(self, field=None, of=None): field = field or self.int_f of = of or 2**31-1 width = 8 vals = list(range(of-width, of+width)) ou_dn, ldap_objects = self.make_test_objects(field, vals) # Check ">=overflow" doesn't return vals past overflow expr = "(%s>=%d)" % (field, of-3) py_expr = "%d <= {%s} <= %d" % (of-3, field, of) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) # "<=overflow" returns everything expr = "(%s<=%d)" % (field, of) py_expr = "True" self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) # Values past overflow should be negative expr = "(&(%s<=%d)(%s>=0))" % (field, of, field) py_expr = "{%s} <= %d" % (field, of) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) expr = "(%s<=0)" % (field) py_expr = "{%s} >= %d" % (field, of+1) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) # Get the values back out and check vals past overflow are negative. res = self.samdb.search(base=ou_dn, scope=ldb.SCOPE_SUBTREE, expression="(objectClass=user)") returned_nums = [str(r.get(field)[0]) for r in res] # Note: range(a,b) == [a..b-1] (confusing) up_to_overflow = list(range(of-width, of+1)) negatives = list(range(-of-1, -of+width-2)) expect = [str(n) for n in up_to_overflow + negatives] self.assertEqual(set(returned_nums), set(expect)) def test_enum_overflow(self): self.test_int_overflow(self.enum_f, 2**31-1) # Check cmp works on uSNChanged. We can't insert uSNChanged vals, they get # added automatically so we'll just insert some objects and go with what # we get. def test_usnchanged(self): field = "uSNChanged" n = 10 # Note we can't actually set uSNChanged via LDAP (LDB ignores it), # so the input val range doesn't matter here ou_dn, _ = self.make_test_objects(field, list(range(n))) # Get the assigned uSNChanged values res = self.samdb.search(base=ou_dn, scope=ldb.SCOPE_SUBTREE, expression="(objectClass=user)") # Our vals got ignored so make ldap_objects from search result ldap_objects = [{'name': str(r['name'][0]), field: int(r[field][0])} for r in res] # Get the median val and use as the number in the test search expr. nums = [l[field] for l in ldap_objects] nums = list(sorted(nums)) search_num = nums[int(len(nums)/2)] expr = "(&(%s<=%d)(objectClass=user))" % (field, search_num) py_expr = "{%s} <= %d" % (field, search_num) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) expr = "(&(%s>=%d)(objectClass=user))" % (field, search_num) py_expr = "{%s} >= %d" % (field, search_num) self.assertLDAPQuery(expr, ou_dn, py_expr, ldap_objects) # If we're called independently then import subunit, get host from first # arg and run. Otherwise, subunit ran us so just set host from env. # We always try to run over LDAP rather than direct file, so that # search timings are not impacted by opening and closing the tdb file. if __name__ == "__main__": from samba.tests.subunitrun import TestProgram host = args[0] if "://" not in host: if os.path.isfile(host): host = "tdb://%s" % host else: host = "ldap://%s" % host TestProgram(module=__name__) else: host = "ldap://" + os.getenv("SERVER")