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         int num_requests;
  45         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 && module->ldb->flags & 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(module->ldb, LDB_DEBUG_TRACE, "partition_request() -> %s",
  81                                   ldb_dn_get_linearized(partition->dn));
  82                 } else {
  83                         ldb_debug(module->ldb, 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         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 partition_private_data *data;
 138         struct ldb_control *partition_ctrl;
 139
 140         ac = talloc_get_type(req->context, struct partition_context);
 141         data = talloc_get_type(ac->module->private_data, struct partition_private_data);
 142
 143         if (!ares) {
 144                 return ldb_module_done(ac->req, NULL, NULL,
 145                                         LDB_ERR_OPERATIONS_ERROR);
 146         }
 147
 148         partition_ctrl = ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID);
 149         if (partition_ctrl && (ac->num_requests == 1 || ares->type == LDB_REPLY_ENTRY)) {
 150                 /* If we didn't fan this request out to mulitple partitions,
 151                  * or this is an individual search result, we can
 152                  * deterministily tell the caller what partition this was
 153                  * written to (repl_meta_data likes to know) */
 154                 ret = ldb_reply_add_control(ares,
 155                                             DSDB_CONTROL_CURRENT_PARTITION_OID,
 156                                             false, partition_ctrl->data);
 157                 if (ret != LDB_SUCCESS) {
 158                         return ldb_module_done(ac->req, NULL, NULL,
 159                                                ret);
 160                 }
 161         }
 162
 163         if (ares->error != LDB_SUCCESS) {
 164                 return ldb_module_done(ac->req, ares->controls,
 165                                         ares->response, ares->error);
 166         }
 167
 168         switch (ares->type) {
 169         case LDB_REPLY_REFERRAL:
 170                 /* ignore referrals for now */
 171                 break;
 172
 173         case LDB_REPLY_ENTRY:
 174                 if (ac->req->operation != LDB_SEARCH) {
 175                         ldb_set_errstring(ldb_module_get_ctx(ac->module),
 176                                 "partition_req_callback:"
 177                                 " Unsupported reply type for this request");
 178                         return ldb_module_done(ac->req, NULL, NULL,
 179                                                 LDB_ERR_OPERATIONS_ERROR);
 180                 }
 181
 182                 return ldb_module_send_entry(ac->req, ares->message, ares->controls);
 183
 184         case LDB_REPLY_DONE:
 185                 if (ac->req->operation == LDB_EXTENDED) {
 186                         /* FIXME: check for ares->response, replmd does not fill it ! */
 187                         if (ares->response) {
 188                                 if (strcmp(ares->response->oid, LDB_EXTENDED_START_TLS_OID) != 0) {
 189                                         ldb_set_errstring(ldb_module_get_ctx(ac->module),
 190                                                           "partition_req_callback:"
 191                                                           " Unknown extended reply, "
 192                                                           "only supports START_TLS");
 193                                         talloc_free(ares);
 194                                         return ldb_module_done(ac->req, NULL, NULL,
 195                                                                 LDB_ERR_OPERATIONS_ERROR);
 196                                 }
 197                         }
 198                 }
 199
 200                 ac->finished_requests++;
 201                 if (ac->finished_requests == ac->num_requests) {
 202                         /* Send back referrals if they do exist (search ops) */
 203                         if (ac->referrals != NULL) {
 204                                 const char **ref;
 205                                 for (ref = ac->referrals; *ref != NULL; ++ref) {
 206                                         ret = ldb_module_send_referral(ac->req,
 207                                                                        talloc_strdup(ac->req, *ref));
 208                                         if (ret != LDB_SUCCESS) {
 209                                                 return ldb_module_done(ac->req, NULL, NULL,
 210                                                                        ret);
 211                                         }
 212                                 }
 213                         }
 214
 215                         /* this was the last one, call callback */
 216                         return ldb_module_done(ac->req, ares->controls,
 217                                                ares->response,
 218                                                ares->error);
 219                 }
 220
 221                 /* not the last, now call the next one */
 222                 module = ac->part_req[ac->finished_requests].module;
 223                 nreq = ac->part_req[ac->finished_requests].req;
 224
 225                 ret = partition_request(module, nreq);
 226                 if (ret != LDB_SUCCESS) {
 227                         talloc_free(ares);
 228                         return ldb_module_done(ac->req, NULL, NULL, ret);
 229                 }
 230
 231                 break;
 232         }
 233
 234         talloc_free(ares);
 235         return LDB_SUCCESS;
 236 }
 237
 238 static int partition_prep_request(struct partition_context *ac,
 239                                   struct dsdb_partition *partition)
 240 {
 241         int ret;
 242         struct ldb_request *req;
 243
 244         ac->part_req = talloc_realloc(ac, ac->part_req,
 245                                         struct part_request,
 246                                         ac->num_requests + 1);
 247         if (ac->part_req == NULL) {
 248                 ldb_oom(ldb_module_get_ctx(ac->module));
 249                 return LDB_ERR_OPERATIONS_ERROR;
 250         }
 251
 252         switch (ac->req->operation) {
 253         case LDB_SEARCH:
 254                 ret = ldb_build_search_req_ex(&req, ldb_module_get_ctx(ac->module),
 255                                         ac->part_req,
 256                                         ac->req->op.search.base,
 257                                         ac->req->op.search.scope,
 258                                         ac->req->op.search.tree,
 259                                         ac->req->op.search.attrs,
 260                                         ac->req->controls,
 261                                         ac, partition_req_callback,
 262                                         ac->req);
 263                 break;
 264         case LDB_ADD:
 265                 ret = ldb_build_add_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
 266                                         ac->req->op.add.message,
 267                                         ac->req->controls,
 268                                         ac, partition_req_callback,
 269                                         ac->req);
 270                 break;
 271         case LDB_MODIFY:
 272                 ret = ldb_build_mod_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
 273                                         ac->req->op.mod.message,
 274                                         ac->req->controls,
 275                                         ac, partition_req_callback,
 276                                         ac->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                 break;
 285         case LDB_RENAME:
 286                 ret = ldb_build_rename_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
 287                                         ac->req->op.rename.olddn,
 288                                         ac->req->op.rename.newdn,
 289                                         ac->req->controls,
 290                                         ac, partition_req_callback,
 291                                         ac->req);
 292                 break;
 293         case LDB_EXTENDED:
 294                 ret = ldb_build_extended_req(&req, ldb_module_get_ctx(ac->module),
 295                                         ac->part_req,
 296                                         ac->req->op.extended.oid,
 297                                         ac->req->op.extended.data,
 298                                         ac->req->controls,
 299                                         ac, partition_req_callback,
 300                                         ac->req);
 301                 break;
 302         default:
 303                 ldb_set_errstring(ldb_module_get_ctx(ac->module),
 304                                   "Unsupported request type!");
 305                 ret = LDB_ERR_UNWILLING_TO_PERFORM;
 306         }
 307
 308         if (ret != LDB_SUCCESS) {
 309                 return ret;
 310         }
 311
 312         ac->part_req[ac->num_requests].req = req;
 313
 314         if (ac->req->controls) {
 315                 req->controls = talloc_memdup(req, ac->req->controls,
 316                                         talloc_get_size(ac->req->controls));
 317                 if (req->controls == NULL) {
 318                         ldb_oom(ldb_module_get_ctx(ac->module));
 319                         return LDB_ERR_OPERATIONS_ERROR;
 320                 }
 321         }
 322
 323         if (partition) {
 324                 ac->part_req[ac->num_requests].module = partition->module;
 325
 326                 if (!ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID)) {
 327                         ret = ldb_request_add_control(req,
 328                                                       DSDB_CONTROL_CURRENT_PARTITION_OID,
 329                                                       false, partition->ctrl);
 330                         if (ret != LDB_SUCCESS) {
 331                                 return ret;
 332                         }
 333                 }
 334
 335                 if (req->operation == LDB_SEARCH) {
 336                         /* If the search is for 'more' than this partition,
 337                          * then change the basedn, so a remote LDAP server
 338                          * doesn't object */
 339                         if (ldb_dn_compare_base(partition->ctrl->dn,
 340                                                 req->op.search.base) != 0) {
 341                                 req->op.search.base = partition->ctrl->dn;
 342                         }
 343                 }
 344
 345         } else {
 346                 /* make sure you put the module here, or
 347                  * or ldb_next_request() will skip a module */
 348                 ac->part_req[ac->num_requests].module = ac->module;
 349         }
 350
 351         ac->num_requests++;
 352
 353         return LDB_SUCCESS;
 354 }
 355
 356 static int partition_call_first(struct partition_context *ac)
 357 {
 358         return partition_request(ac->part_req[0].module, ac->part_req[0].req);
 359 }
 360
 361 /**
 362  * Send a request down to all the partitions
 363  */
 364 static int partition_send_all(struct ldb_module *module,
 365                               struct partition_context *ac,
 366                               struct ldb_request *req)
 367 {
 368         int i;
 369         struct partition_private_data *data = talloc_get_type(module->private_data,
 370                                                               struct partition_private_data);
 371         int ret = partition_prep_request(ac, NULL);
 372         if (ret != LDB_SUCCESS) {
 373                 return ret;
 374         }
 375         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 376                 ret = partition_prep_request(ac, data->partitions[i]);
 377                 if (ret != LDB_SUCCESS) {
 378                         return ret;
 379                 }
 380         }
 381
 382         /* fire the first one */
 383         return partition_call_first(ac);
 384 }
 385
 386 /**
 387  * Figure out which backend a request needs to be aimed at.  Some
 388  * requests must be replicated to all backends
 389  */
 390 static int partition_replicate(struct ldb_module *module, struct ldb_request *req, struct ldb_dn *dn)
 391 {
 392         struct partition_context *ac;
 393         unsigned i;
 394         int ret;
 395         struct dsdb_partition *partition;
 396         struct partition_private_data *data = talloc_get_type(module->private_data,
 397                                                               struct partition_private_data);
 398         if (!data || !data->partitions) {
 399                 return ldb_next_request(module, req);
 400         }
 401
 402         if (req->operation != LDB_SEARCH && ldb_dn_is_special(dn)) {
 403                 /* Is this a special DN, we need to replicate to every backend? */
 404                 for (i=0; data->replicate && data->replicate[i]; i++) {
 405                         if (ldb_dn_compare(data->replicate[i],
 406                                            dn) == 0) {
 407
 408                                 ac = partition_init_ctx(module, req);
 409                                 if (!ac) {
 410                                         return LDB_ERR_OPERATIONS_ERROR;
 411                                 }
 412
 413                                 return partition_send_all(module, ac, req);
 414                         }
 415                 }
 416         }
 417
 418         /* Otherwise, we need to find the partition to fire it to */
 419
 420         /* Find partition */
 421         partition = find_partition(data, dn, req);
 422         if (!partition) {
 423                 /*
 424                  * if we haven't found a matching partition
 425                  * pass the request to the main ldb
 426                  *
 427                  * TODO: we should maybe return an error here
 428                  *       if it's not a special dn
 429                  */
 430
 431                 return ldb_next_request(module, req);
 432         }
 433
 434         ac = partition_init_ctx(module, req);
 435         if (!ac) {
 436                 return LDB_ERR_OPERATIONS_ERROR;
 437         }
 438
 439         /* we need to add a control but we never touch the original request */
 440         ret = partition_prep_request(ac, partition);
 441         if (ret != LDB_SUCCESS) {
 442                 return ret;
 443         }
 444
 445         /* fire the first one */
 446         return partition_call_first(ac);
 447 }
 448
 449 /* search */
 450 static int partition_search(struct ldb_module *module, struct ldb_request *req)
 451 {
 452         struct ldb_control **saved_controls;
 453         /* Find backend */
 454         struct partition_private_data *data = talloc_get_type(module->private_data,
 455                                                               struct partition_private_data);
 456         struct partition_context *ac;
 457         struct ldb_context *ldb;
 458         struct loadparm_context *lp_ctx;
 459
 460         struct ldb_control *search_control = ldb_request_get_control(req, LDB_CONTROL_SEARCH_OPTIONS_OID);
 461         struct ldb_control *domain_scope_control = ldb_request_get_control(req, LDB_CONTROL_DOMAIN_SCOPE_OID);
 462
 463         struct ldb_search_options_control *search_options = NULL;
 464         struct dsdb_partition *p;
 465         unsigned int i, j;
 466         int ret;
 467         bool domain_scope = false, phantom_root = false;
 468
 469         ret = partition_reload_if_required(module, data);
 470         if (ret != LDB_SUCCESS) {
 471                 return ret;
 472         }
 473
 474         p = find_partition(data, NULL, req);
 475         if (p != NULL) {
 476                 /* the caller specified what partition they want the
 477                  * search - just pass it on
 478                  */
 479                 return ldb_next_request(p->module, req);
 480         }
 481
 482         /* Get back the search options from the search control, and mark it as
 483          * non-critical (to make backends and also dcpromo happy).
 484          */
 485         if (search_control) {
 486                 search_options = talloc_get_type(search_control->data, struct ldb_search_options_control);
 487                 search_control->critical = 0;
 488
 489         }
 490
 491         /* Remove the "domain_scope" control, so we don't confuse a backend
 492          * server */
 493         if (domain_scope_control && !save_controls(domain_scope_control, req, &saved_controls)) {
 494                 ldb_oom(ldb_module_get_ctx(module));
 495                 return LDB_ERR_OPERATIONS_ERROR;
 496         }
 497
 498         /* Locate the options */
 499         domain_scope = (search_options
 500                 && (search_options->search_options & LDB_SEARCH_OPTION_DOMAIN_SCOPE))
 501                 || domain_scope_control;
 502         phantom_root = search_options
 503                 && (search_options->search_options & LDB_SEARCH_OPTION_PHANTOM_ROOT);
 504
 505         /* Remove handled options from the search control flag */
 506         if (search_options) {
 507                 search_options->search_options = search_options->search_options
 508                         & ~LDB_SEARCH_OPTION_DOMAIN_SCOPE
 509                         & ~LDB_SEARCH_OPTION_PHANTOM_ROOT;
 510         }
 511
 512         if (!data || !data->partitions) {
 513                 return ldb_next_request(module, req);
 514         }
 515
 516         ac = partition_init_ctx(module, req);
 517         if (!ac) {
 518                 return LDB_ERR_OPERATIONS_ERROR;
 519         }
 520
 521         ldb = ldb_module_get_ctx(ac->module);
 522         lp_ctx = talloc_get_type(ldb_get_opaque(ldb, "loadparm"),
 523                                                 struct loadparm_context);
 524
 525         /* Search from the base DN */
 526         if (ldb_dn_is_null(req->op.search.base)) {
 527                 return partition_send_all(module, ac, req);
 528         }
 529
 530         for (i=0; data->partitions[i]; i++) {
 531                 bool match = false, stop = false;
 532
 533                 if (phantom_root) {
 534                         /* Phantom root: Find all partitions under the
 535                          * search base. We match if:
 536                          *
 537                          * 1) the DN we are looking for exactly matches a
 538                          *    certain partition and always stop
 539                          * 2) the DN we are looking for is a parent of certain
 540                          *    partitions and it isn't a scope base search
 541                          * 3) the DN we are looking for is a child of a certain
 542                          *    partition and always stop
 543                          *    - we don't need to go any further up in the
 544                          *    hierarchy!
 545                          */
 546                         if (ldb_dn_compare(data->partitions[i]->ctrl->dn,
 547                                            req->op.search.base) == 0) {
 548                                 match = true;
 549                                 stop = true;
 550                         }
 551                         if (!match &&
 552                             (ldb_dn_compare_base(req->op.search.base,
 553                                                  data->partitions[i]->ctrl->dn) == 0 &&
 554                              req->op.search.scope != LDB_SCOPE_BASE)) {
 555                                 match = true;
 556                         }
 557                         if (!match &&
 558                             ldb_dn_compare_base(data->partitions[i]->ctrl->dn,
 559                                                 req->op.search.base) == 0) {
 560                                 match = true;
 561                                 stop = true; /* note that this relies on partition ordering */
 562                         }
 563                 } else {
 564                         /* Domain scope: Find all partitions under the search
 565                          * base.
 566                          *
 567                          * We generate referral candidates if we haven't
 568                          * specified the domain scope control, haven't a base
 569                          * search* scope and the DN we are looking for is a real
 570                          * predecessor of certain partitions. When a new
 571                          * referral candidate is nearer to the DN than an
 572                          * existing one delete the latter (we want to have only
 573                          * the closest ones). When we checked this for all
 574                          * candidates we have the final referrals.
 575                          *
 576                          * We match if the DN we are looking for is a child of
 577                          * a certain partition or the partition
 578                          * DN itself - we don't need to go any further
 579                          * up in the hierarchy!
 580                          */
 581                         if ((!domain_scope) &&
 582                             (req->op.search.scope != LDB_SCOPE_BASE) &&
 583                             (ldb_dn_compare_base(req->op.search.base,
 584                                                  data->partitions[i]->ctrl->dn) == 0) &&
 585                             (ldb_dn_compare(req->op.search.base,
 586                                             data->partitions[i]->ctrl->dn) != 0)) {
 587                                 char *ref = talloc_asprintf(ac,
 588                                                             "ldap://%s/%s%s",
 589                                                             lp_dnsdomain(lp_ctx),
 590                                                             ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
 591                                                             req->op.search.scope == LDB_SCOPE_ONELEVEL ? "??base" : "");
 592
 593                                 if (ref == NULL) {
 594                                         ldb_oom(ldb);
 595                                         return LDB_ERR_OPERATIONS_ERROR;
 596                                 }
 597
 598                                 /* Initialise the referrals list */
 599                                 if (ac->referrals == NULL) {
 600                                         ac->referrals = (const char **) str_list_make_empty(ac);
 601                                         if (ac->referrals == NULL) {
 602                                                 ldb_oom(ldb);
 603                                                 return LDB_ERR_OPERATIONS_ERROR;
 604                                         }
 605                                 }
 606
 607                                 /* Check if the new referral candidate is
 608                                  * closer to the base DN than already
 609                                  * saved ones and delete the latters */
 610                                 j = 0;
 611                                 while (ac->referrals[j] != NULL) {
 612                                         if (strstr(ac->referrals[j],
 613                                                    ldb_dn_get_linearized(data->partitions[i]->ctrl->dn)) != NULL) {
 614                                                 str_list_remove(ac->referrals,
 615                                                                 ac->referrals[j]);
 616                                         } else {
 617                                                 ++j;
 618                                         }
 619                                 }
 620
 621                                 /* Add our new candidate */
 622                                 ac->referrals = str_list_add(ac->referrals, ref);
 623
 624                                 talloc_free(ref);
 625
 626                                 if (ac->referrals == NULL) {
 627                                         ldb_oom(ldb);
 628                                         return LDB_ERR_OPERATIONS_ERROR;
 629                                 }
 630                         }
 631                         if (ldb_dn_compare_base(data->partitions[i]->ctrl->dn, req->op.search.base) == 0) {
 632                                 match = true;
 633                                 stop = true; /* note that this relies on partition ordering */
 634                         }
 635                 }
 636
 637                 if (match) {
 638                         ret = partition_prep_request(ac, data->partitions[i]);
 639                         if (ret != LDB_SUCCESS) {
 640                                 return ret;
 641                         }
 642                 }
 643
 644                 if (stop) break;
 645         }
 646
 647         /* Perhaps we didn't match any partitions. Try the main partition */
 648         if (ac->num_requests == 0) {
 649                 talloc_free(ac);
 650                 return ldb_next_request(module, req);
 651         }
 652
 653         /* fire the first one */
 654         return partition_call_first(ac);
 655 }
 656
 657 /* add */
 658 static int partition_add(struct ldb_module *module, struct ldb_request *req)
 659 {
 660         return partition_replicate(module, req, req->op.add.message->dn);
 661 }
 662
 663 /* modify */
 664 static int partition_modify(struct ldb_module *module, struct ldb_request *req)
 665 {
 666         return partition_replicate(module, req, req->op.mod.message->dn);
 667 }
 668
 669 /* delete */
 670 static int partition_delete(struct ldb_module *module, struct ldb_request *req)
 671 {
 672         return partition_replicate(module, req, req->op.del.dn);
 673 }
 674
 675 /* rename */
 676 static int partition_rename(struct ldb_module *module, struct ldb_request *req)
 677 {
 678         /* Find backend */
 679         struct dsdb_partition *backend, *backend2;
 680
 681         struct partition_private_data *data = talloc_get_type(module->private_data,
 682                                                               struct partition_private_data);
 683
 684         /* Skip the lot if 'data' isn't here yet (initialisation) */
 685         if (!data) {
 686                 return LDB_ERR_OPERATIONS_ERROR;
 687         }
 688
 689         backend = find_partition(data, req->op.rename.olddn, req);
 690         backend2 = find_partition(data, req->op.rename.newdn, req);
 691
 692         if ((backend && !backend2) || (!backend && backend2)) {
 693                 return LDB_ERR_AFFECTS_MULTIPLE_DSAS;
 694         }
 695
 696         if (backend != backend2) {
 697                 ldb_asprintf_errstring(ldb_module_get_ctx(module),
 698                                        "Cannot rename from %s in %s to %s in %s: %s",
 699                                        ldb_dn_get_linearized(req->op.rename.olddn),
 700                                        ldb_dn_get_linearized(backend->ctrl->dn),
 701                                        ldb_dn_get_linearized(req->op.rename.newdn),
 702                                        ldb_dn_get_linearized(backend2->ctrl->dn),
 703                                        ldb_strerror(LDB_ERR_AFFECTS_MULTIPLE_DSAS));
 704                 return LDB_ERR_AFFECTS_MULTIPLE_DSAS;
 705         }
 706
 707         return partition_replicate(module, req, req->op.rename.olddn);
 708 }
 709
 710 /* start a transaction */
 711 static int partition_start_trans(struct ldb_module *module)
 712 {
 713         int i, ret;
 714         struct partition_private_data *data = talloc_get_type(module->private_data,
 715                                                               struct partition_private_data);
 716         /* Look at base DN */
 717         /* Figure out which partition it is under */
 718         /* Skip the lot if 'data' isn't here yet (initialization) */
 719         if ((module && module->ldb->flags & LDB_FLG_ENABLE_TRACING)) {
 720                 ldb_debug(module->ldb, LDB_DEBUG_TRACE, "partition_start_trans() -> (metadata partition)");
 721         }
 722         ret = ldb_next_start_trans(module);
 723         if (ret != LDB_SUCCESS) {
 724                 return ret;
 725         }
 726
 727         ret = partition_reload_if_required(module, data);
 728         if (ret != LDB_SUCCESS) {
 729                 return ret;
 730         }
 731
 732         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 733                 if ((module && module->ldb->flags & LDB_FLG_ENABLE_TRACING)) {
 734                         ldb_debug(module->ldb, LDB_DEBUG_TRACE, "partition_start_trans() -> %s",
 735                                   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
 736                 }
 737                 ret = ldb_next_start_trans(data->partitions[i]->module);
 738                 if (ret != LDB_SUCCESS) {
 739                         /* Back it out, if it fails on one */
 740                         for (i--; i >= 0; i--) {
 741                                 ldb_next_del_trans(data->partitions[i]->module);
 742                         }
 743                         ldb_next_del_trans(module);
 744                         return ret;
 745                 }
 746         }
 747
 748         data->in_transaction++;
 749
 750         return LDB_SUCCESS;
 751 }
 752
 753 /* prepare for a commit */
 754 static int partition_prepare_commit(struct ldb_module *module)
 755 {
 756         int i;
 757         struct partition_private_data *data = talloc_get_type(module->private_data,
 758                                                               struct partition_private_data);
 759
 760         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 761                 int ret;
 762
 763                 if ((module && module->ldb->flags & LDB_FLG_ENABLE_TRACING)) {
 764                         ldb_debug(module->ldb, LDB_DEBUG_TRACE, "partition_prepare_commit() -> %s",
 765                                   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
 766                 }
 767                 ret = ldb_next_prepare_commit(data->partitions[i]->module);
 768                 if (ret != LDB_SUCCESS) {
 769                         ldb_asprintf_errstring(module->ldb, "prepare_commit error on %s: %s",
 770                                                ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
 771                                                ldb_errstring(module->ldb));
 772                         return ret;
 773                 }
 774         }
 775
 776         if ((module && module->ldb->flags & LDB_FLG_ENABLE_TRACING)) {
 777                 ldb_debug(module->ldb, LDB_DEBUG_TRACE, "partition_prepare_commit() -> (metadata partition)");
 778         }
 779         return ldb_next_prepare_commit(module);
 780 }
 781
 782
 783 /* end a transaction */
 784 static int partition_end_trans(struct ldb_module *module)
 785 {
 786         int i, ret, ret2;
 787         struct partition_private_data *data = talloc_get_type(module->private_data,
 788                                                               struct partition_private_data);
 789
 790         ret = LDB_SUCCESS;
 791
 792         if (data->in_transaction == 0) {
 793                 DEBUG(0,("partition end transaction mismatch\n"));
 794                 ret = LDB_ERR_OPERATIONS_ERROR;
 795         } else {
 796                 data->in_transaction--;
 797         }
 798
 799         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 800                 if ((module && module->ldb->flags & LDB_FLG_ENABLE_TRACING)) {
 801                         ldb_debug(module->ldb, LDB_DEBUG_TRACE, "partition_end_trans() -> %s",
 802                                   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
 803                 }
 804                 ret2 = ldb_next_end_trans(data->partitions[i]->module);
 805                 if (ret2 != LDB_SUCCESS) {
 806                         ldb_asprintf_errstring(module->ldb, "end_trans error on %s: %s",
 807                                                ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
 808                                                ldb_errstring(module->ldb));
 809                         ret = ret2;
 810                 }
 811         }
 812
 813         if ((module && module->ldb->flags & LDB_FLG_ENABLE_TRACING)) {
 814                 ldb_debug(module->ldb, LDB_DEBUG_TRACE, "partition_end_trans() -> (metadata partition)");
 815         }
 816         ret2 = ldb_next_end_trans(module);
 817         if (ret2 != LDB_SUCCESS) {
 818                 ret = ret2;
 819         }
 820         return ret;
 821 }
 822
 823 /* delete a transaction */
 824 static int partition_del_trans(struct ldb_module *module)
 825 {
 826         int i, ret, final_ret = LDB_SUCCESS;
 827         struct partition_private_data *data = talloc_get_type(module->private_data,
 828                                                               struct partition_private_data);
 829         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 830                 if ((module && module->ldb->flags & LDB_FLG_ENABLE_TRACING)) {
 831                         ldb_debug(module->ldb, LDB_DEBUG_TRACE, "partition_del_trans() -> %s",
 832                                   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
 833                 }
 834                 ret = ldb_next_del_trans(data->partitions[i]->module);
 835                 if (ret != LDB_SUCCESS) {
 836                         ldb_asprintf_errstring(module->ldb, "del_trans error on %s: %s",
 837                                                ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
 838                                                ldb_errstring(module->ldb));
 839                         final_ret = ret;
 840                 }
 841         }
 842
 843         if (data->in_transaction == 0) {
 844                 DEBUG(0,("partition del transaction mismatch\n"));
 845                 return LDB_ERR_OPERATIONS_ERROR;
 846         }
 847         data->in_transaction--;
 848
 849         if ((module && module->ldb->flags & LDB_FLG_ENABLE_TRACING)) {
 850                 ldb_debug(module->ldb, LDB_DEBUG_TRACE, "partition_del_trans() -> (metadata partition)");
 851         }
 852         ret = ldb_next_del_trans(module);
 853         if (ret != LDB_SUCCESS) {
 854                 final_ret = ret;
 855         }
 856         return final_ret;
 857 }
 858
 859 int partition_primary_sequence_number(struct ldb_module *module, TALLOC_CTX *mem_ctx,
 860                                      enum ldb_sequence_type type, uint64_t *seq_number)
 861 {
 862         int ret;
 863         struct ldb_result *res;
 864         struct ldb_seqnum_request *tseq;
 865         struct ldb_request *treq;
 866         struct ldb_seqnum_result *seqr;
 867         res = talloc_zero(mem_ctx, struct ldb_result);
 868         if (res == NULL) {
 869                 return LDB_ERR_OPERATIONS_ERROR;
 870         }
 871         tseq = talloc_zero(res, struct ldb_seqnum_request);
 872         if (tseq == NULL) {
 873                 talloc_free(res);
 874                 return LDB_ERR_OPERATIONS_ERROR;
 875         }
 876         tseq->type = type;
 877
 878         ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
 879                                      LDB_EXTENDED_SEQUENCE_NUMBER,
 880                                      tseq,
 881                                      NULL,
 882                                      res,
 883                                      ldb_extended_default_callback,
 884                                      NULL);
 885         if (ret != LDB_SUCCESS) {
 886                 talloc_free(res);
 887                 return ret;
 888         }
 889
 890         ret = ldb_next_request(module, treq);
 891         if (ret != LDB_SUCCESS) {
 892                 talloc_free(res);
 893                 return ret;
 894         }
 895         ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
 896         if (ret != LDB_SUCCESS) {
 897                 talloc_free(res);
 898                 return ret;
 899         }
 900
 901         seqr = talloc_get_type(res->extended->data,
 902                                struct ldb_seqnum_result);
 903         if (seqr->flags & LDB_SEQ_TIMESTAMP_SEQUENCE) {
 904                 ret = LDB_ERR_OPERATIONS_ERROR;
 905                 ldb_set_errstring(ldb_module_get_ctx(module), "Primary backend in partitions module returned a timestamp based seq number (must return a normal number)");
 906                 talloc_free(res);
 907                 return ret;
 908         } else {
 909                 *seq_number = seqr->seq_num;
 910         }
 911         talloc_free(res);
 912         return LDB_SUCCESS;
 913 }
 914
 915 /* FIXME: This function is still semi-async */
 916 static int partition_sequence_number(struct ldb_module *module, struct ldb_request *req)
 917 {
 918         int i, ret;
 919         uint64_t seq_number = 0;
 920         uint64_t timestamp_sequence = 0;
 921         uint64_t timestamp = 0;
 922         struct partition_private_data *data = talloc_get_type(module->private_data,
 923                                                               struct partition_private_data);
 924         struct ldb_seqnum_request *seq;
 925         struct ldb_seqnum_result *seqr;
 926         struct ldb_request *treq;
 927         struct ldb_seqnum_request *tseq;
 928         struct ldb_seqnum_result *tseqr;
 929         struct ldb_extended *ext;
 930         struct ldb_result *res;
 931         struct dsdb_partition *p;
 932
 933         p = find_partition(data, NULL, req);
 934         if (p != NULL) {
 935                 /* the caller specified what partition they want the
 936                  * sequence number operation on - just pass it on
 937                  */
 938                 return ldb_next_request(p->module, req);
 939         }
 940
 941         seq = talloc_get_type(req->op.extended.data, struct ldb_seqnum_request);
 942
 943         switch (seq->type) {
 944         case LDB_SEQ_NEXT:
 945         case LDB_SEQ_HIGHEST_SEQ:
 946
 947                 ret = partition_primary_sequence_number(module, req, seq->type, &seq_number);
 948                 if (ret != LDB_SUCCESS) {
 949                         return ret;
 950                 }
 951
 952                 /* Skip the lot if 'data' isn't here yet (initialisation) */
 953                 for (i=0; data && data->partitions && data->partitions[i]; i++) {
 954
 955                         res = talloc_zero(req, struct ldb_result);
 956                         if (res == NULL) {
 957                                 return LDB_ERR_OPERATIONS_ERROR;
 958                         }
 959                         tseq = talloc_zero(res, struct ldb_seqnum_request);
 960                         if (tseq == NULL) {
 961                                 talloc_free(res);
 962                                 return LDB_ERR_OPERATIONS_ERROR;
 963                         }
 964                         tseq->type = seq->type;
 965
 966                         ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
 967                                                      LDB_EXTENDED_SEQUENCE_NUMBER,
 968                                                      tseq,
 969                                                      NULL,
 970                                                      res,
 971                                                      ldb_extended_default_callback,
 972                                                      NULL);
 973                         if (ret != LDB_SUCCESS) {
 974                                 talloc_free(res);
 975                                 return ret;
 976                         }
 977
 978                         if (!ldb_request_get_control(treq, DSDB_CONTROL_CURRENT_PARTITION_OID)) {
 979                                 ret = ldb_request_add_control(treq,
 980                                                               DSDB_CONTROL_CURRENT_PARTITION_OID,
 981                                                               false, data->partitions[i]->ctrl);
 982                                 if (ret != LDB_SUCCESS) {
 983                                         talloc_free(res);
 984                                         return ret;
 985                                 }
 986                         }
 987
 988                         ret = partition_request(data->partitions[i]->module, treq);
 989                         if (ret != LDB_SUCCESS) {
 990                                 talloc_free(res);
 991                                 return ret;
 992                         }
 993                         ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
 994                         if (ret != LDB_SUCCESS) {
 995                                 talloc_free(res);
 996                                 return ret;
 997                         }
 998                         tseqr = talloc_get_type(res->extended->data,
 999                                                 struct ldb_seqnum_result);
1000                         if (tseqr->flags & LDB_SEQ_TIMESTAMP_SEQUENCE) {
1001                                 timestamp_sequence = MAX(timestamp_sequence,
1002                                                          tseqr->seq_num);
1003                         } else {
1004                                 seq_number += tseqr->seq_num;
1005                         }
1006                         talloc_free(res);
1007                 }
1008                 /* fall through */
1009         case LDB_SEQ_HIGHEST_TIMESTAMP:
1010
1011                 res = talloc_zero(req, struct ldb_result);
1012                 if (res == NULL) {
1013                         return LDB_ERR_OPERATIONS_ERROR;
1014                 }
1015
1016                 tseq = talloc_zero(res, struct ldb_seqnum_request);
1017                 if (tseq == NULL) {
1018                         talloc_free(res);
1019                         return LDB_ERR_OPERATIONS_ERROR;
1020                 }
1021                 tseq->type = LDB_SEQ_HIGHEST_TIMESTAMP;
1022
1023                 ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
1024                                              LDB_EXTENDED_SEQUENCE_NUMBER,
1025                                              tseq,
1026                                              NULL,
1027                                              res,
1028                                              ldb_extended_default_callback,
1029                                              NULL);
1030                 if (ret != LDB_SUCCESS) {
1031                         talloc_free(res);
1032                         return ret;
1033                 }
1034
1035                 ret = ldb_next_request(module, treq);
1036                 if (ret != LDB_SUCCESS) {
1037                         talloc_free(res);
1038                         return ret;
1039                 }
1040                 ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
1041                 if (ret != LDB_SUCCESS) {
1042                         talloc_free(res);
1043                         return ret;
1044                 }
1045
1046                 tseqr = talloc_get_type(res->extended->data,
1047                                            struct ldb_seqnum_result);
1048                 timestamp = tseqr->seq_num;
1049
1050                 talloc_free(res);
1051
1052                 /* Skip the lot if 'data' isn't here yet (initialisation) */
1053                 for (i=0; data && data->partitions && data->partitions[i]; i++) {
1054
1055                         res = talloc_zero(req, struct ldb_result);
1056                         if (res == NULL) {
1057                                 return LDB_ERR_OPERATIONS_ERROR;
1058                         }
1059
1060                         tseq = talloc_zero(res, struct ldb_seqnum_request);
1061                         if (tseq == NULL) {
1062                                 talloc_free(res);
1063                                 return LDB_ERR_OPERATIONS_ERROR;
1064                         }
1065                         tseq->type = LDB_SEQ_HIGHEST_TIMESTAMP;
1066
1067                         ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
1068                                                      LDB_EXTENDED_SEQUENCE_NUMBER,
1069                                                      tseq,
1070                                                      NULL,
1071                                                      res,
1072                                                      ldb_extended_default_callback,
1073                                                      NULL);
1074                         if (ret != LDB_SUCCESS) {
1075                                 talloc_free(res);
1076                                 return ret;
1077                         }
1078
1079                         if (!ldb_request_get_control(treq, DSDB_CONTROL_CURRENT_PARTITION_OID)) {
1080                                 ret = ldb_request_add_control(treq,
1081                                                               DSDB_CONTROL_CURRENT_PARTITION_OID,
1082                                                               false, data->partitions[i]->ctrl);
1083                                 if (ret != LDB_SUCCESS) {
1084                                         talloc_free(res);
1085                                         return ret;
1086                                 }
1087                         }
1088
1089                         ret = partition_request(data->partitions[i]->module, treq);
1090                         if (ret != LDB_SUCCESS) {
1091                                 talloc_free(res);
1092                                 return ret;
1093                         }
1094                         ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
1095                         if (ret != LDB_SUCCESS) {
1096                                 talloc_free(res);
1097                                 return ret;
1098                         }
1099
1100                         tseqr = talloc_get_type(res->extended->data,
1101                                                   struct ldb_seqnum_result);
1102                         timestamp = MAX(timestamp, tseqr->seq_num);
1103
1104                         talloc_free(res);
1105                 }
1106
1107                 break;
1108         }
1109
1110         ext = talloc_zero(req, struct ldb_extended);
1111         if (!ext) {
1112                 return LDB_ERR_OPERATIONS_ERROR;
1113         }
1114         seqr = talloc_zero(ext, struct ldb_seqnum_result);
1115         if (seqr == NULL) {
1116                 talloc_free(ext);
1117                 return LDB_ERR_OPERATIONS_ERROR;
1118         }
1119         ext->oid = LDB_EXTENDED_SEQUENCE_NUMBER;
1120         ext->data = seqr;
1121
1122         switch (seq->type) {
1123         case LDB_SEQ_NEXT:
1124         case LDB_SEQ_HIGHEST_SEQ:
1125
1126                 /* Has someone above set a timebase sequence? */
1127                 if (timestamp_sequence) {
1128                         seqr->seq_num = (((unsigned long long)timestamp << 24) | (seq_number & 0xFFFFFF));
1129                 } else {
1130                         seqr->seq_num = seq_number;
1131                 }
1132
1133                 if (timestamp_sequence > seqr->seq_num) {
1134                         seqr->seq_num = timestamp_sequence;
1135                         seqr->flags |= LDB_SEQ_TIMESTAMP_SEQUENCE;
1136                 }
1137
1138                 seqr->flags |= LDB_SEQ_GLOBAL_SEQUENCE;
1139                 break;
1140         case LDB_SEQ_HIGHEST_TIMESTAMP:
1141                 seqr->seq_num = timestamp;
1142                 break;
1143         }
1144
1145         if (seq->type == LDB_SEQ_NEXT) {
1146                 seqr->seq_num++;
1147         }
1148
1149         /* send request done */
1150         return ldb_module_done(req, NULL, ext, LDB_SUCCESS);
1151 }
1152
1153 /* extended */
1154 static int partition_extended(struct ldb_module *module, struct ldb_request *req)
1155 {
1156         struct partition_private_data *data;
1157         struct partition_context *ac;
1158         int ret;
1159
1160         data = talloc_get_type(module->private_data, struct partition_private_data);
1161         if (!data) {
1162                 return ldb_next_request(module, req);
1163         }
1164
1165         ret = partition_reload_if_required(module, data);
1166         if (ret != LDB_SUCCESS) {
1167                 return ret;
1168         }
1169
1170         if (strcmp(req->op.extended.oid, LDB_EXTENDED_SEQUENCE_NUMBER) == 0) {
1171                 return partition_sequence_number(module, req);
1172         }
1173
1174         if (strcmp(req->op.extended.oid, DSDB_EXTENDED_CREATE_PARTITION_OID) == 0) {
1175                 return partition_create(module, req);
1176         }
1177
1178         /*
1179          * as the extended operation has no dn
1180          * we need to send it to all partitions
1181          */
1182
1183         ac = partition_init_ctx(module, req);
1184         if (!ac) {
1185                 return LDB_ERR_OPERATIONS_ERROR;
1186         }
1187
1188         return partition_send_all(module, ac, req);
1189 }
1190
1191 _PUBLIC_ const struct ldb_module_ops ldb_partition_module_ops = {
1192         .name              = "partition",
1193         .init_context      = partition_init,
1194         .search            = partition_search,
1195         .add               = partition_add,
1196         .modify            = partition_modify,
1197         .del               = partition_delete,
1198         .rename            = partition_rename,
1199         .extended          = partition_extended,
1200         .start_transaction = partition_start_trans,
1201         .prepare_commit    = partition_prepare_commit,
1202         .end_transaction   = partition_end_trans,
1203         .del_transaction   = partition_del_trans,
1204 };