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