python2 reduction: Merge remaining compat code into common

[samba.git] / python / samba / emulate / traffic.py

diff --git a/python/samba/emulate/traffic.py b/python/samba/emulate/traffic.py
index c511fef6557e3a2492c6d909e4a1b1663ac26316..9b6bdb6af57c003361f887cbdda49334e57da173 100644 (file)
--- a/python/samba/emulate/traffic.py
+++ b/python/samba/emulate/traffic.py
@@ -53,7 +53,7 @@ from samba.dsdb import (
 from samba.dcerpc.misc import SEC_CHAN_BDC
 from samba import gensec
 from samba import sd_utils
-from samba.compat import get_string
+from samba.common import get_string
 from samba.logger import get_samba_logger
 import bisect
 
@@ -363,7 +363,8 @@ class ReplayContext(object):
                  ou=None,
                  base_dn=None,
                  domain=os.environ.get("DOMAIN"),
-                 domain_sid=None):
+                 domain_sid=None,
+                 instance_id=None):
         self.server                   = server
         self.netlogon_connection      = None
         self.creds                    = creds
@@ -379,6 +380,7 @@ class ReplayContext(object):
         self.global_tempdir           = tempdir
         self.domain_sid               = domain_sid
         self.realm                    = lp.get('realm')
+        self.instance_id              = instance_id
 
         # Bad password attempt controls
         self.badpassword_frequency    = badpassword_frequency
@@ -443,6 +445,61 @@ class ReplayContext(object):
         self.dn_map = dn_map
         self.attribute_clue_map = attribute_clue_map
 
+        # pre-populate DN-based search filters (it's simplest to generate them
+        # once, when the test starts). These are used by guess_search_filter()
+        # to avoid full-scans
+        self.search_filters = {}
+
+        # lookup all the GPO DNs
+        res = db.search(db.domain_dn(), scope=ldb.SCOPE_SUBTREE, attrs=['dn'],
+                        expression='(objectclass=groupPolicyContainer)')
+        gpos_by_dn = "".join("(distinguishedName={0})".format(msg['dn']) for msg in res)
+
+        # a search for the 'gPCFileSysPath' attribute is probably a GPO search
+        # (as per the MS-GPOL spec) which searches for GPOs by DN
+        self.search_filters['gPCFileSysPath'] = "(|{0})".format(gpos_by_dn)
+
+        # likewise, a search for gpLink is probably the Domain SOM search part
+        # of the MS-GPOL, in which case it's looking up a few OUs by DN
+        ou_str = ""
+        for ou in ["Domain Controllers,", "traffic_replay,", ""]:
+            ou_str += "(distinguishedName={0}{1})".format(ou, db.domain_dn())
+        self.search_filters['gpLink'] = "(|{0})".format(ou_str)
+
+        # The CEP Web Service can query the AD DC to get pKICertificateTemplate
+        # objects (as per MS-WCCE)
+        self.search_filters['pKIExtendedKeyUsage'] = \
+            '(objectCategory=pKICertificateTemplate)'
+
+        # assume that anything querying the usnChanged is some kind of
+        # synchronization tool, e.g. AD Change Detection Connector
+        res = db.search('', scope=ldb.SCOPE_BASE, attrs=['highestCommittedUSN'])
+        self.search_filters['usnChanged'] = \
+            '(usnChanged>={0})'.format(res[0]['highestCommittedUSN'])
+
+    # The traffic_learner script doesn't preserve the LDAP search filter, and
+    # having no filter can result in a full DB scan. This is costly for a large
+    # DB, and not necessarily representative of real world traffic. As there
+    # several standard LDAP queries that get used by AD tools, we can apply
+    # some logic and guess what the search filter might have been originally.
+    def guess_search_filter(self, attrs, dn_sig, dn):
+
+        # there are some standard spec-based searches that query fairly unique
+        # attributes. Check if the search is likely one of these
+        for key in self.search_filters.keys():
+            if key in attrs:
+                return self.search_filters[key]
+
+        # if it's the top-level domain, assume we're looking up a single user,
+        # e.g. like powershell Get-ADUser or a similar tool
+        if dn_sig == 'DC,DC':
+            random_user_id = random.random() % self.total_conversations
+            account_name = user_name(self.instance_id, random_user_id)
+            return '(&(sAMAccountName=%s)(objectClass=user))' % account_name
+
+        # otherwise just return everything in the sub-tree
+        return '(objectClass=*)'
+
     def generate_process_local_config(self, account, conversation):
         self.ldap_connections         = []
         self.dcerpc_connections       = []