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