s4:partition LDB module - fill in parent requests for inheriting the flags
[nivanova/samba-autobuild/.git] / source4 / dsdb / samdb / ldb_modules / partition.c
1 /* 
2    Partitions ldb module
3
4    Copyright (C) Andrew Bartlett <abartlet@samba.org> 2006
5    Copyright (C) Stefan Metzmacher <metze@samba.org> 2007
6
7    This program is free software; you can redistribute it and/or modify
8    it under the terms of the GNU General Public License as published by
9    the Free Software Foundation; either version 3 of the License, or
10    (at your option) any later version.
11    
12    This program is distributed in the hope that it will be useful,
13    but WITHOUT ANY WARRANTY; without even the implied warranty of
14    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15    GNU General Public License for more details.
16    
17    You should have received a copy of the GNU General Public License
18    along with this program.  If not, see <http://www.gnu.org/licenses/>.
19 */
20
21 /*
22  *  Name: ldb
23  *
24  *  Component: ldb partitions module
25  *
26  *  Description: Implement LDAP partitions
27  *
28  *  Author: Andrew Bartlett
29  *  Author: Stefan Metzmacher
30  */
31
32 #include "dsdb/samdb/ldb_modules/partition.h"
33
34 struct part_request {
35         struct ldb_module *module;
36         struct ldb_request *req;
37 };
38
39 struct partition_context {
40         struct ldb_module *module;
41         struct ldb_request *req;
42
43         struct part_request *part_req;
44         unsigned int num_requests;
45         unsigned int finished_requests;
46
47         const char **referrals;
48 };
49
50 static struct partition_context *partition_init_ctx(struct ldb_module *module, struct ldb_request *req)
51 {
52         struct partition_context *ac;
53
54         ac = talloc_zero(req, struct partition_context);
55         if (ac == NULL) {
56                 ldb_set_errstring(ldb_module_get_ctx(module), "Out of Memory");
57                 return NULL;
58         }
59
60         ac->module = module;
61         ac->req = req;
62
63         return ac;
64 }
65
66 /*
67  * helper functions to call the next module in chain
68  */
69 int partition_request(struct ldb_module *module, struct ldb_request *request)
70 {
71         if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) { \
72                 const struct dsdb_control_current_partition *partition = NULL;
73                 struct ldb_control *partition_ctrl = ldb_request_get_control(request, DSDB_CONTROL_CURRENT_PARTITION_OID);
74                 if (partition_ctrl) {
75                         partition = talloc_get_type(partition_ctrl->data,
76                                                     struct dsdb_control_current_partition);
77                 }
78
79                 if (partition != NULL) {
80                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_request() -> %s",
81                                   ldb_dn_get_linearized(partition->dn));                        
82                 } else {
83                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_request() -> (metadata partition)");
84                 }
85         }
86
87         return ldb_next_request(module, request);
88 }
89
90 static struct dsdb_partition *find_partition(struct partition_private_data *data,
91                                              struct ldb_dn *dn,
92                                              struct ldb_request *req)
93 {
94         unsigned int i;
95         struct ldb_control *partition_ctrl;
96
97         /* see if the request has the partition DN specified in a
98          * control. The repl_meta_data module can specify this to
99          * ensure that replication happens to the right partition
100          */
101         partition_ctrl = ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID);
102         if (partition_ctrl) {
103                 const struct dsdb_control_current_partition *partition;
104                 partition = talloc_get_type(partition_ctrl->data,
105                                             struct dsdb_control_current_partition);
106                 if (partition != NULL) {
107                         dn = partition->dn;
108                 }
109         }
110
111         if (dn == NULL) {
112                 return NULL;
113         }
114
115         /* Look at base DN */
116         /* Figure out which partition it is under */
117         /* Skip the lot if 'data' isn't here yet (initialisation) */
118         for (i=0; data && data->partitions && data->partitions[i]; i++) {
119                 if (ldb_dn_compare_base(data->partitions[i]->ctrl->dn, dn) == 0) {
120                         return data->partitions[i];
121                 }
122         }
123
124         return NULL;
125 }
126
127 /**
128  * fire the caller's callback for every entry, but only send 'done' once.
129  */
130 static int partition_req_callback(struct ldb_request *req,
131                                   struct ldb_reply *ares)
132 {
133         struct partition_context *ac;
134         struct ldb_module *module;
135         struct ldb_request *nreq;
136         int ret;
137         struct partition_private_data *data;
138         struct ldb_control *partition_ctrl;
139
140         ac = talloc_get_type(req->context, struct partition_context);
141         data = talloc_get_type(ldb_module_get_private(ac->module), struct partition_private_data);
142
143         if (!ares) {
144                 return ldb_module_done(ac->req, NULL, NULL,
145                                         LDB_ERR_OPERATIONS_ERROR);
146         }
147
148         partition_ctrl = ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID);
149         if (partition_ctrl && (ac->num_requests == 1 || ares->type == LDB_REPLY_ENTRY)) {
150                 /* If we didn't fan this request out to mulitple partitions,
151                  * or this is an individual search result, we can
152                  * deterministically tell the caller what partition this was
153                  * written to (repl_meta_data likes to know) */
154                 ret = ldb_reply_add_control(ares,
155                                             DSDB_CONTROL_CURRENT_PARTITION_OID,
156                                             false, partition_ctrl->data);
157                 if (ret != LDB_SUCCESS) {
158                         return ldb_module_done(ac->req, NULL, NULL,
159                                                ret);
160                 }
161         }
162
163         if (ares->error != LDB_SUCCESS) {
164                 return ldb_module_done(ac->req, ares->controls,
165                                         ares->response, ares->error);
166         }
167
168         switch (ares->type) {
169         case LDB_REPLY_REFERRAL:
170                 return ldb_module_send_referral(ac->req, ares->referral);
171
172         case LDB_REPLY_ENTRY:
173                 if (ac->req->operation != LDB_SEARCH) {
174                         ldb_set_errstring(ldb_module_get_ctx(ac->module),
175                                 "partition_req_callback:"
176                                 " Unsupported reply type for this request");
177                         return ldb_module_done(ac->req, NULL, NULL,
178                                                 LDB_ERR_OPERATIONS_ERROR);
179                 }
180                 
181                 return ldb_module_send_entry(ac->req, ares->message, ares->controls);
182
183         case LDB_REPLY_DONE:
184                 if (ac->req->operation == LDB_EXTENDED) {
185                         /* FIXME: check for ares->response, replmd does not fill it ! */
186                         if (ares->response) {
187                                 if (strcmp(ares->response->oid, LDB_EXTENDED_START_TLS_OID) != 0) {
188                                         ldb_set_errstring(ldb_module_get_ctx(ac->module),
189                                                           "partition_req_callback:"
190                                                           " Unknown extended reply, "
191                                                           "only supports START_TLS");
192                                         talloc_free(ares);
193                                         return ldb_module_done(ac->req, NULL, NULL,
194                                                                 LDB_ERR_OPERATIONS_ERROR);
195                                 }
196                         }
197                 }
198
199                 ac->finished_requests++;
200                 if (ac->finished_requests == ac->num_requests) {
201                         /* Send back referrals if they do exist (search ops) */
202                         if (ac->referrals != NULL) {
203                                 const char **ref;
204                                 for (ref = ac->referrals; *ref != NULL; ++ref) {
205                                         ret = ldb_module_send_referral(ac->req,
206                                                                        talloc_strdup(ac->req, *ref));
207                                         if (ret != LDB_SUCCESS) {
208                                                 return ldb_module_done(ac->req, NULL, NULL,
209                                                                        ret);
210                                         }
211                                 }
212                         }
213
214                         /* this was the last one, call callback */
215                         return ldb_module_done(ac->req, ares->controls,
216                                                ares->response, 
217                                                ares->error);
218                 }
219
220                 /* not the last, now call the next one */
221                 module = ac->part_req[ac->finished_requests].module;
222                 nreq = ac->part_req[ac->finished_requests].req;
223
224                 ret = partition_request(module, nreq);
225                 if (ret != LDB_SUCCESS) {
226                         talloc_free(ares);
227                         return ldb_module_done(ac->req, NULL, NULL, ret);
228                 }
229
230                 break;
231         }
232
233         talloc_free(ares);
234         return LDB_SUCCESS;
235 }
236
237 static int partition_prep_request(struct partition_context *ac,
238                                   struct dsdb_partition *partition)
239 {
240         int ret;
241         struct ldb_request *req;
242
243         ac->part_req = talloc_realloc(ac, ac->part_req,
244                                         struct part_request,
245                                         ac->num_requests + 1);
246         if (ac->part_req == NULL) {
247                 return ldb_oom(ldb_module_get_ctx(ac->module));
248         }
249
250         switch (ac->req->operation) {
251         case LDB_SEARCH:
252                 ret = ldb_build_search_req_ex(&req, ldb_module_get_ctx(ac->module),
253                                         ac->part_req,
254                                         ac->req->op.search.base,
255                                         ac->req->op.search.scope,
256                                         ac->req->op.search.tree,
257                                         ac->req->op.search.attrs,
258                                         ac->req->controls,
259                                         ac, partition_req_callback,
260                                         ac->req);
261                 LDB_REQ_SET_LOCATION(req);
262                 break;
263         case LDB_ADD:
264                 ret = ldb_build_add_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
265                                         ac->req->op.add.message,
266                                         ac->req->controls,
267                                         ac, partition_req_callback,
268                                         ac->req);
269                 LDB_REQ_SET_LOCATION(req);
270                 break;
271         case LDB_MODIFY:
272                 ret = ldb_build_mod_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
273                                         ac->req->op.mod.message,
274                                         ac->req->controls,
275                                         ac, partition_req_callback,
276                                         ac->req);
277                 LDB_REQ_SET_LOCATION(req);
278                 break;
279         case LDB_DELETE:
280                 ret = ldb_build_del_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
281                                         ac->req->op.del.dn,
282                                         ac->req->controls,
283                                         ac, partition_req_callback,
284                                         ac->req);
285                 LDB_REQ_SET_LOCATION(req);
286                 break;
287         case LDB_RENAME:
288                 ret = ldb_build_rename_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
289                                         ac->req->op.rename.olddn,
290                                         ac->req->op.rename.newdn,
291                                         ac->req->controls,
292                                         ac, partition_req_callback,
293                                         ac->req);
294                 LDB_REQ_SET_LOCATION(req);
295                 break;
296         case LDB_EXTENDED:
297                 ret = ldb_build_extended_req(&req, ldb_module_get_ctx(ac->module),
298                                         ac->part_req,
299                                         ac->req->op.extended.oid,
300                                         ac->req->op.extended.data,
301                                         ac->req->controls,
302                                         ac, partition_req_callback,
303                                         ac->req);
304                 LDB_REQ_SET_LOCATION(req);
305                 break;
306         default:
307                 ldb_set_errstring(ldb_module_get_ctx(ac->module),
308                                   "Unsupported request type!");
309                 ret = LDB_ERR_UNWILLING_TO_PERFORM;
310         }
311
312         if (ret != LDB_SUCCESS) {
313                 return ret;
314         }
315
316         ac->part_req[ac->num_requests].req = req;
317
318         if (ac->req->controls) {
319                 req->controls = talloc_memdup(req, ac->req->controls,
320                                         talloc_get_size(ac->req->controls));
321                 if (req->controls == NULL) {
322                         return ldb_oom(ldb_module_get_ctx(ac->module));
323                 }
324         }
325
326         if (partition) {
327                 ac->part_req[ac->num_requests].module = partition->module;
328
329                 if (!ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID)) {
330                         ret = ldb_request_add_control(req,
331                                                       DSDB_CONTROL_CURRENT_PARTITION_OID,
332                                                       false, partition->ctrl);
333                         if (ret != LDB_SUCCESS) {
334                                 return ret;
335                         }
336                 }
337
338                 if (req->operation == LDB_SEARCH) {
339                         /* If the search is for 'more' than this partition,
340                          * then change the basedn, so a remote LDAP server
341                          * doesn't object */
342                         if (ldb_dn_compare_base(partition->ctrl->dn,
343                                                 req->op.search.base) != 0) {
344                                 req->op.search.base = partition->ctrl->dn;
345                         }
346                 }
347
348         } else {
349                 /* make sure you put the module here, or
350                  * or ldb_next_request() will skip a module */
351                 ac->part_req[ac->num_requests].module = ac->module;
352         }
353
354         ac->num_requests++;
355
356         return LDB_SUCCESS;
357 }
358
359 static int partition_call_first(struct partition_context *ac)
360 {
361         return partition_request(ac->part_req[0].module, ac->part_req[0].req);
362 }
363
364 /**
365  * Send a request down to all the partitions
366  */
367 static int partition_send_all(struct ldb_module *module, 
368                               struct partition_context *ac, 
369                               struct ldb_request *req) 
370 {
371         unsigned int i;
372         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
373                                                               struct partition_private_data);
374         int ret = partition_prep_request(ac, NULL);
375         if (ret != LDB_SUCCESS) {
376                 return ret;
377         }
378         for (i=0; data && data->partitions && data->partitions[i]; i++) {
379                 ret = partition_prep_request(ac, data->partitions[i]);
380                 if (ret != LDB_SUCCESS) {
381                         return ret;
382                 }
383         }
384
385         /* fire the first one */
386         return partition_call_first(ac);
387 }
388
389 /**
390  * Figure out which backend a request needs to be aimed at.  Some
391  * requests must be replicated to all backends
392  */
393 static int partition_replicate(struct ldb_module *module, struct ldb_request *req, struct ldb_dn *dn) 
394 {
395         struct partition_context *ac;
396         unsigned int i;
397         int ret;
398         struct dsdb_partition *partition;
399         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
400                                                               struct partition_private_data);
401
402         /* if we aren't initialised yet go further */
403         if (!data || !data->partitions) {
404                 return ldb_next_request(module, req);
405         }
406
407         if (req->operation != LDB_SEARCH && ldb_dn_is_special(dn)) {
408                 /* Is this a special DN, we need to replicate to every backend? */
409                 for (i=0; data->replicate && data->replicate[i]; i++) {
410                         if (ldb_dn_compare(data->replicate[i], 
411                                            dn) == 0) {
412                                 
413                                 ac = partition_init_ctx(module, req);
414                                 if (!ac) {
415                                         return ldb_operr(ldb_module_get_ctx(module));
416                                 }
417                                 
418                                 return partition_send_all(module, ac, req);
419                         }
420                 }
421         }
422
423         /* Otherwise, we need to find the partition to fire it to */
424
425         /* Find partition */
426         partition = find_partition(data, dn, req);
427         if (!partition) {
428                 /*
429                  * if we haven't found a matching partition
430                  * pass the request to the main ldb
431                  *
432                  * TODO: we should maybe return an error here
433                  *       if it's not a special dn
434                  */
435
436                 return ldb_next_request(module, req);
437         }
438
439         ac = partition_init_ctx(module, req);
440         if (!ac) {
441                 return ldb_operr(ldb_module_get_ctx(module));
442         }
443
444         /* we need to add a control but we never touch the original request */
445         ret = partition_prep_request(ac, partition);
446         if (ret != LDB_SUCCESS) {
447                 return ret;
448         }
449
450         /* fire the first one */
451         return partition_call_first(ac);
452 }
453
454 /* search */
455 static int partition_search(struct ldb_module *module, struct ldb_request *req)
456 {
457         struct ldb_control **saved_controls;
458         /* Find backend */
459         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
460                                                               struct partition_private_data);
461         struct partition_context *ac;
462         struct ldb_context *ldb;
463         struct loadparm_context *lp_ctx;
464
465         struct ldb_control *search_control = ldb_request_get_control(req, LDB_CONTROL_SEARCH_OPTIONS_OID);
466         struct ldb_control *domain_scope_control = ldb_request_get_control(req, LDB_CONTROL_DOMAIN_SCOPE_OID);
467         
468         struct ldb_search_options_control *search_options = NULL;
469         struct dsdb_partition *p;
470         unsigned int i, j;
471         int ret;
472         bool domain_scope = false, phantom_root = false;
473
474         /* see if we are still up-to-date */
475         ret = partition_reload_if_required(module, data, req);
476         if (ret != LDB_SUCCESS) {
477                 return ret;
478         }
479
480         p = find_partition(data, NULL, req);
481         if (p != NULL) {
482                 /* the caller specified what partition they want the
483                  * search - just pass it on
484                  */
485                 return ldb_next_request(p->module, req);
486         }
487
488         /* Get back the search options from the search control, and mark it as
489          * non-critical (to make backends and also dcpromo happy).
490          */
491         if (search_control) {
492                 search_options = talloc_get_type(search_control->data, struct ldb_search_options_control);
493                 search_control->critical = 0;
494
495         }
496
497         /* Remove the "domain_scope" control, so we don't confuse a backend
498          * server */
499         if (domain_scope_control && !ldb_save_controls(domain_scope_control, req, &saved_controls)) {
500                 return ldb_oom(ldb_module_get_ctx(module));
501         }
502
503         /* Locate the options */
504         domain_scope = (search_options
505                 && (search_options->search_options & LDB_SEARCH_OPTION_DOMAIN_SCOPE))
506                 || domain_scope_control;
507         phantom_root = search_options
508                 && (search_options->search_options & LDB_SEARCH_OPTION_PHANTOM_ROOT);
509
510         /* Remove handled options from the search control flag */
511         if (search_options) {
512                 search_options->search_options = search_options->search_options
513                         & ~LDB_SEARCH_OPTION_DOMAIN_SCOPE
514                         & ~LDB_SEARCH_OPTION_PHANTOM_ROOT;
515         }
516
517         if (!data || !data->partitions) {
518                 return ldb_next_request(module, req);
519         }
520
521         ac = partition_init_ctx(module, req);
522         if (!ac) {
523                 return ldb_operr(ldb_module_get_ctx(module));
524         }
525
526         ldb = ldb_module_get_ctx(ac->module);
527         lp_ctx = talloc_get_type(ldb_get_opaque(ldb, "loadparm"),
528                                                 struct loadparm_context);
529
530         /* Search from the base DN */
531         if (ldb_dn_is_null(req->op.search.base)) {
532                 return partition_send_all(module, ac, req);
533         }
534
535         for (i=0; data->partitions[i]; i++) {
536                 bool match = false, stop = false;
537
538                 if (phantom_root) {
539                         /* Phantom root: Find all partitions under the
540                          * search base. We match if:
541                          *
542                          * 1) the DN we are looking for exactly matches a
543                          *    certain partition and always stop
544                          * 2) the DN we are looking for is a parent of certain
545                          *    partitions and it isn't a scope base search
546                          * 3) the DN we are looking for is a child of a certain
547                          *    partition and always stop
548                          *    - we don't need to go any further up in the
549                          *    hierarchy!
550                          */
551                         if (ldb_dn_compare(data->partitions[i]->ctrl->dn,
552                                            req->op.search.base) == 0) {
553                                 match = true;
554                                 stop = true;
555                         }
556                         if (!match &&
557                             (ldb_dn_compare_base(req->op.search.base,
558                                                  data->partitions[i]->ctrl->dn) == 0 &&
559                              req->op.search.scope != LDB_SCOPE_BASE)) {
560                                 match = true;
561                         }
562                         if (!match &&
563                             ldb_dn_compare_base(data->partitions[i]->ctrl->dn,
564                                                 req->op.search.base) == 0) {
565                                 match = true;
566                                 stop = true; /* note that this relies on partition ordering */
567                         }
568                 } else {
569                         /* Domain scope: Find all partitions under the search
570                          * base.
571                          *
572                          * We generate referral candidates if we haven't
573                          * specified the domain scope control, haven't a base
574                          * search* scope and the DN we are looking for is a real
575                          * predecessor of certain partitions. When a new
576                          * referral candidate is nearer to the DN than an
577                          * existing one delete the latter (we want to have only
578                          * the closest ones). When we checked this for all
579                          * candidates we have the final referrals.
580                          *
581                          * We match if the DN we are looking for is a child of
582                          * a certain partition or the partition
583                          * DN itself - we don't need to go any further
584                          * up in the hierarchy!
585                          */
586                         if ((!domain_scope) &&
587                             (req->op.search.scope != LDB_SCOPE_BASE) &&
588                             (ldb_dn_compare_base(req->op.search.base,
589                                                  data->partitions[i]->ctrl->dn) == 0) &&
590                             (ldb_dn_compare(req->op.search.base,
591                                             data->partitions[i]->ctrl->dn) != 0)) {
592                                 char *ref = talloc_asprintf(ac,
593                                                             "ldap://%s/%s%s",
594                                                             lpcfg_dnsdomain(lp_ctx),
595                                                             ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
596                                                             req->op.search.scope == LDB_SCOPE_ONELEVEL ? "??base" : "");
597
598                                 if (ref == NULL) {
599                                         return ldb_oom(ldb);
600                                 }
601
602                                 /* Initialise the referrals list */
603                                 if (ac->referrals == NULL) {
604                                         ac->referrals = (const char **) str_list_make_empty(ac);
605                                         if (ac->referrals == NULL) {
606                                                 return ldb_oom(ldb);
607                                         }
608                                 }
609
610                                 /* Check if the new referral candidate is
611                                  * closer to the base DN than already
612                                  * saved ones and delete the latters */
613                                 j = 0;
614                                 while (ac->referrals[j] != NULL) {
615                                         if (strstr(ac->referrals[j],
616                                                    ldb_dn_get_linearized(data->partitions[i]->ctrl->dn)) != NULL) {
617                                                 str_list_remove(ac->referrals,
618                                                                 ac->referrals[j]);
619                                         } else {
620                                                 ++j;
621                                         }
622                                 }
623
624                                 /* Add our new candidate */
625                                 ac->referrals = str_list_add(ac->referrals, ref);
626
627                                 talloc_free(ref);
628
629                                 if (ac->referrals == NULL) {
630                                         return ldb_oom(ldb);
631                                 }
632                         }
633                         if (ldb_dn_compare_base(data->partitions[i]->ctrl->dn, req->op.search.base) == 0) {
634                                 match = true;
635                                 stop = true; /* note that this relies on partition ordering */
636                         }
637                 }
638
639                 if (match) {
640                         ret = partition_prep_request(ac, data->partitions[i]);
641                         if (ret != LDB_SUCCESS) {
642                                 return ret;
643                         }
644                 }
645
646                 if (stop) break;
647         }
648
649         /* Perhaps we didn't match any partitions. Try the main partition */
650         if (ac->num_requests == 0) {
651                 talloc_free(ac);
652                 return ldb_next_request(module, req);
653         }
654
655         /* fire the first one */
656         return partition_call_first(ac);
657 }
658
659 /* add */
660 static int partition_add(struct ldb_module *module, struct ldb_request *req)
661 {
662         return partition_replicate(module, req, req->op.add.message->dn);
663 }
664
665 /* modify */
666 static int partition_modify(struct ldb_module *module, struct ldb_request *req)
667 {
668         return partition_replicate(module, req, req->op.mod.message->dn);
669 }
670
671 /* delete */
672 static int partition_delete(struct ldb_module *module, struct ldb_request *req)
673 {
674         return partition_replicate(module, req, req->op.del.dn);
675 }
676
677 /* rename */
678 static int partition_rename(struct ldb_module *module, struct ldb_request *req)
679 {
680         /* Find backend */
681         struct dsdb_partition *backend, *backend2;
682         
683         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
684                                                               struct partition_private_data);
685
686         /* Skip the lot if 'data' isn't here yet (initialisation) */
687         if (!data) {
688                 return ldb_operr(ldb_module_get_ctx(module));
689         }
690
691         backend = find_partition(data, req->op.rename.olddn, req);
692         backend2 = find_partition(data, req->op.rename.newdn, req);
693
694         if ((backend && !backend2) || (!backend && backend2)) {
695                 return LDB_ERR_AFFECTS_MULTIPLE_DSAS;
696         }
697
698         if (backend != backend2) {
699                 ldb_asprintf_errstring(ldb_module_get_ctx(module), 
700                                        "Cannot rename from %s in %s to %s in %s: %s",
701                                        ldb_dn_get_linearized(req->op.rename.olddn),
702                                        ldb_dn_get_linearized(backend->ctrl->dn),
703                                        ldb_dn_get_linearized(req->op.rename.newdn),
704                                        ldb_dn_get_linearized(backend2->ctrl->dn),
705                                        ldb_strerror(LDB_ERR_AFFECTS_MULTIPLE_DSAS));
706                 return LDB_ERR_AFFECTS_MULTIPLE_DSAS;
707         }
708
709         return partition_replicate(module, req, req->op.rename.olddn);
710 }
711
712 /* start a transaction */
713 static int partition_start_trans(struct ldb_module *module)
714 {
715         unsigned int i;
716         int ret;
717         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
718                                                               struct partition_private_data);
719         /* Look at base DN */
720         /* Figure out which partition it is under */
721         /* Skip the lot if 'data' isn't here yet (initialization) */
722         if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
723                 ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_start_trans() -> (metadata partition)");
724         }
725         ret = ldb_next_start_trans(module);
726         if (ret != LDB_SUCCESS) {
727                 return ret;
728         }
729
730         ret = partition_reload_if_required(module, data, NULL);
731         if (ret != LDB_SUCCESS) {
732                 return ret;
733         }
734
735         for (i=0; data && data->partitions && data->partitions[i]; i++) {
736                 if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
737                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_start_trans() -> %s",
738                                   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
739                 }
740                 ret = ldb_next_start_trans(data->partitions[i]->module);
741                 if (ret != LDB_SUCCESS) {
742                         /* Back it out, if it fails on one */
743                         for (i--; i >= 0; i--) {
744                                 ldb_next_del_trans(data->partitions[i]->module);
745                         }
746                         ldb_next_del_trans(module);
747                         return ret;
748                 }
749         }
750
751         data->in_transaction++;
752
753         return LDB_SUCCESS;
754 }
755
756 /* prepare for a commit */
757 static int partition_prepare_commit(struct ldb_module *module)
758 {
759         unsigned int i;
760         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
761                                                               struct partition_private_data);
762
763         for (i=0; data && data->partitions && data->partitions[i]; i++) {
764                 int ret;
765
766                 if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
767                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_prepare_commit() -> %s",
768                                   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
769                 }
770                 ret = ldb_next_prepare_commit(data->partitions[i]->module);
771                 if (ret != LDB_SUCCESS) {
772                         ldb_asprintf_errstring(ldb_module_get_ctx(module), "prepare_commit error on %s: %s",
773                                                ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
774                                                ldb_errstring(ldb_module_get_ctx(module)));
775                         return ret;
776                 }
777         }
778
779         if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
780                 ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_prepare_commit() -> (metadata partition)");
781         }
782         return ldb_next_prepare_commit(module);
783 }
784
785
786 /* end a transaction */
787 static int partition_end_trans(struct ldb_module *module)
788 {
789         int ret, ret2;
790         unsigned int i;
791         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
792                                                               struct partition_private_data);
793
794         ret = LDB_SUCCESS;
795
796         if (data->in_transaction == 0) {
797                 DEBUG(0,("partition end transaction mismatch\n"));
798                 ret = LDB_ERR_OPERATIONS_ERROR;
799         } else {
800                 data->in_transaction--;
801         }
802
803         for (i=0; data && data->partitions && data->partitions[i]; i++) {
804                 if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
805                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_end_trans() -> %s",
806                                   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
807                 }
808                 ret2 = ldb_next_end_trans(data->partitions[i]->module);
809                 if (ret2 != LDB_SUCCESS) {
810                         ldb_asprintf_errstring(ldb_module_get_ctx(module), "end_trans error on %s: %s",
811                                                ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
812                                                ldb_errstring(ldb_module_get_ctx(module)));
813                         ret = ret2;
814                 }
815         }
816
817         if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
818                 ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_end_trans() -> (metadata partition)");
819         }
820         ret2 = ldb_next_end_trans(module);
821         if (ret2 != LDB_SUCCESS) {
822                 ret = ret2;
823         }
824         return ret;
825 }
826
827 /* delete a transaction */
828 static int partition_del_trans(struct ldb_module *module)
829 {
830         int ret, final_ret = LDB_SUCCESS;
831         unsigned int i;
832         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
833                                                               struct partition_private_data);
834         for (i=0; data && data->partitions && data->partitions[i]; i++) {
835                 if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
836                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_del_trans() -> %s",
837                                   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
838                 }
839                 ret = ldb_next_del_trans(data->partitions[i]->module);
840                 if (ret != LDB_SUCCESS) {
841                         ldb_asprintf_errstring(ldb_module_get_ctx(module), "del_trans error on %s: %s",
842                                                ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
843                                                ldb_errstring(ldb_module_get_ctx(module)));
844                         final_ret = ret;
845                 }
846         }       
847
848         if (data->in_transaction == 0) {
849                 DEBUG(0,("partition del transaction mismatch\n"));
850                 return ldb_operr(ldb_module_get_ctx(module));
851         }
852         data->in_transaction--;
853
854         if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
855                 ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_del_trans() -> (metadata partition)");
856         }
857         ret = ldb_next_del_trans(module);
858         if (ret != LDB_SUCCESS) {
859                 final_ret = ret;
860         }
861         return final_ret;
862 }
863
864 int partition_primary_sequence_number(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
865                                      enum ldb_sequence_type type, uint64_t *seq_number) 
866 {
867         int ret;
868         struct ldb_result *res;
869         struct ldb_seqnum_request *tseq;
870         struct ldb_request *treq;
871         struct ldb_seqnum_result *seqr;
872         res = talloc_zero(mem_ctx, struct ldb_result);
873         if (res == NULL) {
874                 return ldb_oom(ldb_module_get_ctx(module));
875         }
876         tseq = talloc_zero(res, struct ldb_seqnum_request);
877         if (tseq == NULL) {
878                 talloc_free(res);
879                 return ldb_oom(ldb_module_get_ctx(module));
880         }
881         tseq->type = type;
882         
883         ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
884                                      LDB_EXTENDED_SEQUENCE_NUMBER,
885                                      tseq,
886                                      NULL,
887                                      res,
888                                      ldb_extended_default_callback,
889                                      NULL);
890         LDB_REQ_SET_LOCATION(treq);
891         if (ret != LDB_SUCCESS) {
892                 talloc_free(res);
893                 return ret;
894         }
895         
896         ret = ldb_next_request(module, treq);
897         if (ret != LDB_SUCCESS) {
898                 talloc_free(res);
899                 return ret;
900         }
901         ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
902         if (ret != LDB_SUCCESS) {
903                 talloc_free(res);
904                 return ret;
905         }
906         
907         seqr = talloc_get_type(res->extended->data,
908                                struct ldb_seqnum_result);
909         if (seqr->flags & LDB_SEQ_TIMESTAMP_SEQUENCE) {
910                 ret = LDB_ERR_OPERATIONS_ERROR;
911                 ldb_set_errstring(ldb_module_get_ctx(module), "Primary backend in partitions module returned a timestamp based seq number (must return a normal number)");
912                 talloc_free(res);
913                 return ret;
914         } else {
915                 *seq_number = seqr->seq_num;
916         }
917         talloc_free(res);
918         return LDB_SUCCESS;
919 }
920
921 /* FIXME: This function is still semi-async */
922 static int partition_sequence_number(struct ldb_module *module, struct ldb_request *req)
923 {
924         int ret;
925         unsigned int i;
926         uint64_t seq_number = 0;
927         uint64_t timestamp_sequence = 0;
928         uint64_t timestamp = 0;
929         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
930                                                               struct partition_private_data);
931         struct ldb_seqnum_request *seq;
932         struct ldb_seqnum_result *seqr;
933         struct ldb_request *treq;
934         struct ldb_seqnum_request *tseq;
935         struct ldb_seqnum_result *tseqr;
936         struct ldb_extended *ext;
937         struct ldb_result *res;
938         struct dsdb_partition *p;
939
940         p = find_partition(data, NULL, req);
941         if (p != NULL) {
942                 /* the caller specified what partition they want the
943                  * sequence number operation on - just pass it on
944                  */
945                 return ldb_next_request(p->module, req);                
946         }
947
948         seq = talloc_get_type(req->op.extended.data, struct ldb_seqnum_request);
949
950         switch (seq->type) {
951         case LDB_SEQ_NEXT:
952         case LDB_SEQ_HIGHEST_SEQ:
953
954                 ret = partition_primary_sequence_number(module, req, seq->type, &seq_number);
955                 if (ret != LDB_SUCCESS) {
956                         return ret;
957                 }
958
959                 /* Skip the lot if 'data' isn't here yet (initialisation) */
960                 for (i=0; data && data->partitions && data->partitions[i]; i++) {
961
962                         res = talloc_zero(req, struct ldb_result);
963                         if (res == NULL) {
964                                 return ldb_oom(ldb_module_get_ctx(module));
965                         }
966                         tseq = talloc_zero(res, struct ldb_seqnum_request);
967                         if (tseq == NULL) {
968                                 talloc_free(res);
969                                 return ldb_oom(ldb_module_get_ctx(module));
970                         }
971                         tseq->type = seq->type;
972
973                         ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
974                                                      LDB_EXTENDED_SEQUENCE_NUMBER,
975                                                      tseq,
976                                                      NULL,
977                                                      res,
978                                                      ldb_extended_default_callback,
979                                                      req);
980                         LDB_REQ_SET_LOCATION(treq);
981                         if (ret != LDB_SUCCESS) {
982                                 talloc_free(res);
983                                 return ret;
984                         }
985
986                         if (!ldb_request_get_control(treq, DSDB_CONTROL_CURRENT_PARTITION_OID)) {
987                                 ret = ldb_request_add_control(treq,
988                                                               DSDB_CONTROL_CURRENT_PARTITION_OID,
989                                                               false, data->partitions[i]->ctrl);
990                                 if (ret != LDB_SUCCESS) {
991                                         talloc_free(res);
992                                         return ret;
993                                 }
994                         }
995
996                         ret = partition_request(data->partitions[i]->module, treq);
997                         if (ret != LDB_SUCCESS) {
998                                 talloc_free(res);
999                                 return ret;
1000                         }
1001                         ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
1002                         if (ret != LDB_SUCCESS) {
1003                                 talloc_free(res);
1004                                 return ret;
1005                         }
1006                         tseqr = talloc_get_type(res->extended->data,
1007                                                 struct ldb_seqnum_result);
1008                         if (tseqr->flags & LDB_SEQ_TIMESTAMP_SEQUENCE) {
1009                                 timestamp_sequence = MAX(timestamp_sequence,
1010                                                          tseqr->seq_num);
1011                         } else {
1012                                 seq_number += tseqr->seq_num;
1013                         }
1014                         talloc_free(res);
1015                 }
1016                 /* fall through */
1017         case LDB_SEQ_HIGHEST_TIMESTAMP:
1018
1019                 res = talloc_zero(req, struct ldb_result);
1020                 if (res == NULL) {
1021                         return ldb_oom(ldb_module_get_ctx(module));
1022                 }
1023
1024                 tseq = talloc_zero(res, struct ldb_seqnum_request);
1025                 if (tseq == NULL) {
1026                         talloc_free(res);
1027                         return ldb_oom(ldb_module_get_ctx(module));
1028                 }
1029                 tseq->type = LDB_SEQ_HIGHEST_TIMESTAMP;
1030
1031                 ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
1032                                              LDB_EXTENDED_SEQUENCE_NUMBER,
1033                                              tseq,
1034                                              NULL,
1035                                              res,
1036                                              ldb_extended_default_callback,
1037                                              req);
1038                 LDB_REQ_SET_LOCATION(treq);
1039                 if (ret != LDB_SUCCESS) {
1040                         talloc_free(res);
1041                         return ret;
1042                 }
1043
1044                 ret = ldb_next_request(module, treq);
1045                 if (ret != LDB_SUCCESS) {
1046                         talloc_free(res);
1047                         return ret;
1048                 }
1049                 ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
1050                 if (ret != LDB_SUCCESS) {
1051                         talloc_free(res);
1052                         return ret;
1053                 }
1054
1055                 tseqr = talloc_get_type(res->extended->data,
1056                                            struct ldb_seqnum_result);
1057                 timestamp = tseqr->seq_num;
1058
1059                 talloc_free(res);
1060
1061                 /* Skip the lot if 'data' isn't here yet (initialisation) */
1062                 for (i=0; data && data->partitions && data->partitions[i]; i++) {
1063
1064                         res = talloc_zero(req, struct ldb_result);
1065                         if (res == NULL) {
1066                                 return ldb_oom(ldb_module_get_ctx(module));
1067                         }
1068
1069                         tseq = talloc_zero(res, struct ldb_seqnum_request);
1070                         if (tseq == NULL) {
1071                                 talloc_free(res);
1072                                 return ldb_oom(ldb_module_get_ctx(module));
1073                         }
1074                         tseq->type = LDB_SEQ_HIGHEST_TIMESTAMP;
1075
1076                         ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
1077                                                      LDB_EXTENDED_SEQUENCE_NUMBER,
1078                                                      tseq,
1079                                                      NULL,
1080                                                      res,
1081                                                      ldb_extended_default_callback,
1082                                                      req);
1083                         LDB_REQ_SET_LOCATION(treq);
1084                         if (ret != LDB_SUCCESS) {
1085                                 talloc_free(res);
1086                                 return ret;
1087                         }
1088
1089                         if (!ldb_request_get_control(treq, DSDB_CONTROL_CURRENT_PARTITION_OID)) {
1090                                 ret = ldb_request_add_control(treq,
1091                                                               DSDB_CONTROL_CURRENT_PARTITION_OID,
1092                                                               false, data->partitions[i]->ctrl);
1093                                 if (ret != LDB_SUCCESS) {
1094                                         talloc_free(res);
1095                                         return ret;
1096                                 }
1097                         }
1098
1099                         ret = partition_request(data->partitions[i]->module, treq);
1100                         if (ret != LDB_SUCCESS) {
1101                                 talloc_free(res);
1102                                 return ret;
1103                         }
1104                         ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
1105                         if (ret != LDB_SUCCESS) {
1106                                 talloc_free(res);
1107                                 return ret;
1108                         }
1109
1110                         tseqr = talloc_get_type(res->extended->data,
1111                                                   struct ldb_seqnum_result);
1112                         timestamp = MAX(timestamp, tseqr->seq_num);
1113
1114                         talloc_free(res);
1115                 }
1116
1117                 break;
1118         }
1119
1120         ext = talloc_zero(req, struct ldb_extended);
1121         if (!ext) {
1122                 return ldb_oom(ldb_module_get_ctx(module));
1123         }
1124         seqr = talloc_zero(ext, struct ldb_seqnum_result);
1125         if (seqr == NULL) {
1126                 talloc_free(ext);
1127                 return ldb_oom(ldb_module_get_ctx(module));
1128         }
1129         ext->oid = LDB_EXTENDED_SEQUENCE_NUMBER;
1130         ext->data = seqr;
1131
1132         switch (seq->type) {
1133         case LDB_SEQ_NEXT:
1134         case LDB_SEQ_HIGHEST_SEQ:
1135
1136                 /* Has someone above set a timebase sequence? */
1137                 if (timestamp_sequence) {
1138                         seqr->seq_num = (((unsigned long long)timestamp << 24) | (seq_number & 0xFFFFFF));
1139                 } else {
1140                         seqr->seq_num = seq_number;
1141                 }
1142
1143                 if (timestamp_sequence > seqr->seq_num) {
1144                         seqr->seq_num = timestamp_sequence;
1145                         seqr->flags |= LDB_SEQ_TIMESTAMP_SEQUENCE;
1146                 }
1147
1148                 seqr->flags |= LDB_SEQ_GLOBAL_SEQUENCE;
1149                 break;
1150         case LDB_SEQ_HIGHEST_TIMESTAMP:
1151                 seqr->seq_num = timestamp;
1152                 break;
1153         }
1154
1155         if (seq->type == LDB_SEQ_NEXT) {
1156                 seqr->seq_num++;
1157         }
1158
1159         /* send request done */
1160         return ldb_module_done(req, NULL, ext, LDB_SUCCESS);
1161 }
1162
1163 /* extended */
1164 static int partition_extended(struct ldb_module *module, struct ldb_request *req)
1165 {
1166         struct partition_private_data *data;
1167         struct partition_context *ac;
1168         int ret;
1169
1170         data = talloc_get_type(ldb_module_get_private(module), struct partition_private_data);
1171         if (!data) {
1172                 return ldb_next_request(module, req);
1173         }
1174
1175         ret = partition_reload_if_required(module, data, req);
1176         if (ret != LDB_SUCCESS) {
1177                 return ret;
1178         }
1179         
1180         if (strcmp(req->op.extended.oid, LDB_EXTENDED_SEQUENCE_NUMBER) == 0) {
1181                 return partition_sequence_number(module, req);
1182         }
1183
1184         if (strcmp(req->op.extended.oid, DSDB_EXTENDED_CREATE_PARTITION_OID) == 0) {
1185                 return partition_create(module, req);
1186         }
1187
1188         /* 
1189          * as the extended operation has no dn
1190          * we need to send it to all partitions
1191          */
1192
1193         ac = partition_init_ctx(module, req);
1194         if (!ac) {
1195                 return ldb_operr(ldb_module_get_ctx(module));
1196         }
1197
1198         return partition_send_all(module, ac, req);
1199 }
1200
1201 static const struct ldb_module_ops ldb_partition_module_ops = {
1202         .name              = "partition",
1203         .init_context      = partition_init,
1204         .search            = partition_search,
1205         .add               = partition_add,
1206         .modify            = partition_modify,
1207         .del               = partition_delete,
1208         .rename            = partition_rename,
1209         .extended          = partition_extended,
1210         .start_transaction = partition_start_trans,
1211         .prepare_commit    = partition_prepare_commit,
1212         .end_transaction   = partition_end_trans,
1213         .del_transaction   = partition_del_trans,
1214 };
1215
1216 int ldb_partition_module_init(const char *version)
1217 {
1218         LDB_MODULE_CHECK_VERSION(version);
1219         return ldb_register_module(&ldb_partition_module_ops);
1220 }