Repository navigation
Expand file tree
/
Copy patht_ldap_syncrepl.py
More file actions
661 lines (559 loc) · 21.3 KB
/
Copy patht_ldap_syncrepl.py
File metadata and controls
661 lines (559 loc) · 21.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
"""
Automatic tests for python-ldap's module ldap.syncrepl
See https://www.python-ldap.org/ for details.
"""
import os
import shelve
import unittest
import binascii
# Switch off processing .ldaprc or ldap.conf before importing _ldap
os.environ['LDAPNOINIT'] = '1'
import ldap
from ldap.ldapobject import SimpleLDAPObject
from ldap.syncrepl import SyncreplConsumer, SyncInfoMessage, \
OpenLDAPSyncreplCookie
from slapdtest import SlapdObject, SlapdTestCase
# a template string for generating simple slapd.conf file
SLAPD_CONF_PROVIDER_TEMPLATE = r"""dn: cn=config
objectClass: olcGlobal
cn: config
olcServerID: %(serverid)s
olcLogLevel: %(loglevel)s
olcAllows: bind_v2
olcAuthzRegexp: {0}"gidnumber=%(root_gid)s\+uidnumber=%(root_uid)s,cn=peercred,cn=external,cn=auth" "%(rootdn)s"
olcAuthzRegexp: {1}"C=DE, O=python-ldap, OU=slapd-test, CN=([A-Za-z]+)" "ldap://ou=people,dc=local???($1)"
olcTLSCACertificateFile: %(cafile)s
olcTLSCertificateFile: %(servercert)s
olcTLSCertificateKeyFile: %(serverkey)s
olcTLSVerifyClient: try
dn: cn=module,cn=config
objectClass: olcModuleList
cn: module
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
olcDatabase: %(database)s
olcSuffix: %(suffix)s
olcRootDN: %(rootdn)s
olcRootPW: %(rootpw)s
olcDbDirectory: %(directory)s
olcDbIndex: objectclass,entryCSN,entryUUID eq
dn: olcOverlay=syncprov,olcDatabase={1}%(database)s,cn=config
objectClass: olcOverlayConfig
objectClass: olcSyncProvConfig
olcOverlay: syncprov
olcSpCheckpoint: 100 10
olcSpSessionlog: 100
"""
OTHER_CONF = r"""
"""
# Define initial data load, both as an LDIF and as a dictionary.
LDIF_TEMPLATE = """dn: %(suffix)s
objectClass: dcObject
objectClass: organization
dc: %(dc)s
o: %(dc)s
dn: %(rootdn)s
objectClass: applicationProcess
objectClass: simpleSecurityObject
cn: %(rootcn)s
userPassword: %(rootpw)s
dn: cn=Foo1,%(suffix)s
objectClass: organizationalRole
cn: Foo1
dn: cn=Foo2,%(suffix)s
objectClass: organizationalRole
cn: Foo2
dn: cn=Foo3,%(suffix)s
objectClass: organizationalRole
cn: Foo3
dn: ou=Container,%(suffix)s
objectClass: organizationalUnit
ou: Container
dn: cn=Foo4,ou=Container,%(suffix)s
objectClass: organizationalRole
cn: Foo4
"""
# NOTE: For the dict, it needs to be kept up-to-date as we make changes!
LDAP_ENTRIES = {
'ou=Container,dc=slapd-test,dc=python-ldap,dc=org': {
'objectClass': [b'organizationalUnit'],
'ou': [b'Container']
},
'cn=Foo2,dc=slapd-test,dc=python-ldap,dc=org': {
'objectClass': [b'organizationalRole'],
'cn': [b'Foo2']
},
'cn=Foo4,ou=Container,dc=slapd-test,dc=python-ldap,dc=org': {
'objectClass': [b'organizationalRole'],
'cn': [b'Foo4']
},
'cn=Manager,dc=slapd-test,dc=python-ldap,dc=org': {
'objectClass': [b'applicationProcess', b'simpleSecurityObject'],
'userPassword': [b'password'],
'cn': [b'Manager']
},
'cn=Foo3,dc=slapd-test,dc=python-ldap,dc=org': {
'objectClass': [b'organizationalRole'],
'cn': [b'Foo3']
},
'cn=Foo1,dc=slapd-test,dc=python-ldap,dc=org': {
'objectClass': [b'organizationalRole'],
'cn': [b'Foo1']
},
'dc=slapd-test,dc=python-ldap,dc=org': {
'objectClass': [b'dcObject', b'organization'],
'dc': [b'slapd-test'],
'o': [b'slapd-test']
}
}
class SyncreplProvider(SlapdObject):
slapd_conf_template = SLAPD_CONF_PROVIDER_TEMPLATE
class SyncreplClient(SimpleLDAPObject, SyncreplConsumer):
"""
This is a very simple class to start up the syncrepl search
and handle callbacks that come in.
Needs to be separate, because once an LDAP client starts a syncrepl
search, it can't be used for anything else.
"""
def __init__(self, uri, dn, password, storage=None, **kwargs):
"""
Set up our object by creating a search client, connecting, and binding.
"""
if storage is not None:
self.data = shelve.open(storage)
self.uuid_dn = shelve.open(storage + 'uuid_dn')
self.dn_attrs = shelve.open(storage + 'dn_attrs')
self.using_shelve = True
else:
self.data = {}
self.uuid_dn = {}
self.dn_attrs = {}
self.using_shelve = False
self.data['cookie'] = None
self.present = []
self.refresh_done = False
SimpleLDAPObject.__init__(self, uri, **kwargs)
self.simple_bind_s(dn, password)
def unbind_s(self):
"""
In addition to unbinding from LDAP, we need to close the shelf.
"""
if self.using_shelve is True:
self.data.close()
self.uuid_dn.close()
self.dn_attrs.close()
SimpleLDAPObject.unbind_s(self)
def search(self, search_base, search_mode):
"""
Start a syncrepl search operation, given a base DN and search mode.
"""
self.search_id = self.syncrepl_search(
search_base,
ldap.SCOPE_SUBTREE,
mode=search_mode,
)
def cancel(self):
"""
A simple wrapper to call parent class with syncrepl search ID.
"""
SimpleLDAPObject.cancel(self, self.search_id)
def poll(self, timeout=None, all=0):
"""
Take the params, add the syncrepl search ID, and call the proper poll.
"""
return self.syncrepl_poll(
self.search_id,
timeout=timeout,
all=all
)
def syncrepl_get_cookie(self):
"""
Pull cookie from storage, if one exists.
"""
return self.data['cookie']
def syncrepl_set_cookie(self, cookie):
"""
Update stored cookie.
"""
self.data['cookie'] = cookie
def syncrepl_refreshdone(self):
"""
Just update a variable.
"""
self.refresh_done = True
def syncrepl_delete(self, uuids):
"""
Delete the given items from both maps.
"""
for uuid in uuids:
del self.dn_attrs[self.uuid_dn[uuid]]
del self.uuid_dn[uuid]
def syncrepl_entry(self, dn, attrs, uuid):
"""
Handles adds and changes (including DN changes).
"""
if uuid in self.uuid_dn:
# Catch changing DNs.
if dn != self.uuid_dn[uuid]:
# Delete data associated with old DN.
del self.dn_attrs[self.uuid_dn[uuid]]
# Update both maps.
self.uuid_dn[uuid] = dn
self.dn_attrs[dn] = attrs
def syncrepl_present(self, uuids, refreshDeletes=False):
"""
The 'present' message from the LDAP server is the most complicated
part of the refresh phase. Suggest looking here for more info:
https://syncrepl-client.readthedocs.io/en/latest/client.html
"""
if (uuids is not None) and (refreshDeletes is False):
self.present.extend(uuids)
elif (uuids is None) and (refreshDeletes is False):
deleted_uuids = []
for uuid in self.uuid_dn:
if uuid not in self.present:
deleted_uuids.append(uuid)
if len(deleted_uuids) > 0:
self.syncrepl_delete(deleted_uuids)
elif (uuids is not None) and (refreshDeletes is True):
self.syncrepl_delete(uuids)
elif (uuids is None) and (refreshDeletes is True):
pass
class BaseSyncreplTests:
"""
This is a test of all the basic Syncrepl operations. It covers starting a
search (both types of search), doing the refresh part of the search,
and checking that we got everything that we expected. We also test that
timeouts and cancellation are working properly.
"""
server_class = SyncreplProvider
ldap_object_class = SimpleLDAPObject
@classmethod
def setUpClass(cls):
super().setUpClass()
# insert some Foo* objects via ldapadd
cls.server.ldapadd(
LDIF_TEMPLATE % {
'suffix':cls.server.suffix,
'rootdn':cls.server.root_dn,
'rootcn':cls.server.root_cn,
'rootpw':cls.server.root_pw,
'dc': cls.server.suffix.split(',')[0][3:],
}
)
def setUp(self):
super().setUp()
self.tester = None
self.suffix = None
def tearDown(self):
self.tester.unbind_s()
super().tearDown()
def create_client(self):
raise NotImplementedError
def test_refreshOnly_search(self):
'''
Test to see if we can initialize a syncrepl search.
'''
self.tester.search(
self.suffix,
'refreshOnly'
)
def test_refreshAndPersist_search(self):
self.tester.search(
self.suffix,
'refreshAndPersist'
)
def test_refreshOnly_poll_full(self):
"""
Test doing a full refresh cycle, and check what we got.
"""
self.tester.search(
self.suffix,
'refreshOnly'
)
poll_result = self.tester.poll(
all=1,
timeout=None
)
self.assertFalse(poll_result)
self.assertEqual(self.tester.dn_attrs, LDAP_ENTRIES)
def test_refreshAndPersist_poll_only(self):
"""
Test the refresh part of refresh-and-persist, and check what we got.
"""
self.tester.search(
self.suffix,
'refreshAndPersist'
)
# Make sure to stop the test before going into persist mode.
while self.tester.refresh_done is not True:
poll_result = self.tester.poll(
all=0,
timeout=None
)
self.assertTrue(poll_result)
self.assertEqual(self.tester.dn_attrs, LDAP_ENTRIES)
def test_refreshAndPersist_timeout(self):
"""
Make sure refreshAndPersist can handle a search with timeouts.
"""
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)
# Run a search with timeout.
# Nothing is changing the server, so it shoud timeout.
self.assertRaises(
ldap.TIMEOUT,
self.tester.poll,
all=0,
timeout=1
)
def test_refreshAndPersist_cancelled(self):
"""
Make sure refreshAndPersist can handle cancelling a syncrepl search.
"""
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)
# Request cancellation.
self.tester.cancel()
# Run another poll, without timeout, but which should cancel out.
self.assertRaises(
ldap.CANCELLED,
self.tester.poll,
all=1,
timeout=None
)
# Server data should still be intact.
self.assertEqual(self.tester.dn_attrs, LDAP_ENTRIES)
# TODO:
# * Make a new client, with a data store, and close. Then, load a new
# client with the same datastore, and see if the data store loads OK.
# * Make a new client, with a data store, and close. Then, load a new
# client with the same datastore. Delete an entry, and the cookie.
# Start the sync, and everything should sync up OK.
# * Load the refreshOnly client, using existing data. Make a change
# on the server, and the client should pick it up in the refresh phase.
# * Load the refreshAndPersist client, using existing data. Make a change
# on the server, and the client should pick it up in the refresh phase.
# * Load the refreshAndPersist client, using existing data. Let the
# refresh phase complete. Make a change on the server, and the client
# should pick it up during the persist phase.
class TestSyncrepl(BaseSyncreplTests, SlapdTestCase):
def setUp(self):
super().setUp()
self.tester = SyncreplClient(
self.server.ldap_uri,
self.server.root_dn,
self.server.root_pw,
bytes_mode=False
)
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.skipTest("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
syncrepl messages are handled correctly.
"""
def test_syncidset_message(self):
"""
A syncrepl server may send a sync info message, with a syncIdSet
of uuids to delete. A regression was found in the original
sync info message implementation due to how the choice was
evaluated, because refreshPresent and refreshDelete were both
able to be fully expressed as defaults, causing the parser
to mistakenly catch a syncIdSet as a refreshPresent/refereshDelete.
This tests that a syncIdSet request is properly decoded.
reference: https://tools.ietf.org/html/rfc4533#section-2.5
"""
# This is a dump of a syncidset message from wireshark + 389-ds
msg = """
a36b04526c6461706b64632e6578616d706c652e636f6d3a333839303123636e
3d6469726563746f7279206d616e616765723a64633d6578616d706c652c6463
3d636f6d3a286f626a656374436c6173733d2a2923330101ff311204108dc446
01a93611ea8aaff248c5fa5780
""".replace(' ', '').replace('\n', '')
msgraw = binascii.unhexlify(msg)
sim = SyncInfoMessage(msgraw)
self.assertEqual(sim.refreshDelete, None)
self.assertEqual(sim.refreshPresent, None)
self.assertEqual(sim.newcookie, None)
self.assertEqual(sim.syncIdSet,
{
'cookie': 'ldapkdc.example.com:38901#cn=directory manager:dc=example,dc=com:(objectClass=*)#3',
'syncUUIDs': ['8dc44601-a936-11ea-8aaf-f248c5fa5780'],
'refreshDeletes': True
}
)
if __name__ == '__main__':
unittest.main()