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