source/dsdb/samdb/ldb_modules/partition.c

   1 /*
   2    Partitions ldb module
   3
   4    Copyright (C) Andrew Bartlett <abartlet@samba.org> 2006
   5    Copyright (C) Stefan Metzmacher <metze@samba.org> 2007
   6
   7    * NOTICE: this module is NOT released under the GNU LGPL license as
   8    * other ldb code. This module is release under the GNU GPL v2 or
   9    * later license.
  10
  11    This program is free software; you can redistribute it and/or modify
  12    it under the terms of the GNU General Public License as published by
  13    the Free Software Foundation; either version 2 of the License, or
  14    (at your option) any later version.
  15
  16    This program is distributed in the hope that it will be useful,
  17    but WITHOUT ANY WARRANTY; without even the implied warranty of
  18    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  19    GNU General Public License for more details.
  20
  21    You should have received a copy of the GNU General Public License
  22    along with this program; if not, write to the Free Software
  23    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  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/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 static struct ldb_module *find_backend(struct ldb_module *module, struct ldb_request *req, struct ldb_dn *dn)
 123 {
 124         struct dsdb_control_current_partition *partition;
 125         struct partition_private_data *data = talloc_get_type(module->private_data,
 126                                                               struct partition_private_data);
 127
 128         /* Skip the lot if 'data' isn't here yet (initialistion) */
 129         if (!data) {
 130                 return module;
 131         }
 132
 133         partition = find_partition(data, dn);
 134         if (!partition) {
 135                 return module;
 136         }
 137
 138         return make_module_for_next_request(req, module->ldb, partition->module);
 139 };
 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_module *partition,
 209                                   struct ldb_dn *partition_base_dn)
 210 {
 211         int ret;
 212         struct ldb_module *next = make_module_for_next_request(ac->module, ac->module->ldb, partition);
 213         struct ldb_request *req;
 214         ac->down_req = talloc_realloc(ac, ac->down_req,
 215                                         struct ldb_request *, ac->num_requests + 1);
 216         if (!ac->down_req) {
 217                 ldb_set_errstring(ac->module->ldb, "Out of Memory");
 218                 return LDB_ERR_OPERATIONS_ERROR;
 219         }
 220         req = ac->down_req[ac->num_requests] = talloc(ac, struct ldb_request);
 221         if (req == NULL) {
 222                 ldb_set_errstring(ac->module->ldb, "Out of Memory");
 223                 return LDB_ERR_OPERATIONS_ERROR;
 224         }
 225
 226         *ac->down_req[ac->num_requests] = *ac->orig_req; /* copy the request */
 227
 228         if (req->operation == LDB_SEARCH) {
 229                 /* If the search is for 'more' than this partition,
 230                  * then change the basedn, so a remote LDAP server
 231                  * doesn't object */
 232                 if (ldb_dn_compare_base(partition_base_dn, req->op.search.base) != 0) {
 233                         req->op.search.base = partition_base_dn;
 234                 }
 235                 req->callback = partition_search_callback;
 236                 req->context = ac;
 237         } else {
 238                 req->callback = partition_other_callback;
 239                 req->context = ac;
 240         }
 241
 242         /* Spray off search requests to all backends */
 243         ret = ldb_next_request(next, req);
 244         if (ret != LDB_SUCCESS) {
 245                 return ret;
 246         }
 247
 248         ac->num_requests++;
 249         return LDB_SUCCESS;
 250 }
 251
 252 /* Send a request down to all the partitions */
 253 static int partition_send_all(struct ldb_module *module,
 254                               struct partition_context *ac, struct ldb_request *req)
 255 {
 256         int i;
 257         struct partition_private_data *data = talloc_get_type(module->private_data,
 258                                                               struct partition_private_data);
 259         int ret = partition_send_request(ac, module->next, NULL);
 260         if (ret != LDB_SUCCESS) {
 261                 return ret;
 262         }
 263         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 264                 ret = partition_send_request(ac, data->partitions[i]->module, data->partitions[i]->dn);
 265                 if (ret != LDB_SUCCESS) {
 266                         return ret;
 267                 }
 268         }
 269         return LDB_SUCCESS;
 270 }
 271
 272 /* Figure out which backend a request needs to be aimed at.  Some
 273  * requests must be replicated to all backends */
 274 static int partition_replicate(struct ldb_module *module, struct ldb_request *req, struct ldb_dn *dn)
 275 {
 276         int i;
 277         struct ldb_module *backend;
 278         struct partition_private_data *data = talloc_get_type(module->private_data,
 279                                                               struct partition_private_data);
 280
 281         /* Is this a special DN, we need to replicate to every backend? */
 282         for (i=0; data->replicate && data->replicate[i]; i++) {
 283                 if (ldb_dn_compare(data->replicate[i],
 284                                    dn) == 0) {
 285                         struct partition_context *ac;
 286
 287                         ac = partition_init_handle(req, module);
 288                         if (!ac) {
 289                                 return LDB_ERR_OPERATIONS_ERROR;
 290                         }
 291
 292                         return partition_send_all(module, ac, req);
 293                 }
 294         }
 295
 296         /* Otherwise, we need to find the backend to fire it to */
 297
 298         /* Find backend */
 299         backend = find_backend(module, req, dn);
 300
 301         /* issue request */
 302         return ldb_next_request(backend, req);
 303
 304 }
 305
 306 /* search */
 307 static int partition_search(struct ldb_module *module, struct ldb_request *req)
 308 {
 309         /* Find backend */
 310         struct partition_private_data *data = talloc_get_type(module->private_data,
 311                                                               struct partition_private_data);
 312         /* issue request */
 313
 314         /* (later) consider if we should be searching multiple
 315          * partitions (for 'invisible' partition behaviour */
 316         if (ldb_get_opaque(module->ldb, "global_catalog")) {
 317                 int ret, i;
 318                 struct partition_context *ac;
 319
 320                 ac = partition_init_handle(req, module);
 321                 if (!ac) {
 322                         return LDB_ERR_OPERATIONS_ERROR;
 323                 }
 324
 325                 /* Search from the base DN */
 326                 if (!req->op.search.base || ldb_dn_is_null(req->op.search.base)) {
 327                         return partition_send_all(module, ac, req);
 328                 }
 329                 for (i=0; data && data->partitions && data->partitions[i]; i++) {
 330                         /* Find all partitions under the search base */
 331                         if (ldb_dn_compare_base(req->op.search.base, data->partitions[i]->dn) == 0) {
 332                                 ret = partition_send_request(ac, data->partitions[i]->module, data->partitions[i]->dn);
 333                                 if (ret != LDB_SUCCESS) {
 334                                         return ret;
 335                                 }
 336                         }
 337                 }
 338
 339                 /* Perhaps we didn't match any partitions.  Try the main partition, only */
 340                 if (ac->num_requests == 0) {
 341                         talloc_free(ac);
 342                         return ldb_next_request(module, req);
 343                 }
 344
 345                 return LDB_SUCCESS;
 346         } else {
 347                 struct ldb_module *backend = find_backend(module, req, req->op.search.base);
 348
 349                 return ldb_next_request(backend, req);
 350         }
 351 }
 352
 353 /* add */
 354 static int partition_add(struct ldb_module *module, struct ldb_request *req)
 355 {
 356         return partition_replicate(module, req, req->op.add.message->dn);
 357 }
 358
 359 /* modify */
 360 static int partition_modify(struct ldb_module *module, struct ldb_request *req)
 361 {
 362         return partition_replicate(module, req, req->op.mod.message->dn);
 363 }
 364
 365 /* delete */
 366 static int partition_delete(struct ldb_module *module, struct ldb_request *req)
 367 {
 368         return partition_replicate(module, req, req->op.del.dn);
 369 }
 370
 371 /* rename */
 372 static int partition_rename(struct ldb_module *module, struct ldb_request *req)
 373 {
 374         /* Find backend */
 375         struct ldb_module *backend = find_backend(module, req, req->op.rename.olddn);
 376         struct ldb_module *backend2 = find_backend(module, req, req->op.rename.newdn);
 377
 378         if (backend->next != backend2->next) {
 379                 return LDB_ERR_AFFECTS_MULTIPLE_DSAS;
 380         }
 381
 382         return partition_replicate(module, req, req->op.rename.olddn);
 383 }
 384
 385 /* start a transaction */
 386 static int partition_start_trans(struct ldb_module *module)
 387 {
 388         int i, ret;
 389         struct partition_private_data *data = talloc_get_type(module->private_data,
 390                                                               struct partition_private_data);
 391         /* Look at base DN */
 392         /* Figure out which partition it is under */
 393         /* Skip the lot if 'data' isn't here yet (initialistion) */
 394         ret = ldb_next_start_trans(module);
 395         if (ret != LDB_SUCCESS) {
 396                 return ret;
 397         }
 398
 399         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 400                 struct ldb_module *next = make_module_for_next_request(module, module->ldb, data->partitions[i]->module);
 401
 402                 ret = ldb_next_start_trans(next);
 403                 talloc_free(next);
 404                 if (ret != LDB_SUCCESS) {
 405                         /* Back it out, if it fails on one */
 406                         for (i--; i >= 0; i--) {
 407                                 next = make_module_for_next_request(module, module->ldb, data->partitions[i]->module);
 408                                 ldb_next_del_trans(next);
 409                                 talloc_free(next);
 410                         }
 411                         return ret;
 412                 }
 413         }
 414         return LDB_SUCCESS;
 415 }
 416
 417 /* end a transaction */
 418 static int partition_end_trans(struct ldb_module *module)
 419 {
 420         int i, ret, ret2 = LDB_SUCCESS;
 421         struct partition_private_data *data = talloc_get_type(module->private_data,
 422                                                               struct partition_private_data);
 423         ret = ldb_next_end_trans(module);
 424         if (ret != LDB_SUCCESS) {
 425                 return ret;
 426         }
 427
 428         /* Look at base DN */
 429         /* Figure out which partition it is under */
 430         /* Skip the lot if 'data' isn't here yet (initialistion) */
 431         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 432                 struct ldb_module *next = make_module_for_next_request(module, module->ldb, data->partitions[i]->module);
 433
 434                 ret = ldb_next_end_trans(next);
 435                 talloc_free(next);
 436                 if (ret != LDB_SUCCESS) {
 437                         ret2 = ret;
 438                 }
 439         }
 440
 441         if (ret != LDB_SUCCESS) {
 442                 /* Back it out, if it fails on one */
 443                 for (i=0; data && data->partitions && data->partitions[i]; i++) {
 444                         struct ldb_module *next = make_module_for_next_request(module, module->ldb, data->partitions[i]->module);
 445                         ldb_next_del_trans(next);
 446                         talloc_free(next);
 447                 }
 448         }
 449         return ret;
 450 }
 451
 452 /* delete a transaction */
 453 static int partition_del_trans(struct ldb_module *module)
 454 {
 455         int i, ret, ret2 = LDB_SUCCESS;
 456         struct partition_private_data *data = talloc_get_type(module->private_data,
 457                                                               struct partition_private_data);
 458         ret = ldb_next_del_trans(module);
 459         if (ret != LDB_SUCCESS) {
 460                 ret2 = ret;
 461         }
 462
 463         /* Look at base DN */
 464         /* Figure out which partition it is under */
 465         /* Skip the lot if 'data' isn't here yet (initialistion) */
 466         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 467                 struct ldb_module *next = make_module_for_next_request(module, module->ldb, data->partitions[i]->module);
 468
 469                 ret = ldb_next_del_trans(next);
 470                 talloc_free(next);
 471                 if (ret != LDB_SUCCESS) {
 472                         ret2 = ret;
 473                 }
 474         }
 475         return ret2;
 476 }
 477
 478 static int partition_sequence_number(struct ldb_module *module, struct ldb_request *req)
 479 {
 480         int i, ret;
 481         uint64_t seq_number = 0;
 482         uint64_t timestamp_sequence = 0;
 483         uint64_t timestamp = 0;
 484         struct partition_private_data *data = talloc_get_type(module->private_data,
 485                                                               struct partition_private_data);
 486
 487         switch (req->op.seq_num.type) {
 488         case LDB_SEQ_NEXT:
 489         case LDB_SEQ_HIGHEST_SEQ:
 490                 ret = ldb_next_request(module, req);
 491                 if (ret != LDB_SUCCESS) {
 492                         return ret;
 493                 }
 494                 if (req->op.seq_num.flags & LDB_SEQ_TIMESTAMP_SEQUENCE) {
 495                         timestamp_sequence = req->op.seq_num.seq_num;
 496                 } else {
 497                         seq_number = seq_number + req->op.seq_num.seq_num;
 498                 }
 499
 500                 /* Look at base DN */
 501                 /* Figure out which partition it is under */
 502                 /* Skip the lot if 'data' isn't here yet (initialistion) */
 503                 for (i=0; data && data->partitions && data->partitions[i]; i++) {
 504                         struct ldb_module *next = make_module_for_next_request(req, module->ldb, data->partitions[i]->module);
 505
 506                         ret = ldb_next_request(next, req);
 507                         talloc_free(next);
 508                         if (ret != LDB_SUCCESS) {
 509                                 return ret;
 510                         }
 511                         if (req->op.seq_num.flags & LDB_SEQ_TIMESTAMP_SEQUENCE) {
 512                                 timestamp_sequence = MAX(timestamp_sequence, req->op.seq_num.seq_num);
 513                         } else {
 514                                 seq_number = seq_number + req->op.seq_num.seq_num;
 515                         }
 516                 }
 517                 /* fall though */
 518         case LDB_SEQ_HIGHEST_TIMESTAMP:
 519         {
 520                 struct ldb_request *date_req = talloc(req, struct ldb_request);
 521                 if (!date_req) {
 522                         return LDB_ERR_OPERATIONS_ERROR;
 523                 }
 524                 *date_req = *req;
 525                 date_req->op.seq_num.flags = LDB_SEQ_HIGHEST_TIMESTAMP;
 526
 527                 ret = ldb_next_request(module, date_req);
 528                 if (ret != LDB_SUCCESS) {
 529                         return ret;
 530                 }
 531                 timestamp = date_req->op.seq_num.seq_num;
 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(req, module->ldb, data->partitions[i]->module);
 538
 539                         ret = ldb_next_request(next, date_req);
 540                         talloc_free(next);
 541                         if (ret != LDB_SUCCESS) {
 542                                 return ret;
 543                         }
 544                         timestamp = MAX(timestamp, date_req->op.seq_num.seq_num);
 545                 }
 546                 break;
 547         }
 548         }
 549
 550         switch (req->op.seq_num.flags) {
 551         case LDB_SEQ_NEXT:
 552         case LDB_SEQ_HIGHEST_SEQ:
 553
 554                 req->op.seq_num.flags = 0;
 555
 556                 /* Has someone above set a timebase sequence? */
 557                 if (timestamp_sequence) {
 558                         req->op.seq_num.seq_num = (((unsigned long long)timestamp << 24) | (seq_number & 0xFFFFFF));
 559                 } else {
 560                         req->op.seq_num.seq_num = seq_number;
 561                 }
 562
 563                 if (timestamp_sequence > req->op.seq_num.seq_num) {
 564                         req->op.seq_num.seq_num = timestamp_sequence;
 565                         req->op.seq_num.flags |= LDB_SEQ_TIMESTAMP_SEQUENCE;
 566                 }
 567
 568                 req->op.seq_num.flags |= LDB_SEQ_GLOBAL_SEQUENCE;
 569                 break;
 570         case LDB_SEQ_HIGHEST_TIMESTAMP:
 571                 req->op.seq_num.seq_num = timestamp;
 572                 break;
 573         }
 574
 575         switch (req->op.seq_num.flags) {
 576         case LDB_SEQ_NEXT:
 577                 req->op.seq_num.seq_num++;
 578         }
 579         return LDB_SUCCESS;
 580 }
 581
 582 static int partition_extended_replicated_objects(struct ldb_module *module, struct ldb_request *req)
 583 {
 584         struct dsdb_extended_replicated_objects *ext;
 585
 586         ext = talloc_get_type(req->op.extended.data, struct dsdb_extended_replicated_objects);
 587         if (!ext) {
 588                 ldb_debug(module->ldb, LDB_DEBUG_FATAL, "partition_extended_replicated_objects: invalid extended data\n");
 589                 return LDB_ERR_PROTOCOL_ERROR;
 590         }
 591
 592         if (ext->version != DSDB_EXTENDED_REPLICATED_OBJECTS_VERSION) {
 593                 ldb_debug(module->ldb, LDB_DEBUG_FATAL, "partition_extended_replicated_objects: extended data invalid version [%u != %u]\n",
 594                           ext->version, DSDB_EXTENDED_REPLICATED_OBJECTS_VERSION);
 595                 return LDB_ERR_PROTOCOL_ERROR;
 596         }
 597
 598         return partition_replicate(module, req, ext->partition_dn);
 599 }
 600
 601 /* extended */
 602 static int partition_extended(struct ldb_module *module, struct ldb_request *req)
 603 {
 604         struct partition_context *ac;
 605
 606         if (strcmp(req->op.extended.oid, DSDB_EXTENDED_REPLICATED_OBJECTS_OID) == 0) {
 607                 return partition_extended_replicated_objects(module, req);
 608         }
 609
 610         /*
 611          * as the extended operation has no dn
 612          * we need to send it to all partitions
 613          */
 614
 615         ac = partition_init_handle(req, module);
 616         if (!ac) {
 617                 return LDB_ERR_OPERATIONS_ERROR;
 618         }
 619
 620         return partition_send_all(module, ac, req);
 621 }
 622
 623 static int sort_compare(void *void1,
 624                         void *void2, void *opaque)
 625 {
 626         struct dsdb_control_current_partition **pp1 = void1;
 627         struct dsdb_control_current_partition **pp2 = void2;
 628         struct dsdb_control_current_partition *partition1 = talloc_get_type(*pp1,
 629                                                             struct dsdb_control_current_partition);
 630         struct dsdb_control_current_partition *partition2 = talloc_get_type(*pp2,
 631                                                             struct dsdb_control_current_partition);
 632
 633         return ldb_dn_compare(partition1->dn, partition2->dn);
 634 }
 635
 636 static int partition_init(struct ldb_module *module)
 637 {
 638         int ret, i;
 639         TALLOC_CTX *mem_ctx = talloc_new(module);
 640         static const char *attrs[] = { "partition", "replicateEntries", "modules", NULL };
 641         struct ldb_result *res;
 642         struct ldb_message *msg;
 643         struct ldb_message_element *partition_attributes;
 644         struct ldb_message_element *replicate_attributes;
 645         struct ldb_message_element *modules_attributes;
 646
 647         struct partition_private_data *data;
 648
 649         if (!mem_ctx) {
 650                 return LDB_ERR_OPERATIONS_ERROR;
 651         }
 652
 653         data = talloc(mem_ctx, struct partition_private_data);
 654         if (data == NULL) {
 655                 return LDB_ERR_OPERATIONS_ERROR;
 656         }
 657
 658         ret = ldb_search(module->ldb, ldb_dn_new(mem_ctx, module->ldb, "@PARTITION"),
 659                          LDB_SCOPE_BASE,
 660                          NULL, attrs,
 661                          &res);
 662         if (ret != LDB_SUCCESS) {
 663                 talloc_free(mem_ctx);
 664                 return ret;
 665         }
 666         talloc_steal(mem_ctx, res);
 667         if (res->count == 0) {
 668                 talloc_free(mem_ctx);
 669                 return ldb_next_init(module);
 670         }
 671
 672         if (res->count > 1) {
 673                 talloc_free(mem_ctx);
 674                 return LDB_ERR_CONSTRAINT_VIOLATION;
 675         }
 676
 677         msg = res->msgs[0];
 678
 679         partition_attributes = ldb_msg_find_element(msg, "partition");
 680         if (!partition_attributes) {
 681                 ldb_set_errstring(module->ldb, "partition_init: no partitions specified");
 682                 talloc_free(mem_ctx);
 683                 return LDB_ERR_CONSTRAINT_VIOLATION;
 684         }
 685         data->partitions = talloc_array(data, struct dsdb_control_current_partition *, partition_attributes->num_values + 1);
 686         if (!data->partitions) {
 687                 talloc_free(mem_ctx);
 688                 return LDB_ERR_OPERATIONS_ERROR;
 689         }
 690         for (i=0; i < partition_attributes->num_values; i++) {
 691                 char *base = talloc_strdup(data->partitions, (char *)partition_attributes->values[i].data);
 692                 char *p = strchr(base, ':');
 693                 if (!p) {
 694                         ldb_asprintf_errstring(module->ldb,
 695                                                 "partition_init: "
 696                                                 "invalid form for partition record (missing ':'): %s", base);
 697                         talloc_free(mem_ctx);
 698                         return LDB_ERR_CONSTRAINT_VIOLATION;
 699                 }
 700                 p[0] = '\0';
 701                 p++;
 702                 if (!p[0]) {
 703                         ldb_asprintf_errstring(module->ldb,
 704                                                 "partition_init: "
 705                                                 "invalid form for partition record (missing backend database): %s", base);
 706                         talloc_free(mem_ctx);
 707                         return LDB_ERR_CONSTRAINT_VIOLATION;
 708                 }
 709                 data->partitions[i] = talloc(data->partitions, struct dsdb_control_current_partition);
 710                 if (!data->partitions[i]) {
 711                         talloc_free(mem_ctx);
 712                         return LDB_ERR_OPERATIONS_ERROR;
 713                 }
 714                 data->partitions[i]->version = DSDB_CONTROL_CURRENT_PARTITION_VERSION;
 715
 716                 data->partitions[i]->dn = ldb_dn_new(data->partitions[i], module->ldb, base);
 717                 if (!data->partitions[i]->dn) {
 718                         ldb_asprintf_errstring(module->ldb,
 719                                                 "partition_init: invalid DN in partition record: %s", base);
 720                         talloc_free(mem_ctx);
 721                         return LDB_ERR_CONSTRAINT_VIOLATION;
 722                 }
 723
 724                 data->partitions[i]->backend = private_path(data->partitions[i], p);
 725                 ret = ldb_connect_backend(module->ldb, data->partitions[i]->backend, NULL, &data->partitions[i]->module);
 726                 if (ret != LDB_SUCCESS) {
 727                         talloc_free(mem_ctx);
 728                         return ret;
 729                 }
 730         }
 731         data->partitions[i] = NULL;
 732
 733         /* sort these into order, most to least specific */
 734         ldb_qsort(data->partitions, partition_attributes->num_values, sizeof(*data->partitions),
 735                   module->ldb, sort_compare);
 736
 737         for (i=0; data->partitions[i]; i++) {
 738                 struct ldb_request *req;
 739                 req = talloc_zero(mem_ctx, struct ldb_request);
 740                 if (req == NULL) {
 741                         ldb_debug(module->ldb, LDB_DEBUG_ERROR, "partition: Out of memory!\n");
 742                         talloc_free(mem_ctx);
 743                         return LDB_ERR_OPERATIONS_ERROR;
 744                 }
 745
 746                 req->operation = LDB_REQ_REGISTER_PARTITION;
 747                 req->op.reg_partition.dn = data->partitions[i]->dn;
 748
 749                 ret = ldb_request(module->ldb, req);
 750                 if (ret != LDB_SUCCESS) {
 751                         ldb_debug(module->ldb, LDB_DEBUG_ERROR, "partition: Unable to register partition with rootdse!\n");
 752                         talloc_free(mem_ctx);
 753                         return LDB_ERR_OTHER;
 754                 }
 755                 talloc_free(req);
 756         }
 757
 758         replicate_attributes = ldb_msg_find_element(msg, "replicateEntries");
 759         if (!replicate_attributes) {
 760                 data->replicate = NULL;
 761         } else {
 762                 data->replicate = talloc_array(data, struct ldb_dn *, replicate_attributes->num_values + 1);
 763                 if (!data->replicate) {
 764                         talloc_free(mem_ctx);
 765                         return LDB_ERR_OPERATIONS_ERROR;
 766                 }
 767
 768                 for (i=0; i < replicate_attributes->num_values; i++) {
 769                         data->replicate[i] = ldb_dn_new(data->replicate, module->ldb, (const char *)replicate_attributes->values[i].data);
 770                         if (!ldb_dn_validate(data->replicate[i])) {
 771                                 ldb_asprintf_errstring(module->ldb,
 772                                                         "partition_init: "
 773                                                         "invalid DN in partition replicate record: %s",
 774                                                         replicate_attributes->values[i].data);
 775                                 talloc_free(mem_ctx);
 776                                 return LDB_ERR_CONSTRAINT_VIOLATION;
 777                         }
 778                 }
 779                 data->replicate[i] = NULL;
 780         }
 781
 782         /* Make the private data available to any searches the modules may trigger in initialisation */
 783         module->private_data = data;
 784         talloc_steal(module, data);
 785
 786         modules_attributes = ldb_msg_find_element(msg, "modules");
 787         if (modules_attributes) {
 788                 for (i=0; i < modules_attributes->num_values; i++) {
 789                         struct ldb_dn *base_dn;
 790                         int partition_idx;
 791                         struct dsdb_control_current_partition *partition = NULL;
 792                         const char **modules = NULL;
 793
 794                         char *base = talloc_strdup(data->partitions, (char *)modules_attributes->values[i].data);
 795                         char *p = strchr(base, ':');
 796                         if (!p) {
 797                                 ldb_asprintf_errstring(module->ldb,
 798                                                         "partition_init: "
 799                                                         "invalid form for partition module record (missing ':'): %s", base);
 800                                 talloc_free(mem_ctx);
 801                                 return LDB_ERR_CONSTRAINT_VIOLATION;
 802                         }
 803                         p[0] = '\0';
 804                         p++;
 805                         if (!p[0]) {
 806                                 ldb_asprintf_errstring(module->ldb,
 807                                                         "partition_init: "
 808                                                         "invalid form for partition module record (missing backend database): %s", base);
 809                                 talloc_free(mem_ctx);
 810                                 return LDB_ERR_CONSTRAINT_VIOLATION;
 811                         }
 812
 813                         modules = ldb_modules_list_from_string(module->ldb, mem_ctx,
 814                                                                p);
 815
 816                         base_dn = ldb_dn_new(mem_ctx, module->ldb, base);
 817                         if (!ldb_dn_validate(base_dn)) {
 818                                 talloc_free(mem_ctx);
 819                                 return LDB_ERR_OPERATIONS_ERROR;
 820                         }
 821
 822                         for (partition_idx = 0; data->partitions[partition_idx]; partition_idx++) {
 823                                 if (ldb_dn_compare(data->partitions[partition_idx]->dn, base_dn) == 0) {
 824                                         partition = data->partitions[partition_idx];
 825                                         break;
 826                                 }
 827                         }
 828
 829                         if (!partition) {
 830                                 ldb_asprintf_errstring(module->ldb,
 831                                                         "partition_init: "
 832                                                         "invalid form for partition module record (no such partition): %s", base);
 833                                 talloc_free(mem_ctx);
 834                                 return LDB_ERR_CONSTRAINT_VIOLATION;
 835                         }
 836
 837                         ret = ldb_load_modules_list(module->ldb, modules, partition->module, &partition->module);
 838                         if (ret != LDB_SUCCESS) {
 839                                 ldb_asprintf_errstring(module->ldb,
 840                                                        "partition_init: "
 841                                                        "loading backend for %s failed: %s",
 842                                                        base, ldb_errstring(module->ldb));
 843                                 talloc_free(mem_ctx);
 844                                 return ret;
 845                         }
 846                         ret = ldb_init_module_chain(module->ldb, partition->module);
 847                         if (ret != LDB_SUCCESS) {
 848                                 ldb_asprintf_errstring(module->ldb,
 849                                                        "partition_init: "
 850                                                        "initialising backend for %s failed: %s",
 851                                                        base, ldb_errstring(module->ldb));
 852                                 talloc_free(mem_ctx);
 853                                 return ret;
 854                         }
 855                 }
 856         }
 857
 858         talloc_free(mem_ctx);
 859         return ldb_next_init(module);
 860 }
 861
 862 static int partition_wait_none(struct ldb_handle *handle) {
 863         struct partition_context *ac;
 864         int ret;
 865         int i;
 866
 867         if (!handle || !handle->private_data) {
 868                 return LDB_ERR_OPERATIONS_ERROR;
 869         }
 870
 871         if (handle->state == LDB_ASYNC_DONE) {
 872                 return handle->status;
 873         }
 874
 875         handle->state = LDB_ASYNC_PENDING;
 876         handle->status = LDB_SUCCESS;
 877
 878         ac = talloc_get_type(handle->private_data, struct partition_context);
 879
 880         for (i=0; i < ac->num_requests; i++) {
 881                 ret = ldb_wait(ac->down_req[i]->handle, LDB_WAIT_NONE);
 882
 883                 if (ret != LDB_SUCCESS) {
 884                         handle->status = ret;
 885                         goto done;
 886                 }
 887                 if (ac->down_req[i]->handle->status != LDB_SUCCESS) {
 888                         handle->status = ac->down_req[i]->handle->status;
 889                         goto done;
 890                 }
 891
 892                 if (ac->down_req[i]->handle->state != LDB_ASYNC_DONE) {
 893                         return LDB_SUCCESS;
 894                 }
 895         }
 896
 897         ret = LDB_SUCCESS;
 898
 899 done:
 900         handle->state = LDB_ASYNC_DONE;
 901         return ret;
 902 }
 903
 904
 905 static int partition_wait_all(struct ldb_handle *handle) {
 906
 907         int ret;
 908
 909         while (handle->state != LDB_ASYNC_DONE) {
 910                 ret = partition_wait_none(handle);
 911                 if (ret != LDB_SUCCESS) {
 912                         return ret;
 913                 }
 914         }
 915
 916         return handle->status;
 917 }
 918
 919 static int partition_wait(struct ldb_handle *handle, enum ldb_wait_type type)
 920 {
 921         if (type == LDB_WAIT_ALL) {
 922                 return partition_wait_all(handle);
 923         } else {
 924                 return partition_wait_none(handle);
 925         }
 926 }
 927
 928 static const struct ldb_module_ops partition_ops = {
 929         .name              = "partition",
 930         .init_context      = partition_init,
 931         .search            = partition_search,
 932         .add               = partition_add,
 933         .modify            = partition_modify,
 934         .del               = partition_delete,
 935         .rename            = partition_rename,
 936         .extended          = partition_extended,
 937         .sequence_number   = partition_sequence_number,
 938         .start_transaction = partition_start_trans,
 939         .end_transaction   = partition_end_trans,
 940         .del_transaction   = partition_del_trans,
 941         .wait              = partition_wait
 942 };
 943
 944 int ldb_partition_init(void)
 945 {
 946         return ldb_register_module(&partition_ops);
 947 }