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