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