source4/dsdb/samdb/ldb_modules/partition.c

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