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 "dsdb/samdb/ldb_modules/partition.h"
  33
  34 struct part_request {
  35         struct ldb_module *module;
  36         struct ldb_request *req;
  37 };
  38
  39 struct partition_context {
  40         struct ldb_module *module;
  41         struct ldb_request *req;
  42
  43         struct part_request *part_req;
  44         unsigned int num_requests;
  45         unsigned int finished_requests;
  46
  47         const char **referrals;
  48 };
  49
  50 static struct partition_context *partition_init_ctx(struct ldb_module *module, struct ldb_request *req)
  51 {
  52         struct partition_context *ac;
  53
  54         ac = talloc_zero(req, struct partition_context);
  55         if (ac == NULL) {
  56                 ldb_set_errstring(ldb_module_get_ctx(module), "Out of Memory");
  57                 return NULL;
  58         }
  59
  60         ac->module = module;
  61         ac->req = req;
  62
  63         return ac;
  64 }
  65
  66 /*
  67  * helper functions to call the next module in chain
  68  */
  69 int partition_request(struct ldb_module *module, struct ldb_request *request)
  70 {
  71         if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) { \
  72                 const struct dsdb_control_current_partition *partition = NULL;
  73                 struct ldb_control *partition_ctrl = ldb_request_get_control(request, DSDB_CONTROL_CURRENT_PARTITION_OID);
  74                 if (partition_ctrl) {
  75                         partition = talloc_get_type(partition_ctrl->data,
  76                                                     struct dsdb_control_current_partition);
  77                 }
  78
  79                 if (partition != NULL) {
  80                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_request() -> %s",
  81                                   ldb_dn_get_linearized(partition->dn));
  82                 } else {
  83                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_request() -> (metadata partition)");
  84                 }
  85         }
  86
  87         return ldb_next_request(module, request);
  88 }
  89
  90 static struct dsdb_partition *find_partition(struct partition_private_data *data,
  91                                              struct ldb_dn *dn,
  92                                              struct ldb_request *req)
  93 {
  94         unsigned int i;
  95         struct ldb_control *partition_ctrl;
  96
  97         /* see if the request has the partition DN specified in a
  98          * control. The repl_meta_data module can specify this to
  99          * ensure that replication happens to the right partition
 100          */
 101         partition_ctrl = ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID);
 102         if (partition_ctrl) {
 103                 const struct dsdb_control_current_partition *partition;
 104                 partition = talloc_get_type(partition_ctrl->data,
 105                                             struct dsdb_control_current_partition);
 106                 if (partition != NULL) {
 107                         dn = partition->dn;
 108                 }
 109         }
 110
 111         if (dn == NULL) {
 112                 return NULL;
 113         }
 114
 115         /* Look at base DN */
 116         /* Figure out which partition it is under */
 117         /* Skip the lot if 'data' isn't here yet (initialisation) */
 118         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 119                 if (ldb_dn_compare_base(data->partitions[i]->ctrl->dn, dn) == 0) {
 120                         return data->partitions[i];
 121                 }
 122         }
 123
 124         return NULL;
 125 }
 126
 127 /**
 128  * fire the caller's callback for every entry, but only send 'done' once.
 129  */
 130 static int partition_req_callback(struct ldb_request *req,
 131                                   struct ldb_reply *ares)
 132 {
 133         struct partition_context *ac;
 134         struct ldb_module *module;
 135         struct ldb_request *nreq;
 136         int ret;
 137         struct ldb_control *partition_ctrl;
 138
 139         ac = talloc_get_type(req->context, struct partition_context);
 140
 141         if (!ares) {
 142                 return ldb_module_done(ac->req, NULL, NULL,
 143                                         LDB_ERR_OPERATIONS_ERROR);
 144         }
 145
 146         partition_ctrl = ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID);
 147         if (partition_ctrl && (ac->num_requests == 1 || ares->type == LDB_REPLY_ENTRY)) {
 148                 /* If we didn't fan this request out to mulitple partitions,
 149                  * or this is an individual search result, we can
 150                  * deterministically tell the caller what partition this was
 151                  * written to (repl_meta_data likes to know) */
 152                 ret = ldb_reply_add_control(ares,
 153                                             DSDB_CONTROL_CURRENT_PARTITION_OID,
 154                                             false, partition_ctrl->data);
 155                 if (ret != LDB_SUCCESS) {
 156                         return ldb_module_done(ac->req, NULL, NULL,
 157                                                ret);
 158                 }
 159         }
 160
 161         if (ares->error != LDB_SUCCESS) {
 162                 return ldb_module_done(ac->req, ares->controls,
 163                                         ares->response, ares->error);
 164         }
 165
 166         switch (ares->type) {
 167         case LDB_REPLY_REFERRAL:
 168                 return ldb_module_send_referral(ac->req, ares->referral);
 169
 170         case LDB_REPLY_ENTRY:
 171                 if (ac->req->operation != LDB_SEARCH) {
 172                         ldb_set_errstring(ldb_module_get_ctx(ac->module),
 173                                 "partition_req_callback:"
 174                                 " Unsupported reply type for this request");
 175                         return ldb_module_done(ac->req, NULL, NULL,
 176                                                 LDB_ERR_OPERATIONS_ERROR);
 177                 }
 178
 179                 return ldb_module_send_entry(ac->req, ares->message, ares->controls);
 180
 181         case LDB_REPLY_DONE:
 182                 if (ac->req->operation == LDB_EXTENDED) {
 183                         /* FIXME: check for ares->response, replmd does not fill it ! */
 184                         if (ares->response) {
 185                                 if (strcmp(ares->response->oid, LDB_EXTENDED_START_TLS_OID) != 0) {
 186                                         ldb_set_errstring(ldb_module_get_ctx(ac->module),
 187                                                           "partition_req_callback:"
 188                                                           " Unknown extended reply, "
 189                                                           "only supports START_TLS");
 190                                         talloc_free(ares);
 191                                         return ldb_module_done(ac->req, NULL, NULL,
 192                                                                 LDB_ERR_OPERATIONS_ERROR);
 193                                 }
 194                         }
 195                 }
 196
 197                 ac->finished_requests++;
 198                 if (ac->finished_requests == ac->num_requests) {
 199                         /* Send back referrals if they do exist (search ops) */
 200                         if (ac->referrals != NULL) {
 201                                 const char **ref;
 202                                 for (ref = ac->referrals; *ref != NULL; ++ref) {
 203                                         ret = ldb_module_send_referral(ac->req,
 204                                                                        talloc_strdup(ac->req, *ref));
 205                                         if (ret != LDB_SUCCESS) {
 206                                                 return ldb_module_done(ac->req, NULL, NULL,
 207                                                                        ret);
 208                                         }
 209                                 }
 210                         }
 211
 212                         /* this was the last one, call callback */
 213                         return ldb_module_done(ac->req, ares->controls,
 214                                                ares->response,
 215                                                ares->error);
 216                 }
 217
 218                 /* not the last, now call the next one */
 219                 module = ac->part_req[ac->finished_requests].module;
 220                 nreq = ac->part_req[ac->finished_requests].req;
 221
 222                 ret = partition_request(module, nreq);
 223                 if (ret != LDB_SUCCESS) {
 224                         talloc_free(ares);
 225                         return ldb_module_done(ac->req, NULL, NULL, ret);
 226                 }
 227
 228                 break;
 229         }
 230
 231         talloc_free(ares);
 232         return LDB_SUCCESS;
 233 }
 234
 235 static int partition_prep_request(struct partition_context *ac,
 236                                   struct dsdb_partition *partition)
 237 {
 238         int ret;
 239         struct ldb_request *req;
 240         struct ldb_control *partition_ctrl = NULL;
 241
 242         ac->part_req = talloc_realloc(ac, ac->part_req,
 243                                         struct part_request,
 244                                         ac->num_requests + 1);
 245         if (ac->part_req == NULL) {
 246                 return ldb_oom(ldb_module_get_ctx(ac->module));
 247         }
 248
 249         switch (ac->req->operation) {
 250         case LDB_SEARCH:
 251                 ret = ldb_build_search_req_ex(&req, ldb_module_get_ctx(ac->module),
 252                                         ac->part_req,
 253                                         ac->req->op.search.base,
 254                                         ac->req->op.search.scope,
 255                                         ac->req->op.search.tree,
 256                                         ac->req->op.search.attrs,
 257                                         ac->req->controls,
 258                                         ac, partition_req_callback,
 259                                         ac->req);
 260                 LDB_REQ_SET_LOCATION(req);
 261                 break;
 262         case LDB_ADD:
 263                 ret = ldb_build_add_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
 264                                         ac->req->op.add.message,
 265                                         ac->req->controls,
 266                                         ac, partition_req_callback,
 267                                         ac->req);
 268                 LDB_REQ_SET_LOCATION(req);
 269                 break;
 270         case LDB_MODIFY:
 271                 ret = ldb_build_mod_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
 272                                         ac->req->op.mod.message,
 273                                         ac->req->controls,
 274                                         ac, partition_req_callback,
 275                                         ac->req);
 276                 LDB_REQ_SET_LOCATION(req);
 277                 break;
 278         case LDB_DELETE:
 279                 ret = ldb_build_del_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
 280                                         ac->req->op.del.dn,
 281                                         ac->req->controls,
 282                                         ac, partition_req_callback,
 283                                         ac->req);
 284                 LDB_REQ_SET_LOCATION(req);
 285                 break;
 286         case LDB_RENAME:
 287                 ret = ldb_build_rename_req(&req, ldb_module_get_ctx(ac->module), 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                 LDB_REQ_SET_LOCATION(req);
 294                 break;
 295         case LDB_EXTENDED:
 296                 ret = ldb_build_extended_req(&req, ldb_module_get_ctx(ac->module),
 297                                         ac->part_req,
 298                                         ac->req->op.extended.oid,
 299                                         ac->req->op.extended.data,
 300                                         ac->req->controls,
 301                                         ac, partition_req_callback,
 302                                         ac->req);
 303                 LDB_REQ_SET_LOCATION(req);
 304                 break;
 305         default:
 306                 ldb_set_errstring(ldb_module_get_ctx(ac->module),
 307                                   "Unsupported request type!");
 308                 ret = LDB_ERR_UNWILLING_TO_PERFORM;
 309         }
 310
 311         if (ret != LDB_SUCCESS) {
 312                 return ret;
 313         }
 314
 315         ac->part_req[ac->num_requests].req = req;
 316
 317         if (ac->req->controls) {
 318                 /* Duplicate everything beside the current partition control */
 319                 partition_ctrl = ldb_request_get_control(ac->req,
 320                                                          DSDB_CONTROL_CURRENT_PARTITION_OID);
 321                 if (!ldb_save_controls(partition_ctrl, req, NULL)) {
 322                         return ldb_module_oom(ac->module);
 323                 }
 324         }
 325
 326         if (partition) {
 327                 void *part_data = partition->ctrl;
 328
 329                 ac->part_req[ac->num_requests].module = partition->module;
 330
 331                 if (partition_ctrl != NULL) {
 332                         if (partition_ctrl->data != NULL) {
 333                                 part_data = partition_ctrl->data;
 334                         }
 335
 336                         /*
 337                          * If the provided current partition control is without
 338                          * data then use the calculated one.
 339                          */
 340                         ret = ldb_request_add_control(req,
 341                                                       DSDB_CONTROL_CURRENT_PARTITION_OID,
 342                                                       false, part_data);
 343                         if (ret != LDB_SUCCESS) {
 344                                 return ret;
 345                         }
 346                 }
 347
 348                 if (req->operation == LDB_SEARCH) {
 349                         /* If the search is for 'more' than this partition,
 350                          * then change the basedn, so a remote LDAP server
 351                          * doesn't object */
 352                         if (ldb_dn_compare_base(partition->ctrl->dn,
 353                                                 req->op.search.base) != 0) {
 354                                 req->op.search.base = partition->ctrl->dn;
 355                         }
 356                 }
 357
 358         } else {
 359                 /* make sure you put the module here, or
 360                  * or ldb_next_request() will skip a module */
 361                 ac->part_req[ac->num_requests].module = ac->module;
 362         }
 363
 364         ac->num_requests++;
 365
 366         return LDB_SUCCESS;
 367 }
 368
 369 static int partition_call_first(struct partition_context *ac)
 370 {
 371         return partition_request(ac->part_req[0].module, ac->part_req[0].req);
 372 }
 373
 374 /**
 375  * Send a request down to all the partitions (but not the sam.ldb file)
 376  */
 377 static int partition_send_all(struct ldb_module *module,
 378                               struct partition_context *ac,
 379                               struct ldb_request *req)
 380 {
 381         unsigned int i;
 382         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 383                                                               struct partition_private_data);
 384         int ret;
 385
 386         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 387                 ret = partition_prep_request(ac, data->partitions[i]);
 388                 if (ret != LDB_SUCCESS) {
 389                         return ret;
 390                 }
 391         }
 392
 393         /* fire the first one */
 394         return partition_call_first(ac);
 395 }
 396
 397 struct partition_copy_context {
 398         struct ldb_module *module;
 399         struct partition_context *partition_context;
 400         struct ldb_request *request;
 401         struct ldb_dn *dn;
 402 };
 403
 404 /*
 405  * A special DN has been updated in the primary partition. Now propagate those
 406  * changes to the remaining partitions.
 407  *
 408  * Note: that the operations are asynchronous and this function is called
 409  *       from partition_copy_all_callback_handler in response to an async
 410  *       callback.
 411  */
 412 static int partition_copy_all_callback_action(
 413         struct ldb_module *module,
 414         struct partition_context *ac,
 415         struct ldb_request *req,
 416         struct ldb_dn *dn)
 417
 418 {
 419
 420         unsigned int i;
 421         struct partition_private_data *data =
 422                 talloc_get_type(
 423                         ldb_module_get_private(module),
 424                         struct partition_private_data);
 425         int search_ret;
 426         struct ldb_result *res;
 427         /* now fetch the resulting object, and then copy it to all the
 428          * other partitions. We need this approach to cope with the
 429          * partitions getting out of sync. If for example the
 430          * @ATTRIBUTES object exists on one partition but not the
 431          * others then just doing each of the partitions in turn will
 432          * lead to an error
 433          */
 434         search_ret = dsdb_module_search_dn(module, ac, &res, dn, NULL, DSDB_FLAG_NEXT_MODULE, req);
 435         if (search_ret != LDB_SUCCESS) {
 436                 return search_ret;
 437         }
 438
 439         /* now delete the object in the other partitions, if requried
 440         */
 441         if (search_ret == LDB_ERR_NO_SUCH_OBJECT) {
 442                 for (i=0; data->partitions && data->partitions[i]; i++) {
 443                         int pret;
 444                         pret = dsdb_module_del(data->partitions[i]->module,
 445                                                dn,
 446                                                DSDB_FLAG_NEXT_MODULE,
 447                                                req);
 448                         if (pret != LDB_SUCCESS && pret != LDB_ERR_NO_SUCH_OBJECT) {
 449                                 /* we should only get success or no
 450                                    such object from the other partitions */
 451                                 return pret;
 452                         }
 453                 }
 454
 455                 return ldb_module_done(req, NULL, NULL, LDB_SUCCESS);
 456         }
 457
 458         /* now add/modify in the other partitions */
 459         for (i=0; data->partitions && data->partitions[i]; i++) {
 460                 struct ldb_message *modify_msg = NULL;
 461                 int pret;
 462                 unsigned int el_idx;
 463
 464                 pret = dsdb_module_add(data->partitions[i]->module,
 465                                        res->msgs[0],
 466                                        DSDB_FLAG_NEXT_MODULE,
 467                                        req);
 468                 if (pret == LDB_SUCCESS) {
 469                         continue;
 470                 }
 471
 472                 if (pret != LDB_ERR_ENTRY_ALREADY_EXISTS) {
 473                         return pret;
 474                 }
 475
 476                 modify_msg = ldb_msg_copy(req, res->msgs[0]);
 477                 if (modify_msg == NULL) {
 478                         return ldb_module_oom(module);
 479                 }
 480
 481                 /*
 482                  * mark all the message elements as
 483                  * LDB_FLAG_MOD_REPLACE
 484                  */
 485                 for (el_idx=0;
 486                      el_idx < modify_msg->num_elements;
 487                      el_idx++) {
 488                         modify_msg->elements[el_idx].flags
 489                                 = LDB_FLAG_MOD_REPLACE;
 490                 }
 491
 492                 if (req->operation == LDB_MODIFY) {
 493                         const struct ldb_message *req_msg = req->op.mod.message;
 494                         /*
 495                          * mark elements to be removed, if there were
 496                          * deleted entirely above we need to delete
 497                          * them here too
 498                          */
 499                         for (el_idx=0; el_idx < req_msg->num_elements; el_idx++) {
 500                                 if (req_msg->elements[el_idx].flags & LDB_FLAG_MOD_DELETE
 501                                     || ((req_msg->elements[el_idx].flags & LDB_FLAG_MOD_REPLACE) &&
 502                                         req_msg->elements[el_idx].num_values == 0)) {
 503                                         if (ldb_msg_find_element(modify_msg,
 504                                                                  req_msg->elements[el_idx].name) != NULL) {
 505                                                 continue;
 506                                         }
 507                                         pret = ldb_msg_add_empty(
 508                                                 modify_msg,
 509                                                 req_msg->elements[el_idx].name,
 510                                                 LDB_FLAG_MOD_REPLACE,
 511                                                 NULL);
 512                                         if (pret != LDB_SUCCESS) {
 513                                                 return pret;
 514                                         }
 515                                 }
 516                         }
 517                 }
 518
 519                 pret = dsdb_module_modify(data->partitions[i]->module,
 520                                           modify_msg,
 521                                           DSDB_FLAG_NEXT_MODULE,
 522                                           req);
 523
 524                 if (pret != LDB_SUCCESS) {
 525                         return pret;
 526                 }
 527         }
 528
 529         return ldb_module_done(req, NULL, NULL, LDB_SUCCESS);
 530 }
 531
 532
 533 /*
 534  * @brief call back function for the ldb operations on special DN's.
 535  *
 536  * As the LDB operations are async, and we wish to use the result
 537  * the operations, a callback needs to be registered to process the results
 538  * of the LDB operations.
 539  *
 540  * @param req the ldb request
 541  * @param res the result of the operation
 542  *
 543  * @return the LDB_STATUS
 544  */
 545 static int partition_copy_all_callback_handler(
 546         struct ldb_request *req,
 547         struct ldb_reply *ares)
 548 {
 549         struct partition_copy_context *ac = NULL;
 550
 551         ac = talloc_get_type(
 552                 req->context,
 553                 struct partition_copy_context);
 554
 555         if (!ares) {
 556                 return ldb_module_done(
 557                         ac->request,
 558                         NULL,
 559                         NULL,
 560                         LDB_ERR_OPERATIONS_ERROR);
 561         }
 562
 563         /* pass on to the callback */
 564         switch (ares->type) {
 565         case LDB_REPLY_ENTRY:
 566                 return ldb_module_send_entry(
 567                         ac->request,
 568                         ares->message,
 569                         ares->controls);
 570
 571         case LDB_REPLY_REFERRAL:
 572                 return ldb_module_send_referral(
 573                         ac->request,
 574                         ares->referral);
 575
 576         case LDB_REPLY_DONE: {
 577                 int error = ares->error;
 578                 if (error == LDB_SUCCESS) {
 579                         error = partition_copy_all_callback_action(
 580                                 ac->module,
 581                                 ac->partition_context,
 582                                 ac->request,
 583                                 ac->dn);
 584                 }
 585                 return ldb_module_done(
 586                         ac->request,
 587                         ares->controls,
 588                         ares->response,
 589                         error);
 590         }
 591
 592         default:
 593                 /* Can't happen */
 594                 return LDB_ERR_OPERATIONS_ERROR;
 595         }
 596 }
 597
 598 /**
 599  * send an operation to the top partition, then copy the resulting
 600  * object to all other partitions.
 601  */
 602 static int partition_copy_all(
 603         struct ldb_module *module,
 604         struct partition_context *partition_context,
 605         struct ldb_request *req,
 606         struct ldb_dn *dn)
 607 {
 608         struct ldb_request *new_req = NULL;
 609         struct ldb_context *ldb = NULL;
 610         struct partition_copy_context *context = NULL;
 611
 612         int ret;
 613
 614         ldb = ldb_module_get_ctx(module);
 615
 616         context = talloc_zero(req, struct partition_copy_context);
 617         if (context == NULL) {
 618                 return ldb_oom(ldb);
 619         }
 620         context->module = module;
 621         context->request = req;
 622         context->dn = dn;
 623         context->partition_context = partition_context;
 624
 625         switch (req->operation) {
 626         case LDB_ADD:
 627                 ret = ldb_build_add_req(
 628                         &new_req,
 629                         ldb,
 630                         req,
 631                         req->op.add.message,
 632                         req->controls,
 633                         context,
 634                         partition_copy_all_callback_handler,
 635                         req);
 636                 break;
 637         case LDB_MODIFY:
 638                 ret = ldb_build_mod_req(
 639                         &new_req,
 640                         ldb,
 641                         req,
 642                         req->op.mod.message,
 643                         req->controls,
 644                         context,
 645                         partition_copy_all_callback_handler,
 646                         req);
 647                 break;
 648         case LDB_DELETE:
 649                 ret = ldb_build_del_req(
 650                         &new_req,
 651                         ldb,
 652                         req,
 653                         req->op.del.dn,
 654                         req->controls,
 655                         context,
 656                         partition_copy_all_callback_handler,
 657                         req);
 658                 break;
 659         case LDB_RENAME:
 660                 ret = ldb_build_rename_req(
 661                         &new_req,
 662                         ldb,
 663                         req,
 664                         req->op.rename.olddn,
 665                         req->op.rename.newdn,
 666                         req->controls,
 667                         context,
 668                         partition_copy_all_callback_handler,
 669                         req);
 670                 break;
 671         default:
 672                 /*
 673                  * Shouldn't happen.
 674                  */
 675                 ldb_debug(
 676                         ldb,
 677                         LDB_DEBUG_ERROR,
 678                         "Unexpected operation type (%d)\n", req->operation);
 679                 ret = LDB_ERR_OPERATIONS_ERROR;
 680                 break;
 681         }
 682         if (ret != LDB_SUCCESS) {
 683                 return ret;
 684         }
 685         return ldb_next_request(module, new_req);
 686 }
 687 /**
 688  * Figure out which backend a request needs to be aimed at.  Some
 689  * requests must be replicated to all backends
 690  */
 691 static int partition_replicate(struct ldb_module *module, struct ldb_request *req, struct ldb_dn *dn)
 692 {
 693         struct partition_context *ac;
 694         unsigned int i;
 695         int ret;
 696         struct dsdb_partition *partition;
 697         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 698                                                               struct partition_private_data);
 699
 700         /* if we aren't initialised yet go further */
 701         if (!data || !data->partitions) {
 702                 return ldb_next_request(module, req);
 703         }
 704
 705         if (ldb_dn_is_special(dn)) {
 706                 /* Is this a special DN, we need to replicate to every backend? */
 707                 for (i=0; data->replicate && data->replicate[i]; i++) {
 708                         if (ldb_dn_compare(data->replicate[i],
 709                                            dn) == 0) {
 710
 711                                 ac = partition_init_ctx(module, req);
 712                                 if (!ac) {
 713                                         return ldb_operr(ldb_module_get_ctx(module));
 714                                 }
 715
 716                                 return partition_copy_all(module, ac, req, dn);
 717                         }
 718                 }
 719         }
 720
 721         /* Otherwise, we need to find the partition to fire it to */
 722
 723         /* Find partition */
 724         partition = find_partition(data, dn, req);
 725         if (!partition) {
 726                 /*
 727                  * if we haven't found a matching partition
 728                  * pass the request to the main ldb
 729                  *
 730                  * TODO: we should maybe return an error here
 731                  *       if it's not a special dn
 732                  */
 733
 734                 return ldb_next_request(module, req);
 735         }
 736
 737         ac = partition_init_ctx(module, req);
 738         if (!ac) {
 739                 return ldb_operr(ldb_module_get_ctx(module));
 740         }
 741
 742         /* we need to add a control but we never touch the original request */
 743         ret = partition_prep_request(ac, partition);
 744         if (ret != LDB_SUCCESS) {
 745                 return ret;
 746         }
 747
 748         /* fire the first one */
 749         return partition_call_first(ac);
 750 }
 751
 752 /* search */
 753 static int partition_search(struct ldb_module *module, struct ldb_request *req)
 754 {
 755         struct ldb_control **saved_controls;
 756         /* Find backend */
 757         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 758                                                               struct partition_private_data);
 759         struct partition_context *ac;
 760         struct ldb_context *ldb;
 761         struct loadparm_context *lp_ctx;
 762
 763         struct ldb_control *search_control = ldb_request_get_control(req, LDB_CONTROL_SEARCH_OPTIONS_OID);
 764         struct ldb_control *domain_scope_control = ldb_request_get_control(req, LDB_CONTROL_DOMAIN_SCOPE_OID);
 765         struct ldb_control *no_gc_control = ldb_request_get_control(req, DSDB_CONTROL_NO_GLOBAL_CATALOG);
 766
 767         struct ldb_search_options_control *search_options = NULL;
 768         struct dsdb_partition *p;
 769         unsigned int i, j;
 770         int ret;
 771         bool domain_scope = false, phantom_root = false;
 772
 773         p = find_partition(data, NULL, req);
 774         if (p != NULL) {
 775                 /* the caller specified what partition they want the
 776                  * search - just pass it on
 777                  */
 778                 return ldb_next_request(p->module, req);
 779         }
 780
 781         /* Get back the search options from the search control, and mark it as
 782          * non-critical (to make backends and also dcpromo happy).
 783          */
 784         if (search_control) {
 785                 search_options = talloc_get_type(search_control->data, struct ldb_search_options_control);
 786                 search_control->critical = 0;
 787
 788         }
 789
 790         /* Remove the "domain_scope" control, so we don't confuse a backend
 791          * server */
 792         if (domain_scope_control && !ldb_save_controls(domain_scope_control, req, &saved_controls)) {
 793                 return ldb_oom(ldb_module_get_ctx(module));
 794         }
 795
 796         /* if we aren't initialised yet go further */
 797         if (!data || !data->partitions) {
 798                 return ldb_next_request(module, req);
 799         }
 800
 801         /* Special DNs without specified partition should go further */
 802         if (ldb_dn_is_special(req->op.search.base)) {
 803                 return ldb_next_request(module, req);
 804         }
 805
 806         /* Locate the options */
 807         domain_scope = (search_options
 808                 && (search_options->search_options & LDB_SEARCH_OPTION_DOMAIN_SCOPE))
 809                 || domain_scope_control;
 810         phantom_root = search_options
 811                 && (search_options->search_options & LDB_SEARCH_OPTION_PHANTOM_ROOT);
 812
 813         /* Remove handled options from the search control flag */
 814         if (search_options) {
 815                 search_options->search_options = search_options->search_options
 816                         & ~LDB_SEARCH_OPTION_DOMAIN_SCOPE
 817                         & ~LDB_SEARCH_OPTION_PHANTOM_ROOT;
 818         }
 819
 820         ac = partition_init_ctx(module, req);
 821         if (!ac) {
 822                 return ldb_operr(ldb_module_get_ctx(module));
 823         }
 824
 825         ldb = ldb_module_get_ctx(ac->module);
 826         lp_ctx = talloc_get_type(ldb_get_opaque(ldb, "loadparm"),
 827                                                 struct loadparm_context);
 828
 829         /* Search from the base DN */
 830         if (ldb_dn_is_null(req->op.search.base)) {
 831                 if (!phantom_root) {
 832                         return ldb_error(ldb, LDB_ERR_NO_SUCH_OBJECT, "empty base DN");
 833                 }
 834                 return partition_send_all(module, ac, req);
 835         }
 836
 837         for (i=0; data->partitions[i]; i++) {
 838                 bool match = false, stop = false;
 839
 840                 if (data->partitions[i]->partial_replica && no_gc_control != NULL) {
 841                         if (ldb_dn_compare_base(data->partitions[i]->ctrl->dn,
 842                                                 req->op.search.base) == 0) {
 843                                 /* base DN is in a partial replica
 844                                    with the NO_GLOBAL_CATALOG
 845                                    control. This partition is invisible */
 846                                 /* DEBUG(0,("DENYING NON-GC OP: %s\n", ldb_module_call_chain(req, req))); */
 847                                 continue;
 848                         }
 849                 }
 850
 851                 if (phantom_root) {
 852                         /* Phantom root: Find all partitions under the
 853                          * search base. We match if:
 854                          *
 855                          * 1) the DN we are looking for exactly matches a
 856                          *    certain partition and always stop
 857                          * 2) the DN we are looking for is a parent of certain
 858                          *    partitions and it isn't a scope base search
 859                          * 3) the DN we are looking for is a child of a certain
 860                          *    partition and always stop
 861                          *    - we don't need to go any further up in the
 862                          *    hierarchy!
 863                          */
 864                         if (ldb_dn_compare(data->partitions[i]->ctrl->dn,
 865                                            req->op.search.base) == 0) {
 866                                 match = true;
 867                                 stop = true;
 868                         }
 869                         if (!match &&
 870                             (ldb_dn_compare_base(req->op.search.base,
 871                                                  data->partitions[i]->ctrl->dn) == 0 &&
 872                              req->op.search.scope != LDB_SCOPE_BASE)) {
 873                                 match = true;
 874                         }
 875                         if (!match &&
 876                             ldb_dn_compare_base(data->partitions[i]->ctrl->dn,
 877                                                 req->op.search.base) == 0) {
 878                                 match = true;
 879                                 stop = true; /* note that this relies on partition ordering */
 880                         }
 881                 } else {
 882                         /* Domain scope: Find all partitions under the search
 883                          * base.
 884                          *
 885                          * We generate referral candidates if we haven't
 886                          * specified the domain scope control, haven't a base
 887                          * search* scope and the DN we are looking for is a real
 888                          * predecessor of certain partitions. When a new
 889                          * referral candidate is nearer to the DN than an
 890                          * existing one delete the latter (we want to have only
 891                          * the closest ones). When we checked this for all
 892                          * candidates we have the final referrals.
 893                          *
 894                          * We match if the DN we are looking for is a child of
 895                          * a certain partition or the partition
 896                          * DN itself - we don't need to go any further
 897                          * up in the hierarchy!
 898                          */
 899                         if ((!domain_scope) &&
 900                             (req->op.search.scope != LDB_SCOPE_BASE) &&
 901                             (ldb_dn_compare_base(req->op.search.base,
 902                                                  data->partitions[i]->ctrl->dn) == 0) &&
 903                             (ldb_dn_compare(req->op.search.base,
 904                                             data->partitions[i]->ctrl->dn) != 0)) {
 905                                 char *ref = talloc_asprintf(ac,
 906                                                             "ldap://%s/%s%s",
 907                                                             lpcfg_dnsdomain(lp_ctx),
 908                                                             ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
 909                                                             req->op.search.scope == LDB_SCOPE_ONELEVEL ? "??base" : "");
 910
 911                                 if (ref == NULL) {
 912                                         return ldb_oom(ldb);
 913                                 }
 914
 915                                 /* Initialise the referrals list */
 916                                 if (ac->referrals == NULL) {
 917                                         char **l = str_list_make_empty(ac);
 918                                         ac->referrals = discard_const_p(const char *, l);
 919                                         if (ac->referrals == NULL) {
 920                                                 return ldb_oom(ldb);
 921                                         }
 922                                 }
 923
 924                                 /* Check if the new referral candidate is
 925                                  * closer to the base DN than already
 926                                  * saved ones and delete the latters */
 927                                 j = 0;
 928                                 while (ac->referrals[j] != NULL) {
 929                                         if (strstr(ac->referrals[j],
 930                                                    ldb_dn_get_linearized(data->partitions[i]->ctrl->dn)) != NULL) {
 931                                                 str_list_remove(ac->referrals,
 932                                                                 ac->referrals[j]);
 933                                         } else {
 934                                                 ++j;
 935                                         }
 936                                 }
 937
 938                                 /* Add our new candidate */
 939                                 ac->referrals = str_list_add(ac->referrals, ref);
 940
 941                                 talloc_free(ref);
 942
 943                                 if (ac->referrals == NULL) {
 944                                         return ldb_oom(ldb);
 945                                 }
 946                         }
 947                         if (ldb_dn_compare_base(data->partitions[i]->ctrl->dn, req->op.search.base) == 0) {
 948                                 match = true;
 949                                 stop = true; /* note that this relies on partition ordering */
 950                         }
 951                 }
 952
 953                 if (match) {
 954                         ret = partition_prep_request(ac, data->partitions[i]);
 955                         if (ret != LDB_SUCCESS) {
 956                                 return ret;
 957                         }
 958                 }
 959
 960                 if (stop) break;
 961         }
 962
 963         /* Perhaps we didn't match any partitions. Try the main partition */
 964         if (ac->num_requests == 0) {
 965                 talloc_free(ac);
 966                 return ldb_next_request(module, req);
 967         }
 968
 969         /* fire the first one */
 970         return partition_call_first(ac);
 971 }
 972
 973 /* add */
 974 static int partition_add(struct ldb_module *module, struct ldb_request *req)
 975 {
 976         return partition_replicate(module, req, req->op.add.message->dn);
 977 }
 978
 979 /* modify */
 980 static int partition_modify(struct ldb_module *module, struct ldb_request *req)
 981 {
 982         return partition_replicate(module, req, req->op.mod.message->dn);
 983 }
 984
 985 /* delete */
 986 static int partition_delete(struct ldb_module *module, struct ldb_request *req)
 987 {
 988         return partition_replicate(module, req, req->op.del.dn);
 989 }
 990
 991 /* rename */
 992 static int partition_rename(struct ldb_module *module, struct ldb_request *req)
 993 {
 994         /* Find backend */
 995         struct dsdb_partition *backend, *backend2;
 996
 997         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 998                                                               struct partition_private_data);
 999
1000         /* Skip the lot if 'data' isn't here yet (initialisation) */
1001         if (!data) {
1002                 return ldb_operr(ldb_module_get_ctx(module));
1003         }
1004
1005         backend = find_partition(data, req->op.rename.olddn, req);
1006         backend2 = find_partition(data, req->op.rename.newdn, req);
1007
1008         if ((backend && !backend2) || (!backend && backend2)) {
1009                 return LDB_ERR_AFFECTS_MULTIPLE_DSAS;
1010         }
1011
1012         if (backend != backend2) {
1013                 ldb_asprintf_errstring(ldb_module_get_ctx(module),
1014                                        "Cannot rename from %s in %s to %s in %s: %s",
1015                                        ldb_dn_get_linearized(req->op.rename.olddn),
1016                                        ldb_dn_get_linearized(backend->ctrl->dn),
1017                                        ldb_dn_get_linearized(req->op.rename.newdn),
1018                                        ldb_dn_get_linearized(backend2->ctrl->dn),
1019                                        ldb_strerror(LDB_ERR_AFFECTS_MULTIPLE_DSAS));
1020                 return LDB_ERR_AFFECTS_MULTIPLE_DSAS;
1021         }
1022
1023         return partition_replicate(module, req, req->op.rename.olddn);
1024 }
1025
1026 /* start a transaction */
1027 int partition_start_trans(struct ldb_module *module)
1028 {
1029         int i;
1030         int ret;
1031         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
1032                                                               struct partition_private_data);
1033         /* Look at base DN */
1034         /* Figure out which partition it is under */
1035         /* Skip the lot if 'data' isn't here yet (initialization) */
1036         if (ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING) {
1037                 ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_start_trans() -> (metadata partition)");
1038         }
1039
1040         /* This order must match that in prepare_commit() and read_lock() */
1041         ret = ldb_next_start_trans(module);
1042         if (ret != LDB_SUCCESS) {
1043                 return ret;
1044         }
1045
1046         ret = partition_reload_if_required(module, data, NULL);
1047         if (ret != LDB_SUCCESS) {
1048                 ldb_next_del_trans(module);
1049                 return ret;
1050         }
1051
1052         for (i=0; data && data->partitions && data->partitions[i]; i++) {
1053                 if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
1054                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_start_trans() -> %s",
1055                                   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
1056                 }
1057                 ret = ldb_next_start_trans(data->partitions[i]->module);
1058                 if (ret != LDB_SUCCESS) {
1059                         /* Back it out, if it fails on one */
1060                         for (i--; i >= 0; i--) {
1061                                 ldb_next_del_trans(data->partitions[i]->module);
1062                         }
1063                         ldb_next_del_trans(module);
1064                         partition_metadata_del_trans(module);
1065                         return ret;
1066                 }
1067         }
1068
1069         /*
1070          * Because in prepare_commit this must come last, to ensure
1071          * lock ordering we have to do this last here also
1072          */
1073         ret = partition_metadata_start_trans(module);
1074         if (ret != LDB_SUCCESS) {
1075                 /* Back it out, if it fails on one */
1076                 for (i--; i >= 0; i--) {
1077                         ldb_next_del_trans(data->partitions[i]->module);
1078                 }
1079                 ldb_next_del_trans(module);
1080                 return ret;
1081         }
1082
1083         data->in_transaction++;
1084
1085         return LDB_SUCCESS;
1086 }
1087
1088 /* prepare for a commit */
1089 int partition_prepare_commit(struct ldb_module *module)
1090 {
1091         unsigned int i;
1092         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
1093                                                               struct partition_private_data);
1094         int ret;
1095
1096         ret = ldb_next_prepare_commit(module);
1097         if (ret != LDB_SUCCESS) {
1098                 return ret;
1099         }
1100
1101         for (i=0; data && data->partitions && data->partitions[i]; i++) {
1102                 if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
1103                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_prepare_commit() -> %s",
1104                                   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
1105                 }
1106                 ret = ldb_next_prepare_commit(data->partitions[i]->module);
1107                 if (ret != LDB_SUCCESS) {
1108                         ldb_asprintf_errstring(ldb_module_get_ctx(module), "prepare_commit error on %s: %s",
1109                                                ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
1110                                                ldb_errstring(ldb_module_get_ctx(module)));
1111                         return ret;
1112                 }
1113         }
1114
1115         if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
1116                 ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_prepare_commit() -> (metadata partition)");
1117         }
1118
1119         /* metadata prepare commit must come last, as other partitions could modify
1120          * the database inside the prepare commit method of a module */
1121         return partition_metadata_prepare_commit(module);
1122 }
1123
1124
1125 /* end a transaction */
1126 int partition_end_trans(struct ldb_module *module)
1127 {
1128         int ret, ret2;
1129         unsigned int i;
1130         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
1131                                                               struct partition_private_data);
1132
1133         ret = LDB_SUCCESS;
1134
1135         if (data->in_transaction == 0) {
1136                 DEBUG(0,("partition end transaction mismatch\n"));
1137                 ret = LDB_ERR_OPERATIONS_ERROR;
1138         } else {
1139                 data->in_transaction--;
1140         }
1141
1142
1143         for (i=0; data && data->partitions && data->partitions[i]; i++) {
1144                 if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
1145                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_end_trans() -> %s",
1146                                   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
1147                 }
1148                 ret2 = ldb_next_end_trans(data->partitions[i]->module);
1149                 if (ret2 != LDB_SUCCESS) {
1150                         ldb_asprintf_errstring(ldb_module_get_ctx(module), "end_trans error on %s: %s",
1151                                                ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
1152                                                ldb_errstring(ldb_module_get_ctx(module)));
1153                         ret = ret2;
1154                 }
1155         }
1156
1157         if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
1158                 ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_end_trans() -> (metadata partition)");
1159         }
1160         ret2 = ldb_next_end_trans(module);
1161         if (ret2 != LDB_SUCCESS) {
1162                 ret = ret2;
1163         }
1164
1165         ret2 = partition_metadata_end_trans(module);
1166         if (ret2 != LDB_SUCCESS) {
1167                 ret = ret2;
1168         }
1169
1170         return ret;
1171 }
1172
1173 /* delete a transaction */
1174 int partition_del_trans(struct ldb_module *module)
1175 {
1176         int ret, final_ret = LDB_SUCCESS;
1177         unsigned int i;
1178         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
1179                                                               struct partition_private_data);
1180
1181         for (i=0; data && data->partitions && data->partitions[i]; i++) {
1182                 if (ldb_module_flags(ldb_module_get_ctx(module)) &
1183                     LDB_FLG_ENABLE_TRACING) {
1184                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_del_trans() -> %s",
1185                                   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
1186                 }
1187                 ret = ldb_next_del_trans(data->partitions[i]->module);
1188                 if (ret != LDB_SUCCESS) {
1189                         ldb_asprintf_errstring(ldb_module_get_ctx(module), "del_trans error on %s: %s",
1190                                                ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
1191                                                ldb_errstring(ldb_module_get_ctx(module)));
1192                         final_ret = ret;
1193                 }
1194         }
1195
1196         if (data->in_transaction == 0) {
1197                 DEBUG(0,("partition del transaction mismatch\n"));
1198                 return ldb_operr(ldb_module_get_ctx(module));
1199         }
1200         data->in_transaction--;
1201
1202         if (ldb_module_flags(ldb_module_get_ctx(module)) &
1203             LDB_FLG_ENABLE_TRACING) {
1204                 ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_del_trans() -> (metadata partition)");
1205         }
1206         ret = ldb_next_del_trans(module);
1207         if (ret != LDB_SUCCESS) {
1208                 final_ret = ret;
1209         }
1210
1211         ret = partition_metadata_del_trans(module);
1212         if (ret != LDB_SUCCESS) {
1213                 final_ret = ret;
1214         }
1215
1216         return final_ret;
1217 }
1218
1219 int partition_primary_sequence_number(struct ldb_module *module, TALLOC_CTX *mem_ctx,
1220                                       uint64_t *seq_number,
1221                                       struct ldb_request *parent)
1222 {
1223         int ret;
1224         struct ldb_result *res;
1225         struct ldb_seqnum_request *tseq;
1226         struct ldb_seqnum_result *seqr;
1227
1228         tseq = talloc_zero(mem_ctx, struct ldb_seqnum_request);
1229         if (tseq == NULL) {
1230                 return ldb_oom(ldb_module_get_ctx(module));
1231         }
1232         tseq->type = LDB_SEQ_HIGHEST_SEQ;
1233
1234         ret = dsdb_module_extended(module, tseq, &res,
1235                                    LDB_EXTENDED_SEQUENCE_NUMBER,
1236                                    tseq,
1237                                    DSDB_FLAG_NEXT_MODULE,
1238                                    parent);
1239         if (ret != LDB_SUCCESS) {
1240                 talloc_free(tseq);
1241                 return ret;
1242         }
1243
1244         seqr = talloc_get_type_abort(res->extended->data,
1245                                      struct ldb_seqnum_result);
1246         if (seqr->flags & LDB_SEQ_TIMESTAMP_SEQUENCE) {
1247                 talloc_free(res);
1248                 return ldb_module_error(module, LDB_ERR_OPERATIONS_ERROR,
1249                         "Primary backend in partition module returned a timestamp based seq");
1250         }
1251
1252         *seq_number = seqr->seq_num;
1253         talloc_free(tseq);
1254         return LDB_SUCCESS;
1255 }
1256
1257
1258 /*
1259  * Older version of sequence number as sum of sequence numbers for each partition
1260  */
1261 int partition_sequence_number_from_partitions(struct ldb_module *module,
1262                                               uint64_t *seqr)
1263 {
1264         int ret;
1265         unsigned int i;
1266         uint64_t seq_number = 0;
1267         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
1268                                                               struct partition_private_data);
1269
1270         ret = partition_primary_sequence_number(module, data, &seq_number, NULL);
1271         if (ret != LDB_SUCCESS) {
1272                 return ret;
1273         }
1274
1275         /* Skip the lot if 'data' isn't here yet (initialisation) */
1276         for (i=0; data && data->partitions && data->partitions[i]; i++) {
1277                 struct ldb_seqnum_request *tseq;
1278                 struct ldb_seqnum_result *tseqr;
1279                 struct ldb_request *treq;
1280                 struct ldb_result *res = talloc_zero(data, struct ldb_result);
1281                 if (res == NULL) {
1282                         return ldb_oom(ldb_module_get_ctx(module));
1283                 }
1284                 tseq = talloc_zero(res, struct ldb_seqnum_request);
1285                 if (tseq == NULL) {
1286                         talloc_free(res);
1287                         return ldb_oom(ldb_module_get_ctx(module));
1288                 }
1289                 tseq->type = LDB_SEQ_HIGHEST_SEQ;
1290
1291                 ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
1292                                              LDB_EXTENDED_SEQUENCE_NUMBER,
1293                                              tseq,
1294                                              NULL,
1295                                              res,
1296                                              ldb_extended_default_callback,
1297                                              NULL);
1298                 LDB_REQ_SET_LOCATION(treq);
1299                 if (ret != LDB_SUCCESS) {
1300                         talloc_free(res);
1301                         return ret;
1302                 }
1303
1304                 ret = partition_request(data->partitions[i]->module, treq);
1305                 if (ret != LDB_SUCCESS) {
1306                         talloc_free(res);
1307                         return ret;
1308                 }
1309                 ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
1310                 if (ret != LDB_SUCCESS) {
1311                         talloc_free(res);
1312                         return ret;
1313                 }
1314                 tseqr = talloc_get_type(res->extended->data,
1315                                         struct ldb_seqnum_result);
1316                 seq_number += tseqr->seq_num;
1317                 talloc_free(res);
1318         }
1319
1320         *seqr = seq_number;
1321         return LDB_SUCCESS;
1322 }
1323
1324
1325 /*
1326  * Newer version of sequence number using metadata tdb
1327  */
1328 static int partition_sequence_number(struct ldb_module *module, struct ldb_request *req)
1329 {
1330         struct ldb_extended *ext;
1331         struct ldb_seqnum_request *seq;
1332         struct ldb_seqnum_result *seqr;
1333         uint64_t seq_number;
1334         int ret;
1335
1336         seq = talloc_get_type_abort(req->op.extended.data, struct ldb_seqnum_request);
1337         switch (seq->type) {
1338         case LDB_SEQ_NEXT:
1339                 ret = partition_metadata_sequence_number_increment(module, &seq_number);
1340                 if (ret != LDB_SUCCESS) {
1341                         return ret;
1342                 }
1343                 break;
1344
1345         case LDB_SEQ_HIGHEST_SEQ:
1346                 ret = partition_metadata_sequence_number(module, &seq_number);
1347                 if (ret != LDB_SUCCESS) {
1348                         return ret;
1349                 }
1350                 break;
1351
1352         case LDB_SEQ_HIGHEST_TIMESTAMP:
1353                 return ldb_module_error(module, LDB_ERR_OPERATIONS_ERROR,
1354                                         "LDB_SEQ_HIGHEST_TIMESTAMP not supported");
1355         }
1356
1357         ext = talloc_zero(req, struct ldb_extended);
1358         if (!ext) {
1359                 return ldb_module_oom(module);
1360         }
1361         seqr = talloc_zero(ext, struct ldb_seqnum_result);
1362         if (seqr == NULL) {
1363                 talloc_free(ext);
1364                 return ldb_module_oom(module);
1365         }
1366         ext->oid = LDB_EXTENDED_SEQUENCE_NUMBER;
1367         ext->data = seqr;
1368
1369         seqr->seq_num = seq_number;
1370         seqr->flags |= LDB_SEQ_GLOBAL_SEQUENCE;
1371
1372         /* send request done */
1373         return ldb_module_done(req, NULL, ext, LDB_SUCCESS);
1374 }
1375
1376 /* lock all the backends */
1377 int partition_read_lock(struct ldb_module *module)
1378 {
1379         int i;
1380         int ret;
1381         int ret2;
1382         struct ldb_context *ldb = ldb_module_get_ctx(module);
1383         struct partition_private_data *data = \
1384                 talloc_get_type(ldb_module_get_private(module),
1385                                 struct partition_private_data);
1386
1387         if (ldb_module_flags(ldb) & LDB_FLG_ENABLE_TRACING) {
1388                 ldb_debug(ldb, LDB_DEBUG_TRACE,
1389                           "partition_read_lock() -> (metadata partition)");
1390         }
1391
1392         /*
1393          * It is important to only do this for LOCK because:
1394          * - we don't want to unlock what we did not lock
1395          *
1396          * - we don't want to make a new lock on the sam.ldb
1397          *   (triggered inside this routine due to the seq num check)
1398          *   during an unlock phase as that will violate the lock
1399          *   ordering
1400          */
1401
1402         if (data == NULL) {
1403                 TALLOC_CTX *mem_ctx = talloc_new(module);
1404
1405                 data = talloc_zero(mem_ctx, struct partition_private_data);
1406                 if (data == NULL) {
1407                         talloc_free(mem_ctx);
1408                         return ldb_operr(ldb);
1409                 }
1410
1411                 /*
1412                  * When used from Samba4, this message is set by the
1413                  * samba4 module, as a fixed value not read from the
1414                  * DB.  This avoids listing modules in the DB
1415                  */
1416                 data->forced_module_msg = talloc_get_type(
1417                         ldb_get_opaque(ldb,
1418                                        DSDB_OPAQUE_PARTITION_MODULE_MSG_OPAQUE_NAME),
1419                         struct ldb_message);
1420
1421                 ldb_module_set_private(module, talloc_steal(module,
1422                                                             data));
1423                 talloc_free(mem_ctx);
1424         }
1425
1426         /*
1427          * This will lock the metadata partition (sam.ldb) and
1428          * will also call event loops, so we do it before we
1429          * get the whole db lock.
1430          */
1431         ret = partition_reload_if_required(module, data, NULL);
1432         if (ret != LDB_SUCCESS) {
1433                 return ret;
1434         }
1435
1436         /*
1437          * This order must match that in prepare_commit(), start with
1438          * the metadata partition (sam.ldb) lock
1439          */
1440         ret = ldb_next_read_lock(module);
1441         if (ret != LDB_SUCCESS) {
1442                 ldb_debug_set(ldb,
1443                               LDB_DEBUG_FATAL,
1444                               "Failed to lock db: %s / %s for metadata partition",
1445                               ldb_errstring(ldb),
1446                               ldb_strerror(ret));
1447
1448                 return ret;
1449         }
1450
1451         /*
1452          * The metadata partition (sam.ldb) lock is not
1453          * enough to block another process in prepare_commit(),
1454          * because prepare_commit() is a no-op, if nothing
1455          * was changed in the specific backend.
1456          *
1457          * That means the following per partition locks are required.
1458          */
1459         for (i=0; data && data->partitions && data->partitions[i]; i++) {
1460                 if ((module && ldb_module_flags(ldb) & LDB_FLG_ENABLE_TRACING)) {
1461                         ldb_debug(ldb, LDB_DEBUG_TRACE,
1462                                   "partition_read_lock() -> %s",
1463                                   ldb_dn_get_linearized(
1464                                           data->partitions[i]->ctrl->dn));
1465                 }
1466                 ret = ldb_next_read_lock(data->partitions[i]->module);
1467                 if (ret == LDB_SUCCESS) {
1468                         continue;
1469                 }
1470
1471                 ldb_debug_set(ldb,
1472                               LDB_DEBUG_FATAL,
1473                               "Failed to lock db: %s / %s for %s",
1474                               ldb_errstring(ldb),
1475                               ldb_strerror(ret),
1476                               ldb_dn_get_linearized(
1477                                       data->partitions[i]->ctrl->dn));
1478
1479                 /* Back it out, if it fails on one */
1480                 for (i--; i >= 0; i--) {
1481                         ret2 = ldb_next_read_unlock(data->partitions[i]->module);
1482                         if (ret2 != LDB_SUCCESS) {
1483                                 ldb_debug(ldb,
1484                                           LDB_DEBUG_FATAL,
1485                                           "Failed to unlock db: %s / %s",
1486                                           ldb_errstring(ldb),
1487                                           ldb_strerror(ret2));
1488                         }
1489                 }
1490                 ret2 = ldb_next_read_unlock(module);
1491                 if (ret2 != LDB_SUCCESS) {
1492                         ldb_debug(ldb,
1493                                   LDB_DEBUG_FATAL,
1494                                   "Failed to unlock db: %s / %s",
1495                                   ldb_errstring(ldb),
1496                                   ldb_strerror(ret2));
1497                 }
1498                 return ret;
1499         }
1500
1501         return LDB_SUCCESS;
1502 }
1503
1504 /* unlock all the backends */
1505 int partition_read_unlock(struct ldb_module *module)
1506 {
1507         int i;
1508         int ret = LDB_SUCCESS;
1509         int ret2;
1510         struct ldb_context *ldb = ldb_module_get_ctx(module);
1511         struct partition_private_data *data = \
1512                 talloc_get_type(ldb_module_get_private(module),
1513                                 struct partition_private_data);
1514
1515         /*
1516          * This order must be similar to partition_{end,del}_trans()
1517          * the metadata partition (sam.ldb) unlock must be at the end.
1518          */
1519
1520         for (i=0; data && data->partitions && data->partitions[i]; i++) {
1521                 if ((module && ldb_module_flags(ldb) & LDB_FLG_ENABLE_TRACING)) {
1522                         ldb_debug(ldb, LDB_DEBUG_TRACE,
1523                                   "partition_read_unlock() -> %s",
1524                                   ldb_dn_get_linearized(
1525                                           data->partitions[i]->ctrl->dn));
1526                 }
1527                 ret2 = ldb_next_read_unlock(data->partitions[i]->module);
1528                 if (ret2 != LDB_SUCCESS) {
1529                         ldb_debug_set(ldb,
1530                                       LDB_DEBUG_FATAL,
1531                                       "Failed to lock db: %s / %s for %s",
1532                                       ldb_errstring(ldb),
1533                                       ldb_strerror(ret),
1534                                       ldb_dn_get_linearized(
1535                                               data->partitions[i]->ctrl->dn));
1536
1537                         /*
1538                          * Don't overwrite the original failure code
1539                          * if there was one
1540                          */
1541                         if (ret == LDB_SUCCESS) {
1542                                 ret = ret2;
1543                         }
1544                 }
1545         }
1546
1547         if (ldb_module_flags(ldb) & LDB_FLG_ENABLE_TRACING) {
1548                 ldb_debug(ldb, LDB_DEBUG_TRACE,
1549                           "partition_read_unlock() -> (metadata partition)");
1550         }
1551
1552         ret2 = ldb_next_read_unlock(module);
1553         if (ret2 != LDB_SUCCESS) {
1554                 ldb_debug_set(ldb,
1555                               LDB_DEBUG_FATAL,
1556                               "Failed to unlock db: %s / %s for metadata partition",
1557                               ldb_errstring(ldb),
1558                               ldb_strerror(ret2));
1559
1560                 /*
1561                  * Don't overwrite the original failure code
1562                  * if there was one
1563                  */
1564                 if (ret == LDB_SUCCESS) {
1565                         ret = ret2;
1566                 }
1567         }
1568
1569         return ret;
1570 }
1571
1572 /* extended */
1573 static int partition_extended(struct ldb_module *module, struct ldb_request *req)
1574 {
1575         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
1576                                                               struct partition_private_data);
1577         struct partition_context *ac;
1578         int ret;
1579
1580         /* if we aren't initialised yet go further */
1581         if (!data) {
1582                 return ldb_next_request(module, req);
1583         }
1584
1585         if (strcmp(req->op.extended.oid, DSDB_EXTENDED_SCHEMA_UPDATE_NOW_OID) == 0) {
1586                 /* Update the metadata.tdb to increment the schema version if needed*/
1587                 DEBUG(10, ("Incrementing the sequence_number after schema_update_now\n"));
1588                 ret = partition_metadata_inc_schema_sequence(module);
1589                 return ldb_module_done(req, NULL, NULL, ret);
1590         }
1591
1592         if (strcmp(req->op.extended.oid, LDB_EXTENDED_SEQUENCE_NUMBER) == 0) {
1593                 return partition_sequence_number(module, req);
1594         }
1595
1596         if (strcmp(req->op.extended.oid, DSDB_EXTENDED_CREATE_PARTITION_OID) == 0) {
1597                 return partition_create(module, req);
1598         }
1599
1600         /*
1601          * as the extended operation has no dn
1602          * we need to send it to all partitions
1603          */
1604
1605         ac = partition_init_ctx(module, req);
1606         if (!ac) {
1607                 return ldb_operr(ldb_module_get_ctx(module));
1608         }
1609
1610         return partition_send_all(module, ac, req);
1611 }
1612
1613 static const struct ldb_module_ops ldb_partition_module_ops = {
1614         .name              = "partition",
1615         .init_context      = partition_init,
1616         .search            = partition_search,
1617         .add               = partition_add,
1618         .modify            = partition_modify,
1619         .del               = partition_delete,
1620         .rename            = partition_rename,
1621         .extended          = partition_extended,
1622         .start_transaction = partition_start_trans,
1623         .prepare_commit    = partition_prepare_commit,
1624         .end_transaction   = partition_end_trans,
1625         .del_transaction   = partition_del_trans,
1626         .read_lock         = partition_read_lock,
1627         .read_unlock       = partition_read_unlock
1628 };
1629
1630 int ldb_partition_module_init(const char *version)
1631 {
1632         LDB_MODULE_CHECK_VERSION(version);
1633         return ldb_register_module(&ldb_partition_module_ops);
1634 }