source4/dsdb/samdb/ldb_modules/partition.c

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