source4/dsdb/samdb/ldb_modules/partition.c

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