4586810d9643606e9e58fd3277aae11cf1f2021b
[samba.git] / source4 / dsdb / samdb / ldb_modules / partition.c
1
2 /* 
3    Partitions ldb module
4
5    Copyright (C) Andrew Bartlett <abartlet@samba.org> 2006
6    Copyright (C) Stefan Metzmacher <metze@samba.org> 2007
7
8    * NOTICE: this module is NOT released under the GNU LGPL license as
9    * other ldb code. This module is release under the GNU GPL v2 or
10    * later license.
11
12    This program is free software; you can redistribute it and/or modify
13    it under the terms of the GNU General Public License as published by
14    the Free Software Foundation; either version 3 of the License, or
15    (at your option) any later version.
16    
17    This program is distributed in the hope that it will be useful,
18    but WITHOUT ANY WARRANTY; without even the implied warranty of
19    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
20    GNU General Public License for more details.
21    
22    You should have received a copy of the GNU General Public License
23    along with this program.  If not, see <http://www.gnu.org/licenses/>.
24 */
25
26 /*
27  *  Name: ldb
28  *
29  *  Component: ldb partitions module
30  *
31  *  Description: Implement LDAP partitions
32  *
33  *  Author: Andrew Bartlett
34  *  Author: Stefan Metzmacher
35  */
36
37 #include "includes.h"
38 #include "ldb/include/ldb_includes.h"
39 #include "dsdb/samdb/samdb.h"
40
41 struct partition_private_data {
42         struct dsdb_control_current_partition **partitions;
43         struct ldb_dn **replicate;
44 };
45
46 struct partition_context {
47         struct ldb_module *module;
48         struct ldb_request *orig_req;
49
50         struct ldb_request **down_req;
51         int num_requests;
52         int finished_requests;
53 };
54
55 static struct partition_context *partition_init_handle(struct ldb_request *req, struct ldb_module *module)
56 {
57         struct partition_context *ac;
58         struct ldb_handle *h;
59
60         h = talloc_zero(req, struct ldb_handle);
61         if (h == NULL) {
62                 ldb_set_errstring(module->ldb, "Out of Memory");
63                 return NULL;
64         }
65
66         h->module = module;
67
68         ac = talloc_zero(h, struct partition_context);
69         if (ac == NULL) {
70                 ldb_set_errstring(module->ldb, "Out of Memory");
71                 talloc_free(h);
72                 return NULL;
73         }
74
75         h->private_data = ac;
76
77         ac->module = module;
78         ac->orig_req = req;
79
80         req->handle = h;
81
82         return ac;
83 }
84
85 static struct ldb_module *make_module_for_next_request(TALLOC_CTX *mem_ctx, 
86                                                        struct ldb_context *ldb,
87                                                        struct ldb_module *module)
88 {
89         struct ldb_module *current;
90         static const struct ldb_module_ops ops; /* zero */
91         current = talloc_zero(mem_ctx, struct ldb_module);
92         if (current == NULL) {
93                 return module;
94         }
95         
96         current->ldb = ldb;
97         current->ops = &ops;
98         current->prev = NULL;
99         current->next = module;
100         return current;
101 }
102
103 static struct dsdb_control_current_partition *find_partition(struct partition_private_data *data,
104                                                              struct ldb_dn *dn)
105 {
106         int i;
107
108         /* Look at base DN */
109         /* Figure out which partition it is under */
110         /* Skip the lot if 'data' isn't here yet (initialistion) */
111         for (i=0; data && data->partitions && data->partitions[i]; i++) {
112                 if (ldb_dn_compare_base(data->partitions[i]->dn, dn) == 0) {
113                         return data->partitions[i];
114                 }
115         }
116
117         return NULL;
118 };
119
120 /**
121  * fire the caller's callback for every entry, but only send 'done' once.
122  */
123 static int partition_search_callback(struct ldb_context *ldb, void *context, struct ldb_reply *ares)
124 {
125         struct partition_context *ac;
126
127         ac = talloc_get_type(context, struct partition_context);
128
129         if (ares->type == LDB_REPLY_ENTRY) {
130                 return ac->orig_req->callback(ldb, ac->orig_req->context, ares);
131         } else {
132                 ac->finished_requests++;
133                 if (ac->finished_requests == ac->num_requests) {
134                         return ac->orig_req->callback(ldb, ac->orig_req->context, ares);
135                 } else {
136                         talloc_free(ares);
137                         return LDB_SUCCESS;
138                 }
139         }
140 }
141
142 /**
143  * only fire the 'last' callback, and only for START-TLS for now 
144  */
145 static int partition_other_callback(struct ldb_context *ldb, void *context, struct ldb_reply *ares)
146 {
147         struct partition_context *ac;
148
149         ac = talloc_get_type(context, struct partition_context);
150
151         if (!ac->orig_req->callback) {
152                 talloc_free(ares);
153                 return LDB_SUCCESS;
154         }
155
156         if (!ares 
157             || (ares->type == LDB_REPLY_EXTENDED 
158                 && strcmp(ares->response->oid, LDB_EXTENDED_START_TLS_OID))) {
159                 ac->finished_requests++;
160                 if (ac->finished_requests == ac->num_requests) {
161                         return ac->orig_req->callback(ldb, ac->orig_req->context, ares);
162                 }
163                 talloc_free(ares);
164                 return LDB_SUCCESS;
165         }
166         ldb_set_errstring(ldb, "partition_other_callback: Unknown reply type, only supports START_TLS");
167         talloc_free(ares);
168         return LDB_ERR_OPERATIONS_ERROR;
169 }
170
171
172 static int partition_send_request(struct partition_context *ac, struct ldb_control *remove_control, 
173                                   struct dsdb_control_current_partition *partition)
174 {
175         int ret;
176         struct ldb_module *backend;
177         struct ldb_request *req;
178         struct ldb_control **saved_controls;
179
180         if (partition) {
181                 backend = make_module_for_next_request(ac, ac->module->ldb, partition->module);
182         } else {
183                 backend = ac->module;
184         }
185
186         ac->down_req = talloc_realloc(ac, ac->down_req, 
187                                         struct ldb_request *, ac->num_requests + 1);
188         if (!ac->down_req) {
189                 ldb_oom(ac->module->ldb);
190                 return LDB_ERR_OPERATIONS_ERROR;
191         }
192         req = ac->down_req[ac->num_requests] = talloc(ac, struct ldb_request);
193         if (req == NULL) {
194                 ldb_oom(ac->module->ldb);
195                 return LDB_ERR_OPERATIONS_ERROR;
196         }
197         
198         *req = *ac->orig_req; /* copy the request */
199
200         if (req->controls) {
201                 req->controls
202                         = talloc_memdup(req,
203                                         ac->orig_req->controls, talloc_get_size(ac->orig_req->controls));
204                 if (req->controls == NULL) {
205                         ldb_oom(ac->module->ldb);
206                         return LDB_ERR_OPERATIONS_ERROR;
207                 }
208         }
209
210         if (req->operation == LDB_SEARCH) {
211                 /* If the search is for 'more' than this partition,
212                  * then change the basedn, so a remote LDAP server
213                  * doesn't object */
214                 if (partition) {
215                         if (ldb_dn_compare_base(partition->dn, req->op.search.base) != 0) {
216                                 req->op.search.base = partition->dn;
217                         }
218                 } else {
219                         req->op.search.base = NULL;
220                 }
221                 req->callback = partition_search_callback;
222                 req->context = ac;
223         } else {
224                 req->callback = partition_other_callback;
225                 req->context = ac;
226         }
227
228         /* Remove a control, so we don't confuse a backend server */
229         if (remove_control && !save_controls(remove_control, req, &saved_controls)) {
230                 ldb_oom(ac->module->ldb);
231                 return LDB_ERR_OPERATIONS_ERROR;
232         }
233         
234         if (partition) {
235                 ret = ldb_request_add_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID, false, partition);
236                 if (ret != LDB_SUCCESS) {
237                         return ret;
238                 }
239         }
240
241         /* Spray off search requests the backend */
242         ret = ldb_next_request(backend, req);
243         if (ret != LDB_SUCCESS) {
244                 return ret;
245         }
246
247         ac->num_requests++;
248         return LDB_SUCCESS;
249 }
250
251 /**
252  * Send a request down to all the partitions
253  */
254 static int partition_send_all(struct ldb_module *module, 
255                               struct partition_context *ac, 
256                               struct ldb_control *remove_control, 
257                               struct ldb_request *req) 
258 {
259         int i;
260         struct partition_private_data *data = talloc_get_type(module->private_data, 
261                                                               struct partition_private_data);
262         int ret = partition_send_request(ac, remove_control, NULL);
263         if (ret != LDB_SUCCESS) {
264                 return ret;
265         }
266         for (i=0; data && data->partitions && data->partitions[i]; i++) {
267                 ret = partition_send_request(ac, remove_control, data->partitions[i]);
268                 if (ret != LDB_SUCCESS) {
269                         return ret;
270                 }
271         }
272         return LDB_SUCCESS;
273 }
274
275 /**
276  * Figure out which backend a request needs to be aimed at.  Some
277  * requests must be replicated to all backends
278  */
279 static int partition_replicate(struct ldb_module *module, struct ldb_request *req, struct ldb_dn *dn) 
280 {
281         unsigned i;
282         int ret;
283         struct dsdb_control_current_partition *partition;
284         struct ldb_module *backend;
285         struct partition_private_data *data = talloc_get_type(module->private_data, 
286                                                               struct partition_private_data);
287         
288         if (req->operation != LDB_SEARCH) {
289                 /* Is this a special DN, we need to replicate to every backend? */
290                 for (i=0; data->replicate && data->replicate[i]; i++) {
291                         if (ldb_dn_compare(data->replicate[i], 
292                                            dn) == 0) {
293                                 struct partition_context *ac;
294                                 
295                                 ac = partition_init_handle(req, module);
296                                 if (!ac) {
297                                         return LDB_ERR_OPERATIONS_ERROR;
298                                 }
299                                 
300                                 return partition_send_all(module, ac, NULL, req);
301                         }
302                 }
303         }
304
305         /* Otherwise, we need to find the partition to fire it to */
306
307         /* Find partition */
308         partition = find_partition(data, dn);
309         if (!partition) {
310                 /*
311                  * if we haven't found a matching partition
312                  * pass the request to the main ldb
313                  *
314                  * TODO: we should maybe return an error here
315                  *       if it's not a special dn
316                  */
317                 return ldb_next_request(module, req);
318         }
319
320         backend = make_module_for_next_request(req, module->ldb, partition->module);
321         if (!backend) {
322                 return LDB_ERR_OPERATIONS_ERROR;
323         }
324
325         ret = ldb_request_add_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID, false, partition);
326         if (ret != LDB_SUCCESS) {
327                 return ret;
328         }
329
330         /* issue request */
331         return ldb_next_request(backend, req);
332 }
333
334 /* search */
335 static int partition_search(struct ldb_module *module, struct ldb_request *req)
336 {
337         /* Find backend */
338         struct partition_private_data *data = talloc_get_type(module->private_data, 
339                                                               struct partition_private_data);
340         /* issue request */
341
342         /* (later) consider if we should be searching multiple
343          * partitions (for 'invisible' partition behaviour */
344         struct ldb_control *search_control = ldb_request_get_control(req, LDB_CONTROL_SEARCH_OPTIONS_OID);
345         
346         struct ldb_search_options_control *search_options = NULL;
347         if (search_control) {
348                 search_options = talloc_get_type(search_control->data, struct ldb_search_options_control);
349         }
350
351         if (search_options && (search_options->search_options & LDB_SEARCH_OPTION_PHANTOM_ROOT)) {
352                 int ret, i;
353                 struct partition_context *ac;
354                 struct ldb_control *remove_control = NULL;
355                 if ((search_options->search_options & ~LDB_SEARCH_OPTION_PHANTOM_ROOT) == 0) {
356                         /* We have processed this flag, so we are done with this control now */
357                         remove_control = search_control;
358                 }
359                 ac = partition_init_handle(req, module);
360                 if (!ac) {
361                         return LDB_ERR_OPERATIONS_ERROR;
362                 }
363
364                 /* Search from the base DN */
365                 if (!req->op.search.base || ldb_dn_is_null(req->op.search.base)) {
366                         return partition_send_all(module, ac, remove_control, req);
367                 }
368                 for (i=0; data && data->partitions && data->partitions[i]; i++) {
369                         /* Find all partitions under the search base */
370                         if (ldb_dn_compare_base(req->op.search.base, data->partitions[i]->dn) == 0) {
371                                 ret = partition_send_request(ac, remove_control, data->partitions[i]);
372                                 if (ret != LDB_SUCCESS) {
373                                         return ret;
374                                 }
375                         }
376                 }
377
378                 /* Perhaps we didn't match any partitions.  Try the main partition, only */
379                 if (ac->num_requests == 0) {
380                         talloc_free(ac);
381                         return ldb_next_request(module, req);
382                 }
383                 
384                 return LDB_SUCCESS;
385         } else {
386                 /* Handle this like all other requests */
387                 return partition_replicate(module, req, req->op.search.base);
388         }
389 }
390
391 /* add */
392 static int partition_add(struct ldb_module *module, struct ldb_request *req)
393 {
394         return partition_replicate(module, req, req->op.add.message->dn);
395 }
396
397 /* modify */
398 static int partition_modify(struct ldb_module *module, struct ldb_request *req)
399 {
400         return partition_replicate(module, req, req->op.mod.message->dn);
401 }
402
403 /* delete */
404 static int partition_delete(struct ldb_module *module, struct ldb_request *req)
405 {
406         return partition_replicate(module, req, req->op.del.dn);
407 }
408
409 /* rename */
410 static int partition_rename(struct ldb_module *module, struct ldb_request *req)
411 {
412         int i, matched = -1;
413         /* Find backend */
414         struct dsdb_control_current_partition *backend, *backend2;
415         
416         struct partition_private_data *data = talloc_get_type(module->private_data, 
417                                                               struct partition_private_data);
418
419         /* Skip the lot if 'data' isn't here yet (initialistion) */
420         if (!data) {
421                 return LDB_ERR_OPERATIONS_ERROR;
422         }
423
424         backend = find_partition(data, req->op.rename.olddn);
425         backend2 = find_partition(data, req->op.rename.newdn);
426
427         if ((backend && !backend2) || (!backend && backend2)) {
428                 return LDB_ERR_AFFECTS_MULTIPLE_DSAS;
429         }
430
431         if (backend != backend2) {
432                 ldb_asprintf_errstring(module->ldb, 
433                                        "Cannot rename from %s in %s to %s in %s: %s",
434                                        ldb_dn_get_linearized(req->op.rename.olddn),
435                                        ldb_dn_get_linearized(backend->dn),
436                                        ldb_dn_get_linearized(req->op.rename.newdn),
437                                        ldb_dn_get_linearized(backend2->dn),
438                                        ldb_strerror(LDB_ERR_AFFECTS_MULTIPLE_DSAS));
439                 return LDB_ERR_AFFECTS_MULTIPLE_DSAS;
440         }
441
442         for (i=0; data && data->partitions && data->partitions[i]; i++) {
443                 if (ldb_dn_compare_base(req->op.rename.olddn, data->partitions[i]->dn) == 0) {
444                         matched = i;
445                 }
446         }
447
448         if (matched > 0) {
449                 ldb_asprintf_errstring(module->ldb, 
450                                        "Cannot rename from %s to %s, subtree rename would cross partition %s: %s",
451                                        ldb_dn_get_linearized(req->op.rename.olddn),
452                                        ldb_dn_get_linearized(req->op.rename.newdn),
453                                        ldb_dn_get_linearized(data->partitions[matched]->dn),
454                                        ldb_strerror(LDB_ERR_AFFECTS_MULTIPLE_DSAS));
455                 return LDB_ERR_AFFECTS_MULTIPLE_DSAS;
456         }
457
458         return partition_replicate(module, req, req->op.rename.olddn);
459 }
460
461 /* start a transaction */
462 static int partition_start_trans(struct ldb_module *module)
463 {
464         int i, ret;
465         struct partition_private_data *data = talloc_get_type(module->private_data, 
466                                                               struct partition_private_data);
467         /* Look at base DN */
468         /* Figure out which partition it is under */
469         /* Skip the lot if 'data' isn't here yet (initialistion) */
470         ret = ldb_next_start_trans(module);
471         if (ret != LDB_SUCCESS) {
472                 return ret;
473         }
474
475         for (i=0; data && data->partitions && data->partitions[i]; i++) {
476                 struct ldb_module *next = make_module_for_next_request(module, module->ldb, data->partitions[i]->module);
477
478                 ret = ldb_next_start_trans(next);
479                 talloc_free(next);
480                 if (ret != LDB_SUCCESS) {
481                         /* Back it out, if it fails on one */
482                         for (i--; i >= 0; i--) {
483                                 next = make_module_for_next_request(module, module->ldb, data->partitions[i]->module);
484                                 ldb_next_del_trans(next);
485                                 talloc_free(next);
486                         }
487                         return ret;
488                 }
489         }
490         return LDB_SUCCESS;
491 }
492
493 /* end a transaction */
494 static int partition_end_trans(struct ldb_module *module)
495 {
496         int i, ret, ret2 = LDB_SUCCESS;
497         struct partition_private_data *data = talloc_get_type(module->private_data, 
498                                                               struct partition_private_data);
499         ret = ldb_next_end_trans(module);
500         if (ret != LDB_SUCCESS) {
501                 return ret;
502         }
503
504         /* Look at base DN */
505         /* Figure out which partition it is under */
506         /* Skip the lot if 'data' isn't here yet (initialistion) */
507         for (i=0; data && data->partitions && data->partitions[i]; i++) {
508                 struct ldb_module *next = make_module_for_next_request(module, module->ldb, data->partitions[i]->module);
509                 
510                 ret = ldb_next_end_trans(next);
511                 talloc_free(next);
512                 if (ret != LDB_SUCCESS) {
513                         ret2 = ret;
514                 }
515         }
516
517         if (ret != LDB_SUCCESS) {
518                 /* Back it out, if it fails on one */
519                 for (i=0; data && data->partitions && data->partitions[i]; i++) {
520                         struct ldb_module *next = make_module_for_next_request(module, module->ldb, data->partitions[i]->module);
521                         ldb_next_del_trans(next);
522                         talloc_free(next);
523                 }
524         }
525         return ret;
526 }
527
528 /* delete a transaction */
529 static int partition_del_trans(struct ldb_module *module)
530 {
531         int i, ret, ret2 = LDB_SUCCESS;
532         struct partition_private_data *data = talloc_get_type(module->private_data, 
533                                                               struct partition_private_data);
534         ret = ldb_next_del_trans(module);
535         if (ret != LDB_SUCCESS) {
536                 ret2 = ret;
537         }
538
539         /* Look at base DN */
540         /* Figure out which partition it is under */
541         /* Skip the lot if 'data' isn't here yet (initialistion) */
542         for (i=0; data && data->partitions && data->partitions[i]; i++) {
543                 struct ldb_module *next = make_module_for_next_request(module, module->ldb, data->partitions[i]->module);
544                 
545                 ret = ldb_next_del_trans(next);
546                 talloc_free(next);
547                 if (ret != LDB_SUCCESS) {
548                         ret2 = ret;
549                 }
550         }
551         return ret2;
552 }
553
554 static int partition_sequence_number(struct ldb_module *module, struct ldb_request *req)
555 {
556         int i, ret;
557         uint64_t seq_number = 0;
558         uint64_t timestamp_sequence = 0;
559         uint64_t timestamp = 0;
560         struct partition_private_data *data = talloc_get_type(module->private_data, 
561                                                               struct partition_private_data);
562
563         switch (req->op.seq_num.type) {
564         case LDB_SEQ_NEXT:
565         case LDB_SEQ_HIGHEST_SEQ:
566                 ret = ldb_next_request(module, req);
567                 if (ret != LDB_SUCCESS) {
568                         return ret;
569                 }
570                 if (req->op.seq_num.flags & LDB_SEQ_TIMESTAMP_SEQUENCE) {
571                         timestamp_sequence = req->op.seq_num.seq_num;
572                 } else {
573                         seq_number = seq_number + req->op.seq_num.seq_num;
574                 }
575
576                 /* Look at base DN */
577                 /* Figure out which partition it is under */
578                 /* Skip the lot if 'data' isn't here yet (initialistion) */
579                 for (i=0; data && data->partitions && data->partitions[i]; i++) {
580                         struct ldb_module *next = make_module_for_next_request(req, module->ldb, data->partitions[i]->module);
581                         
582                         ret = ldb_next_request(next, req);
583                         talloc_free(next);
584                         if (ret != LDB_SUCCESS) {
585                                 return ret;
586                         }
587                         if (req->op.seq_num.flags & LDB_SEQ_TIMESTAMP_SEQUENCE) {
588                                 timestamp_sequence = MAX(timestamp_sequence, req->op.seq_num.seq_num);
589                         } else {
590                                 seq_number = seq_number + req->op.seq_num.seq_num;
591                         }
592                 }
593                 /* fall though */
594         case LDB_SEQ_HIGHEST_TIMESTAMP:
595         {
596                 struct ldb_request *date_req = talloc(req, struct ldb_request);
597                 if (!date_req) {
598                         return LDB_ERR_OPERATIONS_ERROR;
599                 }
600                 *date_req = *req;
601                 date_req->op.seq_num.flags = LDB_SEQ_HIGHEST_TIMESTAMP;
602
603                 ret = ldb_next_request(module, date_req);
604                 if (ret != LDB_SUCCESS) {
605                         return ret;
606                 }
607                 timestamp = date_req->op.seq_num.seq_num;
608                 
609                 /* Look at base DN */
610                 /* Figure out which partition it is under */
611                 /* Skip the lot if 'data' isn't here yet (initialistion) */
612                 for (i=0; data && data->partitions && data->partitions[i]; i++) {
613                         struct ldb_module *next = make_module_for_next_request(req, module->ldb, data->partitions[i]->module);
614                         
615                         ret = ldb_next_request(next, date_req);
616                         talloc_free(next);
617                         if (ret != LDB_SUCCESS) {
618                                 return ret;
619                         }
620                         timestamp = MAX(timestamp, date_req->op.seq_num.seq_num);
621                 }
622                 break;
623         }
624         }
625
626         switch (req->op.seq_num.flags) {
627         case LDB_SEQ_NEXT:
628         case LDB_SEQ_HIGHEST_SEQ:
629                 
630                 req->op.seq_num.flags = 0;
631
632                 /* Has someone above set a timebase sequence? */
633                 if (timestamp_sequence) {
634                         req->op.seq_num.seq_num = (((unsigned long long)timestamp << 24) | (seq_number & 0xFFFFFF));
635                 } else {
636                         req->op.seq_num.seq_num = seq_number;
637                 }
638
639                 if (timestamp_sequence > req->op.seq_num.seq_num) {
640                         req->op.seq_num.seq_num = timestamp_sequence;
641                         req->op.seq_num.flags |= LDB_SEQ_TIMESTAMP_SEQUENCE;
642                 }
643
644                 req->op.seq_num.flags |= LDB_SEQ_GLOBAL_SEQUENCE;
645                 break;
646         case LDB_SEQ_HIGHEST_TIMESTAMP:
647                 req->op.seq_num.seq_num = timestamp;
648                 break;
649         }
650
651         switch (req->op.seq_num.flags) {
652         case LDB_SEQ_NEXT:
653                 req->op.seq_num.seq_num++;
654         }
655         return LDB_SUCCESS;
656 }
657
658 static int partition_extended_replicated_objects(struct ldb_module *module, struct ldb_request *req)
659 {
660         struct dsdb_extended_replicated_objects *ext;
661
662         ext = talloc_get_type(req->op.extended.data, struct dsdb_extended_replicated_objects);
663         if (!ext) {
664                 ldb_debug(module->ldb, LDB_DEBUG_FATAL, "partition_extended_replicated_objects: invalid extended data\n");
665                 return LDB_ERR_PROTOCOL_ERROR;
666         }
667
668         if (ext->version != DSDB_EXTENDED_REPLICATED_OBJECTS_VERSION) {
669                 ldb_debug(module->ldb, LDB_DEBUG_FATAL, "partition_extended_replicated_objects: extended data invalid version [%u != %u]\n",
670                           ext->version, DSDB_EXTENDED_REPLICATED_OBJECTS_VERSION);
671                 return LDB_ERR_PROTOCOL_ERROR;
672         }
673
674         return partition_replicate(module, req, ext->partition_dn);
675 }
676
677 /* extended */
678 static int partition_extended(struct ldb_module *module, struct ldb_request *req)
679 {
680         struct partition_context *ac;
681
682         if (strcmp(req->op.extended.oid, DSDB_EXTENDED_REPLICATED_OBJECTS_OID) == 0) {
683                 return partition_extended_replicated_objects(module, req);
684         }
685
686         /* 
687          * as the extended operation has no dn
688          * we need to send it to all partitions
689          */
690
691         ac = partition_init_handle(req, module);
692         if (!ac) {
693                 return LDB_ERR_OPERATIONS_ERROR;
694         }
695                         
696         return partition_send_all(module, ac, NULL, req);
697 }
698
699 static int sort_compare(void *void1,
700                         void *void2, void *opaque)
701 {
702         struct dsdb_control_current_partition **pp1 = 
703                 (struct dsdb_control_current_partition **)void1;
704         struct dsdb_control_current_partition **pp2 = 
705                 (struct dsdb_control_current_partition **)void2;
706         struct dsdb_control_current_partition *partition1 = talloc_get_type(*pp1,
707                                                             struct dsdb_control_current_partition);
708         struct dsdb_control_current_partition *partition2 = talloc_get_type(*pp2,
709                                                             struct dsdb_control_current_partition);
710
711         return ldb_dn_compare(partition1->dn, partition2->dn);
712 }
713
714 static int partition_init(struct ldb_module *module)
715 {
716         int ret, i;
717         TALLOC_CTX *mem_ctx = talloc_new(module);
718         const char *attrs[] = { "partition", "replicateEntries", "modules", NULL };
719         struct ldb_result *res;
720         struct ldb_message *msg;
721         struct ldb_message_element *partition_attributes;
722         struct ldb_message_element *replicate_attributes;
723         struct ldb_message_element *modules_attributes;
724
725         struct partition_private_data *data;
726
727         if (!mem_ctx) {
728                 return LDB_ERR_OPERATIONS_ERROR;
729         }
730
731         data = talloc(mem_ctx, struct partition_private_data);
732         if (data == NULL) {
733                 return LDB_ERR_OPERATIONS_ERROR;
734         }
735
736         ret = ldb_search(module->ldb, ldb_dn_new(mem_ctx, module->ldb, "@PARTITION"),
737                          LDB_SCOPE_BASE,
738                          NULL, attrs,
739                          &res);
740         if (ret != LDB_SUCCESS) {
741                 talloc_free(mem_ctx);
742                 return ret;
743         }
744         talloc_steal(mem_ctx, res);
745         if (res->count == 0) {
746                 talloc_free(mem_ctx);
747                 return ldb_next_init(module);
748         }
749
750         if (res->count > 1) {
751                 talloc_free(mem_ctx);
752                 return LDB_ERR_CONSTRAINT_VIOLATION;
753         }
754
755         msg = res->msgs[0];
756
757         partition_attributes = ldb_msg_find_element(msg, "partition");
758         if (!partition_attributes) {
759                 ldb_set_errstring(module->ldb, "partition_init: no partitions specified");
760                 talloc_free(mem_ctx);
761                 return LDB_ERR_CONSTRAINT_VIOLATION;
762         }
763         data->partitions = talloc_array(data, struct dsdb_control_current_partition *, partition_attributes->num_values + 1);
764         if (!data->partitions) {
765                 talloc_free(mem_ctx);
766                 return LDB_ERR_OPERATIONS_ERROR;
767         }
768         for (i=0; i < partition_attributes->num_values; i++) {
769                 char *base = talloc_strdup(data->partitions, (char *)partition_attributes->values[i].data);
770                 char *p = strchr(base, ':');
771                 if (!p) {
772                         ldb_asprintf_errstring(module->ldb, 
773                                                 "partition_init: "
774                                                 "invalid form for partition record (missing ':'): %s", base);
775                         talloc_free(mem_ctx);
776                         return LDB_ERR_CONSTRAINT_VIOLATION;
777                 }
778                 p[0] = '\0';
779                 p++;
780                 if (!p[0]) {
781                         ldb_asprintf_errstring(module->ldb, 
782                                                 "partition_init: "
783                                                 "invalid form for partition record (missing backend database): %s", base);
784                         talloc_free(mem_ctx);
785                         return LDB_ERR_CONSTRAINT_VIOLATION;
786                 }
787                 data->partitions[i] = talloc(data->partitions, struct dsdb_control_current_partition);
788                 if (!data->partitions[i]) {
789                         talloc_free(mem_ctx);
790                         return LDB_ERR_OPERATIONS_ERROR;
791                 }
792                 data->partitions[i]->version = DSDB_CONTROL_CURRENT_PARTITION_VERSION;
793
794                 data->partitions[i]->dn = ldb_dn_new(data->partitions[i], module->ldb, base);
795                 if (!data->partitions[i]->dn) {
796                         ldb_asprintf_errstring(module->ldb, 
797                                                 "partition_init: invalid DN in partition record: %s", base);
798                         talloc_free(mem_ctx);
799                         return LDB_ERR_CONSTRAINT_VIOLATION;
800                 }
801
802                 data->partitions[i]->backend = samdb_relative_path(module->ldb, 
803                                                                    data->partitions[i], 
804                                                                    p);
805                 if (!data->partitions[i]->backend) {
806                         ldb_asprintf_errstring(module->ldb, 
807                                                 "partition_init: unable to determine an relative path for partition: %s", base);
808                         talloc_free(mem_ctx);                   
809                 }
810                 ret = ldb_connect_backend(module->ldb, data->partitions[i]->backend, NULL, &data->partitions[i]->module);
811                 if (ret != LDB_SUCCESS) {
812                         talloc_free(mem_ctx);
813                         return ret;
814                 }
815         }
816         data->partitions[i] = NULL;
817
818         /* sort these into order, most to least specific */
819         ldb_qsort(data->partitions, partition_attributes->num_values, sizeof(*data->partitions), 
820                   module->ldb, sort_compare);
821
822         for (i=0; data->partitions[i]; i++) {
823                 struct ldb_request *req;
824                 req = talloc_zero(mem_ctx, struct ldb_request);
825                 if (req == NULL) {
826                         ldb_debug(module->ldb, LDB_DEBUG_ERROR, "partition: Out of memory!\n");
827                         talloc_free(mem_ctx);
828                         return LDB_ERR_OPERATIONS_ERROR;
829                 }
830                 
831                 req->operation = LDB_REQ_REGISTER_PARTITION;
832                 req->op.reg_partition.dn = data->partitions[i]->dn;
833                 
834                 ret = ldb_request(module->ldb, req);
835                 if (ret != LDB_SUCCESS) {
836                         ldb_debug(module->ldb, LDB_DEBUG_ERROR, "partition: Unable to register partition with rootdse!\n");
837                         talloc_free(mem_ctx);
838                         return LDB_ERR_OTHER;
839                 }
840                 talloc_free(req);
841         }
842
843         replicate_attributes = ldb_msg_find_element(msg, "replicateEntries");
844         if (!replicate_attributes) {
845                 data->replicate = NULL;
846         } else {
847                 data->replicate = talloc_array(data, struct ldb_dn *, replicate_attributes->num_values + 1);
848                 if (!data->replicate) {
849                         talloc_free(mem_ctx);
850                         return LDB_ERR_OPERATIONS_ERROR;
851                 }
852                 
853                 for (i=0; i < replicate_attributes->num_values; i++) {
854                         data->replicate[i] = ldb_dn_new(data->replicate, module->ldb, (const char *)replicate_attributes->values[i].data);
855                         if (!ldb_dn_validate(data->replicate[i])) {
856                                 ldb_asprintf_errstring(module->ldb, 
857                                                         "partition_init: "
858                                                         "invalid DN in partition replicate record: %s", 
859                                                         replicate_attributes->values[i].data);
860                                 talloc_free(mem_ctx);
861                                 return LDB_ERR_CONSTRAINT_VIOLATION;
862                         }
863                 }
864                 data->replicate[i] = NULL;
865         }
866
867         /* Make the private data available to any searches the modules may trigger in initialisation */
868         module->private_data = data;
869         talloc_steal(module, data);
870         
871         modules_attributes = ldb_msg_find_element(msg, "modules");
872         if (modules_attributes) {
873                 for (i=0; i < modules_attributes->num_values; i++) {
874                         struct ldb_dn *base_dn;
875                         int partition_idx;
876                         struct dsdb_control_current_partition *partition = NULL;
877                         const char **modules = NULL;
878
879                         char *base = talloc_strdup(data->partitions, (char *)modules_attributes->values[i].data);
880                         char *p = strchr(base, ':');
881                         if (!p) {
882                                 ldb_asprintf_errstring(module->ldb, 
883                                                         "partition_init: "
884                                                         "invalid form for partition module record (missing ':'): %s", base);
885                                 talloc_free(mem_ctx);
886                                 return LDB_ERR_CONSTRAINT_VIOLATION;
887                         }
888                         p[0] = '\0';
889                         p++;
890                         if (!p[0]) {
891                                 ldb_asprintf_errstring(module->ldb, 
892                                                         "partition_init: "
893                                                         "invalid form for partition module record (missing backend database): %s", base);
894                                 talloc_free(mem_ctx);
895                                 return LDB_ERR_CONSTRAINT_VIOLATION;
896                         }
897
898                         modules = ldb_modules_list_from_string(module->ldb, mem_ctx,
899                                                                p);
900                         
901                         base_dn = ldb_dn_new(mem_ctx, module->ldb, base);
902                         if (!ldb_dn_validate(base_dn)) {
903                                 talloc_free(mem_ctx);
904                                 return LDB_ERR_OPERATIONS_ERROR;
905                         }
906                         
907                         for (partition_idx = 0; data->partitions[partition_idx]; partition_idx++) {
908                                 if (ldb_dn_compare(data->partitions[partition_idx]->dn, base_dn) == 0) {
909                                         partition = data->partitions[partition_idx];
910                                         break;
911                                 }
912                         }
913                         
914                         if (!partition) {
915                                 ldb_asprintf_errstring(module->ldb, 
916                                                         "partition_init: "
917                                                         "invalid form for partition module record (no such partition): %s", base);
918                                 talloc_free(mem_ctx);
919                                 return LDB_ERR_CONSTRAINT_VIOLATION;
920                         }
921                         
922                         ret = ldb_load_modules_list(module->ldb, modules, partition->module, &partition->module);
923                         if (ret != LDB_SUCCESS) {
924                                 ldb_asprintf_errstring(module->ldb, 
925                                                        "partition_init: "
926                                                        "loading backend for %s failed: %s", 
927                                                        base, ldb_errstring(module->ldb));
928                                 talloc_free(mem_ctx);
929                                 return ret;
930                         }
931                         ret = ldb_init_module_chain(module->ldb, partition->module);
932                         if (ret != LDB_SUCCESS) {
933                                 ldb_asprintf_errstring(module->ldb, 
934                                                        "partition_init: "
935                                                        "initialising backend for %s failed: %s", 
936                                                        base, ldb_errstring(module->ldb));
937                                 talloc_free(mem_ctx);
938                                 return ret;
939                         }
940                 }
941         }
942
943         talloc_free(mem_ctx);
944         return ldb_next_init(module);
945 }
946
947 static int partition_wait_none(struct ldb_handle *handle) {
948         struct partition_context *ac;
949         int ret;
950         int i;
951     
952         if (!handle || !handle->private_data) {
953                 return LDB_ERR_OPERATIONS_ERROR;
954         }
955
956         if (handle->state == LDB_ASYNC_DONE) {
957                 return handle->status;
958         }
959
960         handle->state = LDB_ASYNC_PENDING;
961         handle->status = LDB_SUCCESS;
962
963         ac = talloc_get_type(handle->private_data, struct partition_context);
964
965         for (i=0; i < ac->num_requests; i++) {
966                 ret = ldb_wait(ac->down_req[i]->handle, LDB_WAIT_NONE);
967                 
968                 if (ret != LDB_SUCCESS) {
969                         handle->status = ret;
970                         goto done;
971                 }
972                 if (ac->down_req[i]->handle->status != LDB_SUCCESS) {
973                         handle->status = ac->down_req[i]->handle->status;
974                         goto done;
975                 }
976                 
977                 if (ac->down_req[i]->handle->state != LDB_ASYNC_DONE) {
978                         return LDB_SUCCESS;
979                 }
980         }
981
982         ret = LDB_SUCCESS;
983
984 done:
985         handle->state = LDB_ASYNC_DONE;
986         return ret;
987 }
988
989
990 static int partition_wait_all(struct ldb_handle *handle) {
991
992         int ret;
993
994         while (handle->state != LDB_ASYNC_DONE) {
995                 ret = partition_wait_none(handle);
996                 if (ret != LDB_SUCCESS) {
997                         return ret;
998                 }
999         }
1000
1001         return handle->status;
1002 }
1003
1004 static int partition_wait(struct ldb_handle *handle, enum ldb_wait_type type)
1005 {
1006         if (type == LDB_WAIT_ALL) {
1007                 return partition_wait_all(handle);
1008         } else {
1009                 return partition_wait_none(handle);
1010         }
1011 }
1012
1013 static const struct ldb_module_ops partition_ops = {
1014         .name              = "partition",
1015         .init_context      = partition_init,
1016         .search            = partition_search,
1017         .add               = partition_add,
1018         .modify            = partition_modify,
1019         .del               = partition_delete,
1020         .rename            = partition_rename,
1021         .extended          = partition_extended,
1022         .sequence_number   = partition_sequence_number,
1023         .start_transaction = partition_start_trans,
1024         .end_transaction   = partition_end_trans,
1025         .del_transaction   = partition_del_trans,
1026         .wait              = partition_wait
1027 };
1028
1029 int ldb_partition_init(void)
1030 {
1031         return ldb_register_module(&partition_ops);
1032 }