python-ldap · mistotebe · Jun 25, 2024
diff --git a/Doc/reference/ldap-syncrepl.rst b/Doc/reference/ldap-syncrepl.rst
@@ -20,3 +20,6 @@ This module defines the following classes:
 
 .. autoclass:: ldap.syncrepl.SyncreplConsumer
    :members:
+
+.. autoclass:: ldap.syncrepl.OpenLDAPSyncreplCookie
+   :members:
diff --git a/Lib/ldap/syncrepl.py b/Lib/ldap/syncrepl.py
@@ -4,6 +4,7 @@
 See https://www.python-ldap.org/ for project details.
 """
 
+from typing import AnyStr, Dict, List, Tuple, Union
 from uuid import UUID
 
 # Imports from pyasn1
@@ -535,3 +536,72 @@ def syncrepl_refreshdone(self):
         follows.
         """
         pass
+
+
+class OpenLDAPSyncreplCookie:
+    """
+    OpenLDAPSyncreplCookie - allows a consumer to track a cookie across a
+    refreshAndPersist syncrepl session against a multi-provider OpenLDAP cluster
+    """
+
+    rid: int = 0
+    sid: int = 0
+    _csnset: Dict[int, str]
+
+    def __init__(self, cookie: AnyStr = "") -> None:
+        self._csnset = {}
+
+        if cookie:
+            self.update(cookie)
+
+    def _parse_csn(self, csn: str) -> Tuple[str, str, str, str]:
+        time, order, sid, other = csn.split('#', 3)
+        return (time, order, sid, other)
+
+    def _parse_cookie(self, cookie: AnyStr) -> Dict[str, Union[str, List[str]]]:
+        if isinstance(cookie, bytes):
+            cookie = cookie.decode()
+
+        result = {}
+        parts = cookie.split(',')
+        for part in parts:
+            if part.startswith('rid='):
+                result['rid'] = part[4:]
+            elif part.startswith('sid='):
+                result['sid'] = part[4:]
+            elif part.startswith('csn='):
+                result['csn'] = part[4:].split(';')
+            elif part.startswith('delcsn='):
+                result['delcsn'] = part[7:]
+            else:
+                # Did not recognise this cookie part
+                pass
+        return result
+
+    def update(self, cookie: str):
+        """
+        Update the CSN set based on a cookie we just received, use in
+        syncrepl_set_cookie() to track the session state.
+        """
+        components = self._parse_cookie(cookie)
+        for csn in components.get('csn', []):
+            _, _, sid, _ = self._parse_csn(csn)
+            if sid not in self._csnset or self._csnset[sid] < csn:
+                self._csnset[sid] = csn
+
+        return self
+
+    def unparse(self) -> str:
+        """
+        Return the cookie as a string, use in syncrepl_get_cookie() or when
+        storing the state for later use.
+        """
+        cookie = 'rid={:03},sid={:03x}'.format(self.rid or 0, self.sid or 0)
+        if self._csnset:
+            cookie += ',csn='
+            cookie += ';'.join(csn
+                               for sid, csn in sorted(self._csnset.items()))
+        return cookie
+
+    def __str__(self):
+        return self.unparse()
diff --git a/Lib/slapdtest/_slapdtest.py b/Lib/slapdtest/_slapdtest.py
@@ -41,6 +41,11 @@
 cn: module
 olcModuleLoad: back_%(database)s
 
+dn: olcDatabase=config,cn=config
+objectClass: olcDatabaseConfig
+olcDatabase: config
+olcRootDN: %(rootdn)s
+
 dn: olcDatabase=%(database)s,cn=config
 objectClass: olcDatabaseConfig
 objectClass: olcMdbConfig

diff --git a/Tests/t_ldap_syncrepl.py b/Tests/t_ldap_syncrepl.py
@@ -13,7 +13,8 @@
 
 import ldap
 from ldap.ldapobject import SimpleLDAPObject
-from ldap.syncrepl import SyncreplConsumer, SyncInfoMessage
+from ldap.syncrepl import SyncreplConsumer, SyncInfoMessage, \
+        OpenLDAPSyncreplCookie
 
 from slapdtest import SlapdObject, SlapdTestCase
 
@@ -37,6 +38,10 @@
 olcModuleLoad: back_%(database)s
 olcModuleLoad: syncprov
 
+dn: olcDatabase=config,cn=config
+objectClass: olcDatabaseConfig
+olcRootDN: %(rootdn)s
+
 dn: olcDatabase=%(database)s,cn=config
 objectClass: olcDatabaseConfig
 objectClass: olcMdbConfig
@@ -442,6 +447,174 @@ def setUp(self):
         self.suffix = self.server.suffix
 
 
+class TestMPRSyncrepl(BaseSyncreplTests, SlapdTestCase):
+    class MPRClient(SyncreplClient):
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+            self.cookie = OpenLDAPSyncreplCookie()
+
+        def syncrepl_set_cookie(self, cookie):
+            self.cookie.update(cookie)
+            super().syncrepl_set_cookie(self.cookie.unparse())
+
+    def setUp(self):
+        super().setUp()
+        self.tester = self.MPRClient(
+            self.server.ldap_uri,
+            self.server.root_dn,
+            self.server.root_pw,
+            bytes_mode=False
+        )
+        self.suffix = self.server.suffix
+
+        # An active MPR should not have a sid=000 server in it
+        if self.server.server_id == 0:
+            self.skip("Server got serverid 0 assigned")
+
+    def test_mpr_refresh_and_persist(self):
+        """
+        Make sure we process cookie updates from a live MPR cluster correctly
+        """
+        # Assumes that server_id is not used before the call to start()
+        self.server2 = self.server_class()
+        if self.server.server_id == self.server2.server_id:
+            self.server2.server_id += 1
+            if self.server2.server_id % 4096 == 0:
+                self.server2.server_id = 1
+
+        with self.server2 as server2:
+            tester2 = self.MPRClient(
+                self.server2.ldap_uri,
+                self.server2.root_dn,
+                self.server2.root_pw,
+                bytes_mode=False
+            )
+            self.addCleanup(tester2.unbind_s)
+
+            self.tester.search(
+                self.suffix,
+                'refreshAndPersist',
+            )
+
+            # Run a quick refresh, that shouldn't have any changes.
+            while self.tester.refresh_done is not True:
+                poll_result = self.tester.poll(
+                    all=0,
+                    timeout=None
+                )
+                self.assertTrue(poll_result)
+
+            # Again, server data should not have changed.
+            self.assertEqual(self.tester.dn_attrs, LDAP_ENTRIES)
+
+            # set up replication between both
+            coords = [(1, self.server.ldap_uri, self.suffix,
+                       self.server.root_dn, self.server.root_pw),
+                      (2, self.server2.ldap_uri, self.suffix,
+                       self.server2.root_dn, self.server2.root_pw),
+            ]
+            modifications = [
+                (ldap.MOD_ADD, "olcSyncrepl", [
+                    ('rid=%d provider=%s searchbase="%s" type=refreshAndPersist '
+                     'bindmethod=simple binddn="%s" credentials="%s" '
+                     'retry="1 +"' % coord).encode() for coord in coords]),
+                # do we still support 2.4.x? Change to olcMultiProvider if not
+                (ldap.MOD_REPLACE, "olcMirrorMode", [b"TRUE"]),
+            ]
+
+            self.tester.modify_s(
+                "olcDatabase={1}%s,cn=config" % (self.server.database),
+                modifications)
+            tester2.modify_s(
+                "olcDatabase={1}%s,cn=config" % (self.server.database),
+                modifications)
+
+            tester2.search(
+                self.suffix,
+                'refreshAndPersist',
+            )
+
+            # Wait till server2 catches up
+            while tester2.refresh_done is not True or \
+                    tester2.cookie.unparse() != self.tester.cookie.unparse():
+                try:
+                    poll_result = tester2.poll(
+                        all=0,
+                        timeout=None
+                    )
+                    self.assertTrue(poll_result)
+                except ldap.NO_SUCH_OBJECT:
+                    # 2.6+ Allows a refreshAndPersist against an empty DB, but
+                    # with older ones we need to retry until there's at least
+                    # one entry
+                    tester2.search(
+                        self.suffix,
+                        'refreshAndPersist',
+                    )
+
+            # Again, server data should not have changed.
+            self.assertEqual(tester2.dn_attrs, LDAP_ENTRIES)
+
+            # From here on, things get little hairy, server1 might not have
+            # finished its refresh from 2 and we can't easily confirm this
+            # without cn=monitor. We just read back our CSNs and make sure
+            # we've seen both.
+
+            # send some mods to both
+            modification = [('objectClass', [b'device'])]
+            self.tester.add_s("cn=server1,%s" % self.suffix, modification)
+
+            csn1 = self.tester.read_s("cn=server1,%s" % self.suffix,
+                                      attrlist=['entryCSN']
+                                      )['entryCSN'][0].decode('utf8')
+
+            tester2.add_s("cn=server2,%s" % self.suffix, modification)
+            csn2 = tester2.read_s("cn=server2,%s" % self.suffix,
+                                  attrlist=['entryCSN']
+                                  )['entryCSN'][0].decode('utf8')
+
+            new_state = LDAP_ENTRIES.copy()
+            new_state["cn=server1,%s" % self.suffix] = {
+                "objectClass": [b"device"],
+                "cn": [b"server1"],
+            }
+            new_state["cn=server2,%s" % self.suffix] = {
+                "objectClass": [b"device"],
+                "cn": [b"server2"],
+            }
+
+            # Wait for the cookie to sync up, a failure would be that this
+            # doesn't happen, so impose a timeout
+            while csn1 not in self.tester.cookie.unparse() or \
+                    csn2 not in self.tester.cookie.unparse() or \
+                    csn1 not in tester2.cookie.unparse() or \
+                    csn2 not in tester2.cookie.unparse():
+                if csn1 not in self.tester.cookie.unparse() or \
+                        csn2 not in self.tester.cookie.unparse():
+                    poll_result = self.tester.poll(
+                        all=0,
+                        timeout=5
+                    )
+                    self.assertTrue(poll_result)
+                if csn1 not in tester2.cookie.unparse() or \
+                        csn2 not in tester2.cookie.unparse():
+                    poll_result = tester2.poll(
+                        all=0,
+                        timeout=5
+                    )
+                    self.assertTrue(poll_result)
+
+            self.assertEqual(self.tester.cookie.unparse(),
+                             tester2.cookie.unparse())
+            self.assertEqual(self.tester.dn_attrs, new_state)
+            self.assertEqual(tester2.dn_attrs, new_state)
+
+            # self.tester seems to have been unbound by the time
+            # self.addCleanup callbacks get called? Cleanup manually...
+            self.tester.delete_s("cn=server1,%s" % self.suffix)
+            self.tester.delete_s("cn=server2,%s" % self.suffix)
+
+
 class DecodeSyncreplProtoTests(unittest.TestCase):
     """
     Tests of the ASN.1 decoder for tricky cases or past issues to ensure that