source4/dsdb/samdb/ldb_modules/partition.c

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