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