diff options
| -rw-r--r-- | source4/scripting/python/samba/samdb.py | 103 | ||||
| -rw-r--r-- | source4/scripting/python/samba/tests/dsdb.py | 26 |
2 files changed, 128 insertions, 1 deletions
diff --git a/source4/scripting/python/samba/samdb.py b/source4/scripting/python/samba/samdb.py index f810926710..f358747b51 100644 --- a/source4/scripting/python/samba/samdb.py +++ b/source4/scripting/python/samba/samdb.py @@ -28,12 +28,16 @@ import samba import ldb import time import base64 +from samba.ndr import ndr_unpack, ndr_pack +from samba.dcerpc import drsblobs, misc __docformat__ = "restructuredText" class SamDB(samba.Ldb): """The SAM database.""" + hash_oid_name = {} + def __init__(self, url=None, lp=None, modules_dir=None, session_info=None, credentials=None, flags=0, options=None, global_schema=True, auto_connect=True, am_rodc=False): @@ -470,5 +474,104 @@ accountExpires: %u def set_schema_from_ldb(self, ldb): dsdb._dsdb_set_schema_from_ldb(self, ldb) + def get_attribute_from_attid(self, attid): + """ Get from an attid the associated attribute + + :param attid: The attribute id for searched attribute + :return: The name of the attribute associated with this id + """ + if len(self.hash_oid_name.keys()) == 0: + self._populate_oid_attid() + if self.hash_oid_name.has_key(self.get_oid_from_attid(attid)): + return self.hash_oid_name[self.get_oid_from_attid(attid)] + else: + return None + + + def _populate_oid_attid(self): + """Populate the hash hash_oid_name + + This hash contains the oid of the attribute as a key and + its display name as a value + """ + self.hash_oid_name = {} + res = self.search(expression="objectClass=attributeSchema", + controls=["search_options:1:2"], + attrs=["attributeID", + "lDAPDisplayName"]) + if len(res) > 0: + for e in res: + strDisplay = str(e.get("lDAPDisplayName")) + self.hash_oid_name[str(e.get("attributeID"))] = strDisplay + + + def get_attribute_replmetadata_version(self, dn, att): + """ Get the version field trom the replPropertyMetaData for + the given field + + :param dn: The on which we want to get the version + :param att: The name of the attribute + :return: The value of the version field in the replPropertyMetaData + for the given attribute. None if the attribute is not replicated + """ + + res = self.search(expression="dn=%s" % dn, + scope=ldb.SCOPE_SUBTREE, + controls=["search_options:1:2"], + attrs=["replPropertyMetaData"]) + if len(res) == 0: + return None + + repl = ndr_unpack(drsblobs.replPropertyMetaDataBlob, + str(res[0]["replPropertyMetaData"])) + ctr = repl.ctr + if len(self.hash_oid_name.keys()) == 0: + self._populate_oid_attid() + for o in ctr.array: + # Search for Description + att_oid = self.get_oid_from_attid(o.attid) + if self.hash_oid_name.has_key(att_oid) and\ + att.lower() == self.hash_oid_name[att_oid].lower(): + return o.version + return None + + + def set_attribute_replmetadata_version(self, dn, att, value): + res = self.search(expression="dn=%s" % dn, + scope=ldb.SCOPE_SUBTREE, + controls=["search_options:1:2"], + attrs=["replPropertyMetaData"]) + if len(res) == 0: + return None + + repl = ndr_unpack(drsblobs.replPropertyMetaDataBlob, + str(res[0]["replPropertyMetaData"])) + ctr = repl.ctr + now = samba.unix2nttime(int(time.time())) + found = False + if len(self.hash_oid_name.keys()) == 0: + self._populate_oid_attid() + for o in ctr.array: + # Search for Description + att_oid = self.get_oid_from_attid(o.attid) + if self.hash_oid_name.has_key(att_oid) and\ + att.lower() == self.hash_oid_name[att_oid].lower(): + found = True + seq = self.sequence_number(ldb.SEQ_NEXT) + o.version = value + o.originating_change_time = now + o.originating_invocation_id = misc.GUID(self.get_invocation_id()) + o.originating_usn = seq + o.local_usn = seq + if found : + replBlob = ndr_pack(repl) + msg = ldb.Message() + msg.dn = res[0].dn + msg["replPropertyMetaData"] = ldb.MessageElement(replBlob, + ldb.FLAG_MOD_REPLACE, + "replPropertyMetaData") + self.modify(msg, ["local_oid:1.3.6.1.4.1.7165.4.3.14:0"]) + + def write_prefixes_from_schema(self): dsdb._dsdb_write_prefixes_from_schema_to_ldb(self) diff --git a/source4/scripting/python/samba/tests/dsdb.py b/source4/scripting/python/samba/tests/dsdb.py index d28be82984..4a50c96e25 100644 --- a/source4/scripting/python/samba/tests/dsdb.py +++ b/source4/scripting/python/samba/tests/dsdb.py @@ -25,7 +25,7 @@ from samba.ndr import ndr_unpack, ndr_pack from samba.dcerpc import drsblobs import ldb import os -import samba.dsdb +import samba class DsdbTests(TestCase): @@ -107,3 +107,27 @@ class DsdbTests(TestCase): msg.dn = res[0].dn msg["replPropertyMetaData"] = ldb.MessageElement(replBlob, ldb.FLAG_MOD_REPLACE, "replPropertyMetaData") self.samdb.modify(msg, ["local_oid:1.3.6.1.4.1.7165.4.3.14:0"]) + + def test_ok_get_attribute_from_attid(self): + self.assertEquals(self.samdb.get_attribute_from_attid(13), "description") + + def test_ko_get_attribute_from_attid(self): + self.assertEquals(self.samdb.get_attribute_from_attid(11979), None) + + def test_get_attribute_replmetadata_version(self): + res = self.samdb.search(expression="cn=Administrator", + scope=ldb.SCOPE_SUBTREE, + attrs=["dn"]) + self.assertEquals(len(res), 1) + dn = str(res[0].dn) + self.assertEqual(self.samdb.get_attribute_replmetadata_version(dn, "unicodePwd"), 1) + + def test_set_attribute_replmetadata_version(self): + res = self.samdb.search(expression="cn=Administrator", + scope=ldb.SCOPE_SUBTREE, + attrs=["dn"]) + self.assertEquals(len(res), 1) + dn = str(res[0].dn) + version = self.samdb.get_attribute_replmetadata_version(dn, "description") + self.samdb.set_attribute_replmetadata_version(dn, "description", version + 2) + self.assertEqual(self.samdb.get_attribute_replmetadata_version(dn, "description"), version + 2) |