source4/dsdb/samdb/ldb_modules/partition_init.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    This program is free software; you can redistribute it and/or modify
   8    it under the terms of the GNU General Public License as published by
   9    the Free Software Foundation; either version 3 of the License, or
  10    (at your option) any later version.
  11
  12    This program is distributed in the hope that it will be useful,
  13    but WITHOUT ANY WARRANTY; without even the implied warranty of
  14    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15    GNU General Public License for more details.
  16
  17    You should have received a copy of the GNU General Public License
  18    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19 */
  20
  21 /*
  22  *  Name: ldb
  23  *
  24  *  Component: ldb partitions module
  25  *
  26  *  Description: Implement LDAP partitions
  27  *
  28  *  Author: Andrew Bartlett
  29  *  Author: Stefan Metzmacher
  30  */
  31
  32 #include "dsdb/samdb/ldb_modules/partition.h"
  33 #include "lib/util/tsort.h"
  34 #include "lib/ldb-samba/ldb_wrap.h"
  35 #include "system/filesys.h"
  36
  37 static int partition_sort_compare(const void *v1, const void *v2)
  38 {
  39         const struct dsdb_partition *p1;
  40         const struct dsdb_partition *p2;
  41
  42         p1 = *((struct dsdb_partition * const*)v1);
  43         p2 = *((struct dsdb_partition * const*)v2);
  44
  45         return ldb_dn_compare(p1->ctrl->dn, p2->ctrl->dn);
  46 }
  47
  48 /* Load the list of DNs that we must replicate to all partitions */
  49 static int partition_load_replicate_dns(struct ldb_context *ldb,
  50                                         struct partition_private_data *data,
  51                                         struct ldb_message *msg)
  52 {
  53         struct ldb_message_element *replicate_attributes = ldb_msg_find_element(msg, "replicateEntries");
  54
  55         talloc_free(data->replicate);
  56         if (!replicate_attributes) {
  57                 data->replicate = NULL;
  58         } else {
  59                 unsigned int i;
  60                 data->replicate = talloc_array(data, struct ldb_dn *, replicate_attributes->num_values + 1);
  61                 if (!data->replicate) {
  62                         return ldb_oom(ldb);
  63                 }
  64
  65                 for (i=0; i < replicate_attributes->num_values; i++) {
  66                         data->replicate[i] = ldb_dn_from_ldb_val(data->replicate, ldb, &replicate_attributes->values[i]);
  67                         if (!ldb_dn_validate(data->replicate[i])) {
  68                                 ldb_asprintf_errstring(ldb,
  69                                                         "partition_init: "
  70                                                         "invalid DN in partition replicate record: %s",
  71                                                         replicate_attributes->values[i].data);
  72                                 return LDB_ERR_CONSTRAINT_VIOLATION;
  73                         }
  74                 }
  75                 data->replicate[i] = NULL;
  76         }
  77         return LDB_SUCCESS;
  78 }
  79
  80 /* Load the list of modules for the partitions */
  81 static int partition_load_modules(struct ldb_context *ldb,
  82                                   struct partition_private_data *data, struct ldb_message *msg)
  83 {
  84         unsigned int i;
  85         struct ldb_message_element *modules_attributes = ldb_msg_find_element(msg, "modules");
  86         talloc_free(data->modules);
  87         if (!modules_attributes) {
  88                 return LDB_SUCCESS;
  89         }
  90
  91         data->modules = talloc_array(data, struct partition_module *, modules_attributes->num_values + 1);
  92         if (!data->modules) {
  93                 return ldb_oom(ldb);
  94         }
  95
  96         for (i=0; i < modules_attributes->num_values; i++) {
  97                 char *p;
  98                 DATA_BLOB dn_blob;
  99                 data->modules[i] = talloc(data->modules, struct partition_module);
 100                 if (!data->modules[i]) {
 101                         return ldb_oom(ldb);
 102                 }
 103
 104                 dn_blob = modules_attributes->values[i];
 105
 106                 p = strchr((const char *)dn_blob.data, ':');
 107                 if (!p) {
 108                         ldb_asprintf_errstring(ldb,
 109                                                "partition_load_modules: "
 110                                                "invalid form for partition module record (missing ':'): %s", (const char *)dn_blob.data);
 111                         return LDB_ERR_CONSTRAINT_VIOLATION;
 112                 }
 113                 /* Now trim off the filename */
 114                 dn_blob.length = ((uint8_t *)p - dn_blob.data);
 115
 116                 p++;
 117                 data->modules[i]->modules = ldb_modules_list_from_string(ldb, data->modules[i],
 118                                                                          p);
 119
 120                 if (dn_blob.length == 1 && dn_blob.data[0] == '*') {
 121                         data->modules[i]->dn = NULL;
 122                 } else {
 123                         data->modules[i]->dn = ldb_dn_from_ldb_val(data->modules[i], ldb, &dn_blob);
 124                         if (!data->modules[i]->dn || !ldb_dn_validate(data->modules[i]->dn)) {
 125                                 return ldb_operr(ldb);
 126                         }
 127                 }
 128         }
 129         data->modules[i] = NULL;
 130         return LDB_SUCCESS;
 131 }
 132
 133 static int partition_reload_metadata(struct ldb_module *module, struct partition_private_data *data,
 134                                      TALLOC_CTX *mem_ctx, struct ldb_message **_msg,
 135                                      struct ldb_request *parent)
 136 {
 137         int ret;
 138         struct ldb_message *msg, *module_msg;
 139         struct ldb_result *res;
 140         struct ldb_context *ldb = ldb_module_get_ctx(module);
 141         const char *attrs[] = { "partition", "replicateEntries", "modules", "ldapBackend",
 142                                 "partialReplica", NULL };
 143         /* perform search for @PARTITION, looking for module, replicateEntries and ldapBackend */
 144         ret = dsdb_module_search_dn(module, mem_ctx, &res,
 145                                     ldb_dn_new(mem_ctx, ldb, DSDB_PARTITION_DN),
 146                                     attrs,
 147                                     DSDB_FLAG_NEXT_MODULE, parent);
 148         if (ret != LDB_SUCCESS) {
 149                 return ret;
 150         }
 151
 152         msg = res->msgs[0];
 153
 154         ret = partition_load_replicate_dns(ldb, data, msg);
 155         if (ret != LDB_SUCCESS) {
 156                 return ret;
 157         }
 158
 159         /* When used from Samba4, this message is set by the samba4
 160          * module, as a fixed value not read from the DB.  This avoids
 161          * listing modules in the DB */
 162         if (data->forced_module_msg) {
 163                 module_msg = data->forced_module_msg;
 164         } else {
 165                 module_msg = msg;
 166         }
 167
 168         ret = partition_load_modules(ldb, data, module_msg);
 169         if (ret != LDB_SUCCESS) {
 170                 return ret;
 171         }
 172
 173         data->ldapBackend = talloc_steal(data, ldb_msg_find_attr_as_string(msg, "ldapBackend", NULL));
 174         if (_msg) {
 175                 *_msg = msg;
 176         } else {
 177                 talloc_free(msg);
 178         }
 179
 180         return LDB_SUCCESS;
 181 }
 182
 183 static const char **find_modules_for_dn(struct partition_private_data *data, struct ldb_dn *dn)
 184 {
 185         unsigned int i;
 186         struct partition_module *default_mod = NULL;
 187         for (i=0; data->modules && data->modules[i]; i++) {
 188                 if (!data->modules[i]->dn) {
 189                         default_mod = data->modules[i];
 190                 } else if (ldb_dn_compare(dn, data->modules[i]->dn) == 0) {
 191                         return data->modules[i]->modules;
 192                 }
 193         }
 194         if (default_mod) {
 195                 return default_mod->modules;
 196         } else {
 197                 return NULL;
 198         }
 199 }
 200
 201 static int new_partition_from_dn(struct ldb_context *ldb, struct partition_private_data *data,
 202                                  TALLOC_CTX *mem_ctx,
 203                                  struct ldb_dn *dn, const char *filename,
 204                                  struct dsdb_partition **partition) {
 205         const char *backend_url;
 206         struct dsdb_control_current_partition *ctrl;
 207         struct ldb_module *backend_module;
 208         struct ldb_module *module_chain;
 209         const char **modules;
 210         int ret;
 211
 212         (*partition) = talloc_zero(mem_ctx, struct dsdb_partition);
 213         if (!*partition) {
 214                 return ldb_oom(ldb);
 215         }
 216
 217         (*partition)->ctrl = ctrl = talloc((*partition), struct dsdb_control_current_partition);
 218         if (!ctrl) {
 219                 talloc_free(*partition);
 220                 return ldb_oom(ldb);
 221         }
 222
 223         /* See if an LDAP backend has been specified */
 224         if (data->ldapBackend) {
 225                 (*partition)->backend_url = data->ldapBackend;
 226         } else {
 227                 /* the backend LDB is the DN (base64 encoded if not 'plain') followed by .ldb */
 228                 backend_url = ldb_relative_path(ldb,
 229                                                   *partition,
 230                                                   filename);
 231                 if (!backend_url) {
 232                         ldb_asprintf_errstring(ldb,
 233                                                "partition_init: unable to determine an relative path for partition: %s", filename);
 234                         talloc_free(*partition);
 235                         return LDB_ERR_OPERATIONS_ERROR;
 236                 }
 237                 (*partition)->backend_url = talloc_steal((*partition), backend_url);
 238
 239                 if (!(ldb_module_flags(ldb) & LDB_FLG_RDONLY)) {
 240                         char *p;
 241                         char *backend_dir = talloc_strdup(*partition, backend_url);
 242
 243                         p = strrchr(backend_dir, '/');
 244                         if (p) {
 245                                 p[0] = '\0';
 246                         }
 247
 248                         /* Failure is quite reasonable, it might alredy exist */
 249                         mkdir(backend_dir, 0700);
 250                         talloc_free(backend_dir);
 251                 }
 252
 253         }
 254
 255         ctrl->version = DSDB_CONTROL_CURRENT_PARTITION_VERSION;
 256         ctrl->dn = talloc_steal(ctrl, dn);
 257
 258         ret = ldb_module_connect_backend(ldb, (*partition)->backend_url, NULL, &backend_module);
 259         if (ret != LDB_SUCCESS) {
 260                 return ret;
 261         }
 262         talloc_steal((*partition), backend_module);
 263
 264         modules = find_modules_for_dn(data, dn);
 265
 266         if (!modules) {
 267                 DEBUG(0, ("Unable to load partition modules for new DN %s, perhaps you need to reprovision?  See partition-upgrade.txt for instructions\n", ldb_dn_get_linearized(dn)));
 268                 talloc_free(*partition);
 269                 return LDB_ERR_CONSTRAINT_VIOLATION;
 270         }
 271         ret = ldb_module_load_list(ldb, modules, backend_module, &module_chain);
 272         if (ret != LDB_SUCCESS) {
 273                 ldb_asprintf_errstring(ldb,
 274                                        "partition_init: "
 275                                        "loading backend for %s failed: %s",
 276                                        ldb_dn_get_linearized(dn), ldb_errstring(ldb));
 277                 talloc_free(*partition);
 278                 return ret;
 279         }
 280         ret = ldb_module_init_chain(ldb, module_chain);
 281         if (ret != LDB_SUCCESS) {
 282                 ldb_asprintf_errstring(ldb,
 283                                        "partition_init: "
 284                                        "initialising backend for %s failed: %s",
 285                                        ldb_dn_get_linearized(dn), ldb_errstring(ldb));
 286                 talloc_free(*partition);
 287                 return ret;
 288         }
 289
 290         /* This weirdness allows us to use ldb_next_request() in partition.c */
 291         (*partition)->module = ldb_module_new(*partition, ldb, "partition_next", NULL);
 292         if (!(*partition)->module) {
 293                 talloc_free(*partition);
 294                 return ldb_oom(ldb);
 295         }
 296         ldb_module_set_next((*partition)->module, talloc_steal((*partition)->module, module_chain));
 297
 298         /* if we were in a transaction then we need to start a
 299            transaction on this new partition, otherwise we'll get a
 300            transaction mismatch when we end the transaction */
 301         if (data->in_transaction) {
 302                 if (ldb_module_flags(ldb) & LDB_FLG_ENABLE_TRACING) {
 303                         ldb_debug(ldb, LDB_DEBUG_TRACE, "partition_start_trans() -> %s (new partition)",
 304                                   ldb_dn_get_linearized((*partition)->ctrl->dn));
 305                 }
 306                 ret = ldb_next_start_trans((*partition)->module);
 307         }
 308
 309         return ret;
 310 }
 311
 312 /* Tell the rootDSE about the new partition */
 313 static int partition_register(struct ldb_context *ldb, struct dsdb_control_current_partition *ctrl)
 314 {
 315         struct ldb_request *req;
 316         int ret;
 317
 318         req = talloc_zero(NULL, struct ldb_request);
 319         if (req == NULL) {
 320                 return ldb_oom(ldb);
 321         }
 322
 323         req->operation = LDB_REQ_REGISTER_PARTITION;
 324         req->op.reg_partition.dn = ctrl->dn;
 325         req->callback = ldb_op_default_callback;
 326
 327         ldb_set_timeout(ldb, req, 0);
 328
 329         req->handle = ldb_handle_new(req, ldb);
 330         if (req->handle == NULL) {
 331                 talloc_free(req);
 332                 return ldb_operr(ldb);
 333         }
 334
 335         ret = ldb_request(ldb, req);
 336         if (ret == LDB_SUCCESS) {
 337                 ret = ldb_wait(req->handle, LDB_WAIT_ALL);
 338         }
 339         if (ret != LDB_SUCCESS) {
 340                 ldb_debug(ldb, LDB_DEBUG_ERROR, "partition: Unable to register partition with rootdse!\n");
 341                 talloc_free(req);
 342                 return LDB_ERR_OTHER;
 343         }
 344         talloc_free(req);
 345
 346         return LDB_SUCCESS;
 347 }
 348
 349 /* Add a newly found partition to the global data */
 350 static int add_partition_to_data(struct ldb_context *ldb, struct partition_private_data *data,
 351                                  struct dsdb_partition *partition)
 352 {
 353         unsigned int i;
 354         int ret;
 355
 356         /* Count the partitions */
 357         for (i=0; data->partitions && data->partitions[i]; i++) { /* noop */};
 358
 359         /* Add partition to list of partitions */
 360         data->partitions = talloc_realloc(data, data->partitions, struct dsdb_partition *, i + 2);
 361         if (!data->partitions) {
 362                 return ldb_oom(ldb);
 363         }
 364         data->partitions[i] = talloc_steal(data->partitions, partition);
 365         data->partitions[i+1] = NULL;
 366
 367         /* Sort again (should use binary insert) */
 368         TYPESAFE_QSORT(data->partitions, i+1, partition_sort_compare);
 369
 370         ret = partition_register(ldb, partition->ctrl);
 371         if (ret != LDB_SUCCESS) {
 372                 return ret;
 373         }
 374         return LDB_SUCCESS;
 375 }
 376
 377 int partition_reload_if_required(struct ldb_module *module,
 378                                  struct partition_private_data *data,
 379                                  struct ldb_request *parent)
 380 {
 381         uint64_t seq;
 382         int ret;
 383         unsigned int i;
 384         struct ldb_context *ldb = ldb_module_get_ctx(module);
 385         struct ldb_message *msg;
 386         struct ldb_message_element *partition_attributes;
 387         struct ldb_message_element *partial_replicas;
 388         TALLOC_CTX *mem_ctx;
 389
 390         if (!data) {
 391                 /* Not initialised yet */
 392                 return LDB_SUCCESS;
 393         }
 394
 395         mem_ctx = talloc_new(data);
 396         if (!mem_ctx) {
 397                 return ldb_oom(ldb);
 398         }
 399
 400         ret = partition_primary_sequence_number(module, mem_ctx, LDB_SEQ_HIGHEST_SEQ, &seq);
 401         if (ret != LDB_SUCCESS) {
 402                 talloc_free(mem_ctx);
 403                 return ret;
 404         }
 405         if (seq == data->metadata_seq) {
 406                 talloc_free(mem_ctx);
 407                 return LDB_SUCCESS;
 408         }
 409
 410         ret = partition_reload_metadata(module, data, mem_ctx, &msg, parent);
 411         if (ret != LDB_SUCCESS) {
 412                 talloc_free(mem_ctx);
 413                 return ret;
 414         }
 415
 416         data->metadata_seq = seq;
 417
 418         partition_attributes = ldb_msg_find_element(msg, "partition");
 419         partial_replicas     = ldb_msg_find_element(msg, "partialReplica");
 420
 421         for (i=0; partition_attributes && i < partition_attributes->num_values; i++) {
 422                 unsigned int j;
 423                 bool new_partition = true;
 424                 const char *filename = NULL;
 425                 DATA_BLOB dn_blob;
 426                 struct ldb_dn *dn;
 427                 struct dsdb_partition *partition;
 428                 struct ldb_result *dn_res;
 429                 const char *no_attrs[] = { NULL };
 430
 431                 for (j=0; data->partitions && data->partitions[j]; j++) {
 432                         if (data_blob_cmp(&data->partitions[j]->orig_record, &partition_attributes->values[i]) == 0) {
 433                                 new_partition = false;
 434                                 break;
 435                         }
 436                 }
 437                 if (new_partition == false) {
 438                         continue;
 439                 }
 440
 441                 dn_blob = partition_attributes->values[i];
 442
 443                 if (dn_blob.length > 4 &&
 444                     (strncmp((const char *)&dn_blob.data[dn_blob.length-4], ".ldb", 4) == 0)) {
 445
 446                         /* Look for DN:filename.ldb */
 447                         char *p = strrchr((const char *)dn_blob.data, ':');
 448                         if (!p) {
 449                                 ldb_asprintf_errstring(ldb,
 450                                                        "partition_init: invalid DN in attempting to parse partition record: %s", (const char *)dn_blob.data);
 451                                 talloc_free(mem_ctx);
 452                                 return LDB_ERR_CONSTRAINT_VIOLATION;
 453                         }
 454                         filename = p+1;
 455
 456                         /* Now trim off the filename */
 457                         dn_blob.length = ((uint8_t *)p - dn_blob.data);
 458                 }
 459
 460                 dn = ldb_dn_from_ldb_val(mem_ctx, ldb, &dn_blob);
 461                 if (!dn) {
 462                         ldb_asprintf_errstring(ldb,
 463                                                "partition_init: invalid DN in partition record: %s", (const char *)dn_blob.data);
 464                         talloc_free(mem_ctx);
 465                         return LDB_ERR_CONSTRAINT_VIOLATION;
 466                 }
 467
 468                 /* Now do a slow check with the DN compare */
 469                 for (j=0; data->partitions && data->partitions[j]; j++) {
 470                         if (ldb_dn_compare(dn, data->partitions[j]->ctrl->dn) == 0) {
 471                                 new_partition = false;
 472                                 break;
 473                         }
 474                 }
 475                 if (new_partition == false) {
 476                         continue;
 477                 }
 478
 479                 if (!filename) {
 480                         char *base64_dn = NULL;
 481                         const char *p;
 482                         for (p = ldb_dn_get_linearized(dn); *p; p++) {
 483                                 /* We have such a strict check because I don't want shell metacharacters in the file name, nor ../ */
 484                                 if (!(isalnum(*p) || *p == ' ' || *p == '=' || *p == ',')) {
 485                                         break;
 486                                 }
 487                         }
 488                         if (*p) {
 489                                 base64_dn = ldb_base64_encode(data, ldb_dn_get_linearized(dn), strlen(ldb_dn_get_linearized(dn)));
 490                                 filename = talloc_asprintf(mem_ctx, "%s.ldb", base64_dn);
 491                         } else {
 492                                 filename = talloc_asprintf(mem_ctx, "%s.ldb", ldb_dn_get_linearized(dn));
 493                         }
 494                 }
 495
 496                 /* We call ldb_dn_get_linearized() because the DN in
 497                  * partition_attributes is already casefolded
 498                  * correctly.  We don't want to mess that up as the
 499                  * schema isn't loaded yet */
 500                 ret = new_partition_from_dn(ldb, data, data->partitions, dn,
 501                                             filename,
 502                                             &partition);
 503                 if (ret != LDB_SUCCESS) {
 504                         talloc_free(mem_ctx);
 505                         return ret;
 506                 }
 507
 508                 talloc_steal(partition, partition_attributes->values[i].data);
 509                 partition->orig_record = partition_attributes->values[i];
 510
 511                 /* Get the 'correct' case of the partition DNs from the database */
 512                 ret = dsdb_module_search_dn(partition->module, data, &dn_res,
 513                                             dn, no_attrs,
 514                                             DSDB_FLAG_NEXT_MODULE, parent);
 515                 if (ret == LDB_SUCCESS) {
 516                         talloc_free(partition->ctrl->dn);
 517                         partition->ctrl->dn = talloc_steal(partition->ctrl, dn_res->msgs[0]->dn);
 518                         talloc_free(dn_res);
 519                 } else if (ret != LDB_ERR_NO_SUCH_OBJECT) {
 520                         ldb_asprintf_errstring(ldb,
 521                                                "Failed to search for partition base %s in new partition at %s: %s",
 522                                                ldb_dn_get_linearized(dn),
 523                                                partition->backend_url,
 524                                                ldb_errstring(ldb));
 525                         talloc_free(mem_ctx);
 526                         return ret;
 527                 }
 528
 529                 /* see if it is a partial replica */
 530                 for (j=0; partial_replicas && j<partial_replicas->num_values; j++) {
 531                         struct ldb_dn *pa_dn = ldb_dn_from_ldb_val(mem_ctx, ldb, &partial_replicas->values[j]);
 532                         if (pa_dn != NULL && ldb_dn_compare(pa_dn, partition->ctrl->dn) == 0) {
 533                                 partition->partial_replica = true;
 534                         }
 535                         talloc_free(pa_dn);
 536                 }
 537
 538                 ret = add_partition_to_data(ldb, data, partition);
 539                 if (ret != LDB_SUCCESS) {
 540                         talloc_free(mem_ctx);
 541                         return ret;
 542                 }
 543         }
 544
 545         talloc_free(mem_ctx);
 546         return LDB_SUCCESS;
 547 }
 548
 549 /* Copy the metadata (@OPTIONS etc) for the new partition into the partition */
 550
 551 static int new_partition_set_replicated_metadata(struct ldb_context *ldb,
 552                                                  struct ldb_module *module, struct ldb_request *last_req,
 553                                                  struct partition_private_data *data,
 554                                                  struct dsdb_partition *partition)
 555 {
 556         unsigned int i;
 557         int ret;
 558         /* for each replicate, copy from main partition.  If we get an error, we report it up the chain */
 559         for (i=0; data->replicate && data->replicate[i]; i++) {
 560                 struct ldb_result *replicate_res;
 561                 struct ldb_request *add_req;
 562                 ret = dsdb_module_search_dn(module, last_req, &replicate_res,
 563                                             data->replicate[i],
 564                                             NULL,
 565                                             DSDB_FLAG_NEXT_MODULE, NULL);
 566                 if (ret == LDB_ERR_NO_SUCH_OBJECT) {
 567                         continue;
 568                 }
 569                 if (ret != LDB_SUCCESS) {
 570                         ldb_asprintf_errstring(ldb,
 571                                                "Failed to search for %s from " DSDB_PARTITION_DN
 572                                                " replicateEntries for new partition at %s on %s: %s",
 573                                                ldb_dn_get_linearized(data->replicate[i]),
 574                                                partition->backend_url,
 575                                                ldb_dn_get_linearized(partition->ctrl->dn),
 576                                                ldb_errstring(ldb));
 577                         return ret;
 578                 }
 579
 580                 /* Build add request */
 581                 ret = ldb_build_add_req(&add_req, ldb, replicate_res,
 582                                         replicate_res->msgs[0], NULL, NULL,
 583                                         ldb_op_default_callback, last_req);
 584                 LDB_REQ_SET_LOCATION(add_req);
 585                 last_req = add_req;
 586                 if (ret != LDB_SUCCESS) {
 587                         /* return directly, this is a very unlikely error */
 588                         return ret;
 589                 }
 590                 /* do request */
 591                 ret = ldb_next_request(partition->module, add_req);
 592                 /* wait */
 593                 if (ret == LDB_SUCCESS) {
 594                         ret = ldb_wait(add_req->handle, LDB_WAIT_ALL);
 595                 }
 596
 597                 switch (ret) {
 598                 case LDB_SUCCESS:
 599                         break;
 600
 601                 case LDB_ERR_ENTRY_ALREADY_EXISTS:
 602                         /* Handle this case specially - if the
 603                          * metadata already exists, replace it */
 604                 {
 605                         struct ldb_request *del_req;
 606
 607                         /* Don't leave a confusing string in the ldb_errstring() */
 608                         ldb_reset_err_string(ldb);
 609                         /* Build del request */
 610                         ret = ldb_build_del_req(&del_req, ldb, replicate_res, replicate_res->msgs[0]->dn, NULL, NULL,
 611                                                 ldb_op_default_callback, last_req);
 612                         LDB_REQ_SET_LOCATION(del_req);
 613                         last_req = del_req;
 614                         if (ret != LDB_SUCCESS) {
 615                                 /* return directly, this is a very unlikely error */
 616                                 return ret;
 617                         }
 618                         /* do request */
 619                         ret = ldb_next_request(partition->module, del_req);
 620
 621                         /* wait */
 622                         if (ret == LDB_SUCCESS) {
 623                                 ret = ldb_wait(del_req->handle, LDB_WAIT_ALL);
 624                         }
 625                         if (ret != LDB_SUCCESS) {
 626                                 ldb_asprintf_errstring(ldb,
 627                                                        "Failed to delete  (for re-add) %s from " DSDB_PARTITION_DN
 628                                                        " replicateEntries in new partition at %s on %s: %s",
 629                                                        ldb_dn_get_linearized(data->replicate[i]),
 630                                                        partition->backend_url,
 631                                                        ldb_dn_get_linearized(partition->ctrl->dn),
 632                                                        ldb_errstring(ldb));
 633                                 return ret;
 634                         }
 635
 636                         /* Build add request */
 637                         ret = ldb_build_add_req(&add_req, ldb, replicate_res, replicate_res->msgs[0], NULL, NULL,
 638                                                 ldb_op_default_callback, last_req);
 639                         LDB_REQ_SET_LOCATION(add_req);
 640                         last_req = add_req;
 641                         if (ret != LDB_SUCCESS) {
 642                                 /* return directly, this is a very unlikely error */
 643                                 return ret;
 644                         }
 645
 646                         /* do the add again */
 647                         ret = ldb_next_request(partition->module, add_req);
 648
 649                         /* wait */
 650                         if (ret == LDB_SUCCESS) {
 651                                 ret = ldb_wait(add_req->handle, LDB_WAIT_ALL);
 652                         }
 653
 654                         if (ret != LDB_SUCCESS) {
 655                                 ldb_asprintf_errstring(ldb,
 656                                                        "Failed to add (after delete) %s from " DSDB_PARTITION_DN
 657                                                        " replicateEntries to new partition at %s on %s: %s",
 658                                                        ldb_dn_get_linearized(data->replicate[i]),
 659                                                        partition->backend_url,
 660                                                        ldb_dn_get_linearized(partition->ctrl->dn),
 661                                                        ldb_errstring(ldb));
 662                                 return ret;
 663                         }
 664                         break;
 665                 }
 666                 default:
 667                 {
 668                         ldb_asprintf_errstring(ldb,
 669                                                "Failed to add %s from " DSDB_PARTITION_DN
 670                                                " replicateEntries to new partition at %s on %s: %s",
 671                                                ldb_dn_get_linearized(data->replicate[i]),
 672                                                partition->backend_url,
 673                                                ldb_dn_get_linearized(partition->ctrl->dn),
 674                                                ldb_errstring(ldb));
 675                         return ret;
 676                 }
 677                 }
 678
 679                 /* And around again, for the next thing we must merge */
 680         }
 681         return LDB_SUCCESS;
 682 }
 683
 684 /* Extended operation to create a new partition, called when
 685  * 'new_partition' detects that one is being added based on it's
 686  * instanceType */
 687 int partition_create(struct ldb_module *module, struct ldb_request *req)
 688 {
 689         unsigned int i;
 690         int ret;
 691         struct ldb_context *ldb = ldb_module_get_ctx(module);
 692         struct ldb_request *mod_req, *last_req = req;
 693         struct ldb_message *mod_msg;
 694         struct partition_private_data *data;
 695         struct dsdb_partition *partition = NULL;
 696         const char *casefold_dn;
 697         bool new_partition = false;
 698
 699         /* Check if this is already a partition */
 700
 701         struct dsdb_create_partition_exop *ex_op = talloc_get_type(req->op.extended.data, struct dsdb_create_partition_exop);
 702         struct ldb_dn *dn = ex_op->new_dn;
 703
 704         data = talloc_get_type(ldb_module_get_private(module), struct partition_private_data);
 705         if (!data) {
 706                 /* We are not going to create a partition before we are even set up */
 707                 return LDB_ERR_UNWILLING_TO_PERFORM;
 708         }
 709
 710         for (i=0; data->partitions && data->partitions[i]; i++) {
 711                 if (ldb_dn_compare(data->partitions[i]->ctrl->dn, dn) == 0) {
 712                         partition = data->partitions[i];
 713                 }
 714         }
 715
 716         if (!partition) {
 717                 char *filename;
 718                 char *partition_record;
 719                 new_partition = true;
 720                 mod_msg = ldb_msg_new(req);
 721                 if (!mod_msg) {
 722                         return ldb_oom(ldb);
 723                 }
 724
 725                 mod_msg->dn = ldb_dn_new(mod_msg, ldb, DSDB_PARTITION_DN);
 726                 ret = ldb_msg_add_empty(mod_msg, DSDB_PARTITION_ATTR, LDB_FLAG_MOD_ADD, NULL);
 727                 if (ret != LDB_SUCCESS) {
 728                         return ret;
 729                 }
 730
 731                 casefold_dn = ldb_dn_get_casefold(dn);
 732
 733                 {
 734                         char *escaped;
 735                         const char *p, *sam_name;
 736                         sam_name = strrchr((const char *)ldb_get_opaque(ldb, "ldb_url"), '/');
 737                         if (!sam_name) {
 738                                 return ldb_operr(ldb);
 739                         }
 740                         sam_name++;
 741
 742                         for (p = casefold_dn; *p; p++) {
 743                                 /* We have such a strict check because
 744                                  * I don't want shell metacharacters
 745                                  * in the file name, nor ../, but I do
 746                                  * want it to be easily typed if SAFE
 747                                  * to do so */
 748                                 if (!(isalnum(*p) || *p == ' ' || *p == '=' || *p == ',')) {
 749                                         break;
 750                                 }
 751                         }
 752                         if (*p) {
 753                                 escaped = rfc1738_escape_part(mod_msg, casefold_dn);
 754                                 if (!escaped) {
 755                                         return ldb_oom(ldb);
 756                                 }
 757                                 filename = talloc_asprintf(mod_msg, "%s.d/%s.ldb", sam_name, escaped);
 758                                 talloc_free(escaped);
 759                         } else {
 760                                 filename = talloc_asprintf(mod_msg, "%s.d/%s.ldb", sam_name, casefold_dn);
 761                         }
 762
 763                         if (!filename) {
 764                                 return ldb_oom(ldb);
 765                         }
 766                 }
 767                 partition_record = talloc_asprintf(mod_msg, "%s:%s", casefold_dn, filename);
 768
 769                 ret = ldb_msg_add_steal_string(mod_msg, DSDB_PARTITION_ATTR, partition_record);
 770                 if (ret != LDB_SUCCESS) {
 771                         return ret;
 772                 }
 773
 774                 if (ldb_request_get_control(req, DSDB_CONTROL_PARTIAL_REPLICA)) {
 775                         /* this new partition is a partial replica */
 776                         ret = ldb_msg_add_empty(mod_msg, "partialReplica", LDB_FLAG_MOD_ADD, NULL);
 777                         if (ret != LDB_SUCCESS) {
 778                                 return ret;
 779                         }
 780                         ret = ldb_msg_add_fmt(mod_msg, "partialReplica", "%s", ldb_dn_get_linearized(dn));
 781                         if (ret != LDB_SUCCESS) {
 782                                 return ret;
 783                         }
 784                 }
 785
 786                 /* Perform modify on @PARTITION record */
 787                 ret = ldb_build_mod_req(&mod_req, ldb, req, mod_msg, NULL, NULL,
 788                                         ldb_op_default_callback, req);
 789                 LDB_REQ_SET_LOCATION(mod_req);
 790                 if (ret != LDB_SUCCESS) {
 791                         return ret;
 792                 }
 793
 794                 last_req = mod_req;
 795
 796                 ret = ldb_next_request(module, mod_req);
 797                 if (ret == LDB_SUCCESS) {
 798                         ret = ldb_wait(mod_req->handle, LDB_WAIT_ALL);
 799                 }
 800
 801                 if (ret != LDB_SUCCESS) {
 802                         return ret;
 803                 }
 804
 805                 /* Make a partition structure for this new partition, so we can copy in the template structure */
 806                 ret = new_partition_from_dn(ldb, data, req, ldb_dn_copy(req, dn), filename, &partition);
 807                 if (ret != LDB_SUCCESS) {
 808                         return ret;
 809                 }
 810                 talloc_steal(partition, partition_record);
 811                 partition->orig_record = data_blob_string_const(partition_record);
 812         }
 813
 814         ret = new_partition_set_replicated_metadata(ldb, module, last_req, data, partition);
 815         if (ret != LDB_SUCCESS) {
 816                 return ret;
 817         }
 818
 819         if (new_partition) {
 820                 ret = add_partition_to_data(ldb, data, partition);
 821                 if (ret != LDB_SUCCESS) {
 822                         return ret;
 823                 }
 824         }
 825
 826         /* send request done */
 827         return ldb_module_done(req, NULL, NULL, LDB_SUCCESS);
 828 }
 829
 830
 831 int partition_init(struct ldb_module *module)
 832 {
 833         int ret;
 834         TALLOC_CTX *mem_ctx = talloc_new(module);
 835         struct ldb_context *ldb = ldb_module_get_ctx(module);
 836         struct partition_private_data *data;
 837
 838         if (!mem_ctx) {
 839                 return ldb_operr(ldb);
 840         }
 841
 842         data = talloc_zero(mem_ctx, struct partition_private_data);
 843         if (data == NULL) {
 844                 return ldb_operr(ldb);
 845         }
 846
 847         /* When used from Samba4, this message is set by the samba4
 848          * module, as a fixed value not read from the DB.  This avoids
 849          * listing modules in the DB */
 850         data->forced_module_msg = talloc_get_type(
 851                 ldb_get_opaque(ldb,
 852                                DSDB_OPAQUE_PARTITION_MODULE_MSG_OPAQUE_NAME),
 853                 struct ldb_message);
 854
 855         /* This loads the partitions */
 856         ret = partition_reload_if_required(module, data, NULL);
 857         if (ret != LDB_SUCCESS) {
 858                 return ret;
 859         }
 860
 861         ldb_module_set_private(module, talloc_steal(module, data));
 862         talloc_free(mem_ctx);
 863
 864         ret = ldb_mod_register_control(module, LDB_CONTROL_DOMAIN_SCOPE_OID);
 865         if (ret != LDB_SUCCESS) {
 866                 ldb_debug(ldb, LDB_DEBUG_ERROR,
 867                         "partition: Unable to register control with rootdse!\n");
 868                 return ldb_operr(ldb);
 869         }
 870
 871         ret = ldb_mod_register_control(module, LDB_CONTROL_SEARCH_OPTIONS_OID);
 872         if (ret != LDB_SUCCESS) {
 873                 ldb_debug(ldb, LDB_DEBUG_ERROR,
 874                         "partition: Unable to register control with rootdse!\n");
 875                 return ldb_operr(ldb);
 876         }
 877
 878         return ldb_next_init(module);
 879 }