lib/ldb/ldb_tdb/ldb_tdb.c

   1 /*
   2    ldb database library
   3
   4    Copyright (C) Andrew Tridgell 2004
   5    Copyright (C) Stefan Metzmacher 2004
   6    Copyright (C) Simo Sorce 2006-2008
   7    Copyright (C) Matthias Dieter Wallnöfer 2009-2010
   8
   9      ** NOTE! The following LGPL license applies to the ldb
  10      ** library. This does NOT imply that all of Samba is released
  11      ** under the LGPL
  12
  13    This library is free software; you can redistribute it and/or
  14    modify it under the terms of the GNU Lesser General Public
  15    License as published by the Free Software Foundation; either
  16    version 3 of the License, or (at your option) any later version.
  17
  18    This library is distributed in the hope that it will be useful,
  19    but WITHOUT ANY WARRANTY; without even the implied warranty of
  20    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  21    Lesser General Public License for more details.
  22
  23    You should have received a copy of the GNU Lesser General Public
  24    License along with this library; if not, see <http://www.gnu.org/licenses/>.
  25 */
  26
  27 /*
  28  *  Name: ldb_tdb
  29  *
  30  *  Component: ldb tdb backend
  31  *
  32  *  Description: core functions for tdb backend
  33  *
  34  *  Author: Andrew Tridgell
  35  *  Author: Stefan Metzmacher
  36  *
  37  *  Modifications:
  38  *
  39  *  - description: make the module use asynchronous calls
  40  *    date: Feb 2006
  41  *    Author: Simo Sorce
  42  *
  43  *  - description: make it possible to use event contexts
  44  *    date: Jan 2008
  45  *    Author: Simo Sorce
  46  *
  47  *  - description: fix up memory leaks and small bugs
  48  *    date: Oct 2009
  49  *    Author: Matthias Dieter Wallnöfer
  50  */
  51
  52 #include "ldb_tdb.h"
  53 #include "ldb_private.h"
  54 #include <tdb.h>
  55
  56 /*
  57   prevent memory errors on callbacks
  58 */
  59 struct ltdb_req_spy {
  60         struct ltdb_context *ctx;
  61 };
  62
  63 /*
  64   map a tdb error code to a ldb error code
  65 */
  66 int ltdb_err_map(enum TDB_ERROR tdb_code)
  67 {
  68         switch (tdb_code) {
  69         case TDB_SUCCESS:
  70                 return LDB_SUCCESS;
  71         case TDB_ERR_CORRUPT:
  72         case TDB_ERR_OOM:
  73         case TDB_ERR_EINVAL:
  74                 return LDB_ERR_OPERATIONS_ERROR;
  75         case TDB_ERR_IO:
  76                 return LDB_ERR_PROTOCOL_ERROR;
  77         case TDB_ERR_LOCK:
  78         case TDB_ERR_NOLOCK:
  79                 return LDB_ERR_BUSY;
  80         case TDB_ERR_LOCK_TIMEOUT:
  81                 return LDB_ERR_TIME_LIMIT_EXCEEDED;
  82         case TDB_ERR_EXISTS:
  83                 return LDB_ERR_ENTRY_ALREADY_EXISTS;
  84         case TDB_ERR_NOEXIST:
  85                 return LDB_ERR_NO_SUCH_OBJECT;
  86         case TDB_ERR_RDONLY:
  87                 return LDB_ERR_INSUFFICIENT_ACCESS_RIGHTS;
  88         default:
  89                 break;
  90         }
  91         return LDB_ERR_OTHER;
  92 }
  93
  94 /*
  95   lock the database for read - use by ltdb_search and ltdb_sequence_number
  96 */
  97 static int ltdb_lock_read(struct ldb_module *module)
  98 {
  99         void *data = ldb_module_get_private(module);
 100         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 101         int tdb_ret = 0;
 102         int ret;
 103         pid_t pid = getpid();
 104
 105         if (ltdb->pid != pid) {
 106                 ldb_asprintf_errstring(
 107                         ldb_module_get_ctx(module),
 108                         __location__": Reusing ldb opend by pid %d in "
 109                         "process %d\n",
 110                         ltdb->pid,
 111                         pid);
 112                 return LDB_ERR_PROTOCOL_ERROR;
 113         }
 114
 115         if (tdb_transaction_active(ltdb->tdb) == false &&
 116             ltdb->read_lock_count == 0) {
 117                 tdb_ret = tdb_lockall_read(ltdb->tdb);
 118         }
 119         if (tdb_ret == 0) {
 120                 ltdb->read_lock_count++;
 121                 return LDB_SUCCESS;
 122         }
 123         ret = ltdb_err_map(tdb_error(ltdb->tdb));
 124         if (ret == LDB_SUCCESS) {
 125                 ret = LDB_ERR_OPERATIONS_ERROR;
 126         }
 127         ldb_debug_set(ldb_module_get_ctx(module),
 128                       LDB_DEBUG_FATAL,
 129                       "Failure during ltdb_lock_read(): %s -> %s",
 130                       tdb_errorstr(ltdb->tdb),
 131                       ldb_strerror(ret));
 132         return ret;
 133 }
 134
 135 /*
 136   unlock the database after a ltdb_lock_read()
 137 */
 138 static int ltdb_unlock_read(struct ldb_module *module)
 139 {
 140         void *data = ldb_module_get_private(module);
 141         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 142         pid_t pid = getpid();
 143
 144         if (ltdb->pid != pid) {
 145                 ldb_asprintf_errstring(
 146                         ldb_module_get_ctx(module),
 147                         __location__": Reusing ldb opend by pid %d in "
 148                         "process %d\n",
 149                         ltdb->pid,
 150                         pid);
 151                 return LDB_ERR_PROTOCOL_ERROR;
 152         }
 153         if (!tdb_transaction_active(ltdb->tdb) && ltdb->read_lock_count == 1) {
 154                 tdb_unlockall_read(ltdb->tdb);
 155                 ltdb->read_lock_count--;
 156                 return 0;
 157         }
 158         ltdb->read_lock_count--;
 159         return 0;
 160 }
 161
 162
 163 /*
 164  * Determine if this key could hold a record.  We allow the new GUID
 165  * index, the old DN index and a possible future ID=
 166  */
 167 bool ltdb_key_is_record(TDB_DATA key)
 168 {
 169         if (key.dsize < 4) {
 170                 return false;
 171         }
 172
 173         if (memcmp(key.dptr, "DN=", 3) == 0) {
 174                 return true;
 175         }
 176
 177         if (memcmp(key.dptr, "ID=", 3) == 0) {
 178                 return true;
 179         }
 180
 181         if (key.dsize < sizeof(LTDB_GUID_KEY_PREFIX)) {
 182                 return false;
 183         }
 184
 185         if (memcmp(key.dptr, LTDB_GUID_KEY_PREFIX,
 186                    sizeof(LTDB_GUID_KEY_PREFIX) - 1) == 0) {
 187                 return true;
 188         }
 189
 190         return false;
 191 }
 192
 193 /*
 194   form a TDB_DATA for a record key
 195   caller frees
 196
 197   note that the key for a record can depend on whether the
 198   dn refers to a case sensitive index record or not
 199 */
 200 TDB_DATA ltdb_key_dn(struct ldb_module *module, TALLOC_CTX *mem_ctx,
 201                      struct ldb_dn *dn)
 202 {
 203         TDB_DATA key;
 204         char *key_str = NULL;
 205         const char *dn_folded = NULL;
 206
 207         /*
 208           most DNs are case insensitive. The exception is index DNs for
 209           case sensitive attributes
 210
 211           there are 3 cases dealt with in this code:
 212
 213           1) if the dn doesn't start with @ then uppercase the attribute
 214              names and the attributes values of case insensitive attributes
 215           2) if the dn starts with @ then leave it alone -
 216              the indexing code handles the rest
 217         */
 218
 219         dn_folded = ldb_dn_get_casefold(dn);
 220         if (!dn_folded) {
 221                 goto failed;
 222         }
 223
 224         key_str = talloc_strdup(mem_ctx, "DN=");
 225         if (!key_str) {
 226                 goto failed;
 227         }
 228
 229         key_str = talloc_strdup_append_buffer(key_str, dn_folded);
 230         if (!key_str) {
 231                 goto failed;
 232         }
 233
 234         key.dptr = (uint8_t *)key_str;
 235         key.dsize = strlen(key_str) + 1;
 236
 237         return key;
 238
 239 failed:
 240         errno = ENOMEM;
 241         key.dptr = NULL;
 242         key.dsize = 0;
 243         return key;
 244 }
 245
 246 /* The caller is to provide a correctly sized key */
 247 int ltdb_guid_to_key(struct ldb_module *module,
 248                      struct ltdb_private *ltdb,
 249                      const struct ldb_val *GUID_val,
 250                      TDB_DATA *key)
 251 {
 252         const char *GUID_prefix = LTDB_GUID_KEY_PREFIX;
 253         const int GUID_prefix_len = sizeof(LTDB_GUID_KEY_PREFIX) - 1;
 254
 255         if (key->dsize != (GUID_val->length+GUID_prefix_len)) {
 256                 return LDB_ERR_OPERATIONS_ERROR;
 257         }
 258
 259         memcpy(key->dptr, GUID_prefix, GUID_prefix_len);
 260         memcpy(&key->dptr[GUID_prefix_len],
 261                GUID_val->data, GUID_val->length);
 262         return LDB_SUCCESS;
 263 }
 264
 265 /*
 266  * The caller is to provide a correctly sized key, used only in
 267  * the GUID index mode
 268  */
 269 int ltdb_idx_to_key(struct ldb_module *module,
 270                     struct ltdb_private *ltdb,
 271                     TALLOC_CTX *mem_ctx,
 272                     const struct ldb_val *idx_val,
 273                     TDB_DATA *key)
 274 {
 275         struct ldb_context *ldb = ldb_module_get_ctx(module);
 276         struct ldb_dn *dn;
 277
 278         if (ltdb->cache->GUID_index_attribute != NULL) {
 279                 return ltdb_guid_to_key(module, ltdb,
 280                                         idx_val, key);
 281         }
 282
 283         dn = ldb_dn_from_ldb_val(mem_ctx, ldb, idx_val);
 284         if (dn == NULL) {
 285                 /*
 286                  * LDB_ERR_INVALID_DN_SYNTAX would just be confusing
 287                  * to the caller, as this in an invalid index value
 288                  */
 289                 return LDB_ERR_OPERATIONS_ERROR;
 290         }
 291         /* form the key */
 292         *key = ltdb_key_dn(module, mem_ctx, dn);
 293         TALLOC_FREE(dn);
 294         if (!key->dptr) {
 295                 return ldb_module_oom(module);
 296         }
 297         return LDB_SUCCESS;
 298 }
 299
 300 /*
 301   form a TDB_DATA for a record key
 302   caller frees mem_ctx, which may or may not have the key
 303   as a child.
 304
 305   note that the key for a record can depend on whether a
 306   GUID index is in use, or the DN is used as the key
 307 */
 308 TDB_DATA ltdb_key_msg(struct ldb_module *module, TALLOC_CTX *mem_ctx,
 309                       const struct ldb_message *msg)
 310 {
 311         void *data = ldb_module_get_private(module);
 312         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 313         TDB_DATA key;
 314         const struct ldb_val *guid_val;
 315         int ret;
 316
 317         if (ltdb->cache->GUID_index_attribute == NULL) {
 318                 return ltdb_key_dn(module, mem_ctx, msg->dn);
 319         }
 320
 321         if (ldb_dn_is_special(msg->dn)) {
 322                 return ltdb_key_dn(module, mem_ctx, msg->dn);
 323         }
 324
 325         guid_val = ldb_msg_find_ldb_val(msg,
 326                                        ltdb->cache->GUID_index_attribute);
 327         if (guid_val == NULL) {
 328                 ldb_asprintf_errstring(ldb_module_get_ctx(module),
 329                                        "Did not find GUID attribute %s "
 330                                        "in %s, required for TDB record "
 331                                        "key in " LTDB_IDXGUID " mode.",
 332                                        ltdb->cache->GUID_index_attribute,
 333                                        ldb_dn_get_linearized(msg->dn));
 334                 errno = EINVAL;
 335                 key.dptr = NULL;
 336                 key.dsize = 0;
 337                 return key;
 338         }
 339
 340         /* In this case, allocate with talloc */
 341         key.dptr = talloc_size(mem_ctx, LTDB_GUID_KEY_SIZE);
 342         if (key.dptr == NULL) {
 343                 errno = ENOMEM;
 344                 key.dptr = NULL;
 345                 key.dsize = 0;
 346                 return key;
 347         }
 348         key.dsize = talloc_get_size(key.dptr);
 349
 350         ret = ltdb_guid_to_key(module, ltdb, guid_val, &key);
 351
 352         if (ret != LDB_SUCCESS) {
 353                 errno = EINVAL;
 354                 key.dptr = NULL;
 355                 key.dsize = 0;
 356                 return key;
 357         }
 358         return key;
 359 }
 360
 361 /*
 362   check special dn's have valid attributes
 363   currently only @ATTRIBUTES is checked
 364 */
 365 static int ltdb_check_special_dn(struct ldb_module *module,
 366                                  const struct ldb_message *msg)
 367 {
 368         struct ldb_context *ldb = ldb_module_get_ctx(module);
 369         unsigned int i, j;
 370
 371         if (! ldb_dn_is_special(msg->dn) ||
 372             ! ldb_dn_check_special(msg->dn, LTDB_ATTRIBUTES)) {
 373                 return LDB_SUCCESS;
 374         }
 375
 376         /* we have @ATTRIBUTES, let's check attributes are fine */
 377         /* should we check that we deny multivalued attributes ? */
 378         for (i = 0; i < msg->num_elements; i++) {
 379                 if (ldb_attr_cmp(msg->elements[i].name, "distinguishedName") == 0) continue;
 380
 381                 for (j = 0; j < msg->elements[i].num_values; j++) {
 382                         if (ltdb_check_at_attributes_values(&msg->elements[i].values[j]) != 0) {
 383                                 ldb_set_errstring(ldb, "Invalid attribute value in an @ATTRIBUTES entry");
 384                                 return LDB_ERR_INVALID_ATTRIBUTE_SYNTAX;
 385                         }
 386                 }
 387         }
 388
 389         return LDB_SUCCESS;
 390 }
 391
 392
 393 /*
 394   we've made a modification to a dn - possibly reindex and
 395   update sequence number
 396 */
 397 static int ltdb_modified(struct ldb_module *module, struct ldb_dn *dn)
 398 {
 399         int ret = LDB_SUCCESS;
 400         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(module), struct ltdb_private);
 401
 402         /* only allow modifies inside a transaction, otherwise the
 403          * ldb is unsafe */
 404         if (ltdb->kv_ops->transaction_active(ltdb) == false) {
 405                 ldb_set_errstring(ldb_module_get_ctx(module), "ltdb modify without transaction");
 406                 return LDB_ERR_OPERATIONS_ERROR;
 407         }
 408
 409         if (ldb_dn_is_special(dn) &&
 410             (ldb_dn_check_special(dn, LTDB_INDEXLIST) ||
 411              ldb_dn_check_special(dn, LTDB_ATTRIBUTES)) )
 412         {
 413                 if (ltdb->warn_reindex) {
 414                         ldb_debug(ldb_module_get_ctx(module),
 415                                 LDB_DEBUG_ERROR, "Reindexing %s due to modification on %s",
 416                                 ltdb->kv_ops->name(ltdb), ldb_dn_get_linearized(dn));
 417                 }
 418                 ret = ltdb_reindex(module);
 419         }
 420
 421         /* If the modify was to a normal record, or any special except @BASEINFO, update the seq number */
 422         if (ret == LDB_SUCCESS &&
 423             !(ldb_dn_is_special(dn) &&
 424               ldb_dn_check_special(dn, LTDB_BASEINFO)) ) {
 425                 ret = ltdb_increase_sequence_number(module);
 426         }
 427
 428         /* If the modify was to @OPTIONS, reload the cache */
 429         if (ret == LDB_SUCCESS &&
 430             ldb_dn_is_special(dn) &&
 431             (ldb_dn_check_special(dn, LTDB_OPTIONS)) ) {
 432                 ret = ltdb_cache_reload(module);
 433         }
 434
 435         if (ret != LDB_SUCCESS) {
 436                 ltdb->reindex_failed = true;
 437         }
 438
 439         return ret;
 440 }
 441
 442 static int ltdb_tdb_store(struct ltdb_private *ltdb, struct ldb_val ldb_key,
 443                           struct ldb_val ldb_data, int flags)
 444 {
 445         TDB_DATA key = {
 446                 .dptr = ldb_key.data,
 447                 .dsize = ldb_key.length
 448         };
 449         TDB_DATA data = {
 450                 .dptr = ldb_data.data,
 451                 .dsize = ldb_data.length
 452         };
 453         bool transaction_active = tdb_transaction_active(ltdb->tdb);
 454         if (transaction_active == false){
 455                 return LDB_ERR_PROTOCOL_ERROR;
 456         }
 457         return tdb_store(ltdb->tdb, key, data, flags);
 458 }
 459
 460 static int ltdb_error(struct ltdb_private *ltdb)
 461 {
 462         return ltdb_err_map(tdb_error(ltdb->tdb));
 463 }
 464
 465 static const char *ltdb_errorstr(struct ltdb_private *ltdb)
 466 {
 467         return tdb_errorstr(ltdb->tdb);
 468 }
 469
 470 /*
 471   store a record into the db
 472 */
 473 int ltdb_store(struct ldb_module *module, const struct ldb_message *msg, int flgs)
 474 {
 475         void *data = ldb_module_get_private(module);
 476         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 477         TDB_DATA tdb_key;
 478         struct ldb_val ldb_key;
 479         struct ldb_val ldb_data;
 480         int ret = LDB_SUCCESS;
 481         TALLOC_CTX *tdb_key_ctx = talloc_new(module);
 482
 483         if (tdb_key_ctx == NULL) {
 484                 return ldb_module_oom(module);
 485         }
 486
 487         if (ltdb->read_only) {
 488                 return LDB_ERR_UNWILLING_TO_PERFORM;
 489         }
 490
 491         tdb_key = ltdb_key_msg(module, tdb_key_ctx, msg);
 492         if (tdb_key.dptr == NULL) {
 493                 TALLOC_FREE(tdb_key_ctx);
 494                 return LDB_ERR_OTHER;
 495         }
 496
 497         ret = ldb_pack_data(ldb_module_get_ctx(module),
 498                             msg, &ldb_data);
 499         if (ret == -1) {
 500                 TALLOC_FREE(tdb_key_ctx);
 501                 return LDB_ERR_OTHER;
 502         }
 503
 504         ldb_key.data = tdb_key.dptr;
 505         ldb_key.length = tdb_key.dsize;
 506
 507         ret = ltdb->kv_ops->store(ltdb, ldb_key, ldb_data, flgs);
 508         if (ret != 0) {
 509                 bool is_special = ldb_dn_is_special(msg->dn);
 510                 ret = ltdb->kv_ops->error(ltdb);
 511
 512                 /*
 513                  * LDB_ERR_ENTRY_ALREADY_EXISTS means the DN, not
 514                  * the GUID, so re-map
 515                  */
 516                 if (ret == LDB_ERR_ENTRY_ALREADY_EXISTS
 517                     && !is_special
 518                     && ltdb->cache->GUID_index_attribute != NULL) {
 519                         ret = LDB_ERR_CONSTRAINT_VIOLATION;
 520                 }
 521                 goto done;
 522         }
 523
 524 done:
 525         TALLOC_FREE(tdb_key_ctx);
 526         talloc_free(ldb_data.data);
 527
 528         return ret;
 529 }
 530
 531
 532 /*
 533   check if a attribute is a single valued, for a given element
 534  */
 535 static bool ldb_tdb_single_valued(const struct ldb_schema_attribute *a,
 536                                   struct ldb_message_element *el)
 537 {
 538         if (!a) return false;
 539         if (el != NULL) {
 540                 if (el->flags & LDB_FLAG_INTERNAL_FORCE_SINGLE_VALUE_CHECK) {
 541                         /* override from a ldb module, for example
 542                            used for the description field, which is
 543                            marked multi-valued in the schema but which
 544                            should not actually accept multiple
 545                            values */
 546                         return true;
 547                 }
 548                 if (el->flags & LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK) {
 549                         /* override from a ldb module, for example used for
 550                            deleted linked attribute entries */
 551                         return false;
 552                 }
 553         }
 554         if (a->flags & LDB_ATTR_FLAG_SINGLE_VALUE) {
 555                 return true;
 556         }
 557         return false;
 558 }
 559
 560 static int ltdb_add_internal(struct ldb_module *module,
 561                              struct ltdb_private *ltdb,
 562                              const struct ldb_message *msg,
 563                              bool check_single_value)
 564 {
 565         struct ldb_context *ldb = ldb_module_get_ctx(module);
 566         int ret = LDB_SUCCESS;
 567         unsigned int i;
 568
 569         for (i=0;i<msg->num_elements;i++) {
 570                 struct ldb_message_element *el = &msg->elements[i];
 571                 const struct ldb_schema_attribute *a = ldb_schema_attribute_by_name(ldb, el->name);
 572
 573                 if (el->num_values == 0) {
 574                         ldb_asprintf_errstring(ldb, "attribute '%s' on '%s' specified, but with 0 values (illegal)",
 575                                                el->name, ldb_dn_get_linearized(msg->dn));
 576                         return LDB_ERR_CONSTRAINT_VIOLATION;
 577                 }
 578                 if (check_single_value &&
 579                                 el->num_values > 1 &&
 580                                 ldb_tdb_single_valued(a, el)) {
 581                         ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
 582                                                el->name, ldb_dn_get_linearized(msg->dn));
 583                         return LDB_ERR_CONSTRAINT_VIOLATION;
 584                 }
 585
 586                 /* Do not check "@ATTRIBUTES" for duplicated values */
 587                 if (ldb_dn_is_special(msg->dn) &&
 588                     ldb_dn_check_special(msg->dn, LTDB_ATTRIBUTES)) {
 589                         continue;
 590                 }
 591
 592                 if (check_single_value &&
 593                     !(el->flags &
 594                       LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK)) {
 595                         struct ldb_val *duplicate = NULL;
 596
 597                         ret = ldb_msg_find_duplicate_val(ldb, discard_const(msg),
 598                                                          el, &duplicate, 0);
 599                         if (ret != LDB_SUCCESS) {
 600                                 return ret;
 601                         }
 602                         if (duplicate != NULL) {
 603                                 ldb_asprintf_errstring(
 604                                         ldb,
 605                                         "attribute '%s': value '%.*s' on '%s' "
 606                                         "provided more than once in ADD object",
 607                                         el->name,
 608                                         (int)duplicate->length,
 609                                         duplicate->data,
 610                                         ldb_dn_get_linearized(msg->dn));
 611                                 return LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
 612                         }
 613                 }
 614         }
 615
 616         ret = ltdb_store(module, msg, TDB_INSERT);
 617         if (ret != LDB_SUCCESS) {
 618                 /*
 619                  * Try really hard to get the right error code for
 620                  * a re-add situation, as this can matter!
 621                  */
 622                 if (ret == LDB_ERR_CONSTRAINT_VIOLATION) {
 623                         int ret2;
 624                         struct ldb_dn *dn2 = NULL;
 625                         TALLOC_CTX *mem_ctx = talloc_new(module);
 626                         if (mem_ctx == NULL) {
 627                                 return ldb_module_operr(module);
 628                         }
 629                         ret2 = ltdb_search_base(module, mem_ctx,
 630                                                 msg->dn, &dn2);
 631                         TALLOC_FREE(mem_ctx);
 632                         if (ret2 == LDB_SUCCESS) {
 633                                 ret = LDB_ERR_ENTRY_ALREADY_EXISTS;
 634                         }
 635                 }
 636                 if (ret == LDB_ERR_ENTRY_ALREADY_EXISTS) {
 637                         ldb_asprintf_errstring(ldb,
 638                                                "Entry %s already exists",
 639                                                ldb_dn_get_linearized(msg->dn));
 640                 }
 641                 return ret;
 642         }
 643
 644         ret = ltdb_index_add_new(module, ltdb, msg);
 645         if (ret != LDB_SUCCESS) {
 646                 /*
 647                  * If we failed to index, delete the message again.
 648                  *
 649                  * This is particularly important for the GUID index
 650                  * case, which will only fail for a duplicate DN
 651                  * in the index add.
 652                  *
 653                  * Note that the caller may not cancel the transation
 654                  * and this means the above add might really show up!
 655                  */
 656                 ltdb_delete_noindex(module, msg);
 657                 return ret;
 658         }
 659
 660         ret = ltdb_modified(module, msg->dn);
 661
 662         return ret;
 663 }
 664
 665 /*
 666   add a record to the database
 667 */
 668 static int ltdb_add(struct ltdb_context *ctx)
 669 {
 670         struct ldb_module *module = ctx->module;
 671         struct ldb_request *req = ctx->req;
 672         void *data = ldb_module_get_private(module);
 673         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 674         int ret = LDB_SUCCESS;
 675
 676         if (ltdb->max_key_length != 0 &&
 677             ltdb->cache->GUID_index_attribute == NULL &&
 678             !ldb_dn_is_special(req->op.add.message->dn))
 679         {
 680                 ldb_set_errstring(ldb_module_get_ctx(module),
 681                                   "Must operate ldb_mdb in GUID "
 682                                   "index mode, but " LTDB_IDXGUID " not set.");
 683                 return LDB_ERR_UNWILLING_TO_PERFORM;
 684         }
 685
 686         ret = ltdb_check_special_dn(module, req->op.add.message);
 687         if (ret != LDB_SUCCESS) {
 688                 return ret;
 689         }
 690
 691         ldb_request_set_state(req, LDB_ASYNC_PENDING);
 692
 693         if (ltdb_cache_load(module) != 0) {
 694                 return LDB_ERR_OPERATIONS_ERROR;
 695         }
 696
 697         ret = ltdb_add_internal(module, ltdb,
 698                                 req->op.add.message, true);
 699
 700         return ret;
 701 }
 702
 703 static int ltdb_tdb_delete(struct ltdb_private *ltdb, struct ldb_val ldb_key)
 704 {
 705         TDB_DATA tdb_key = {
 706                 .dptr = ldb_key.data,
 707                 .dsize = ldb_key.length
 708         };
 709         bool transaction_active = tdb_transaction_active(ltdb->tdb);
 710         if (transaction_active == false){
 711                 return LDB_ERR_PROTOCOL_ERROR;
 712         }
 713         return tdb_delete(ltdb->tdb, tdb_key);
 714 }
 715
 716 /*
 717   delete a record from the database, not updating indexes (used for deleting
 718   index records)
 719 */
 720 int ltdb_delete_noindex(struct ldb_module *module,
 721                         const struct ldb_message *msg)
 722 {
 723         void *data = ldb_module_get_private(module);
 724         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 725         struct ldb_val ldb_key;
 726         TDB_DATA tdb_key;
 727         int ret;
 728         TALLOC_CTX *tdb_key_ctx = talloc_new(module);
 729
 730         if (tdb_key_ctx == NULL) {
 731                 return ldb_module_oom(module);
 732         }
 733
 734         if (ltdb->read_only) {
 735                 return LDB_ERR_UNWILLING_TO_PERFORM;
 736         }
 737
 738         tdb_key = ltdb_key_msg(module, tdb_key_ctx, msg);
 739         if (!tdb_key.dptr) {
 740                 TALLOC_FREE(tdb_key_ctx);
 741                 return LDB_ERR_OTHER;
 742         }
 743
 744         ldb_key.data = tdb_key.dptr;
 745         ldb_key.length = tdb_key.dsize;
 746
 747         ret = ltdb->kv_ops->delete(ltdb, ldb_key);
 748         TALLOC_FREE(tdb_key_ctx);
 749
 750         if (ret != 0) {
 751                 ret = ltdb->kv_ops->error(ltdb);
 752         }
 753
 754         return ret;
 755 }
 756
 757 static int ltdb_delete_internal(struct ldb_module *module, struct ldb_dn *dn)
 758 {
 759         struct ldb_message *msg;
 760         int ret = LDB_SUCCESS;
 761
 762         msg = ldb_msg_new(module);
 763         if (msg == NULL) {
 764                 return LDB_ERR_OPERATIONS_ERROR;
 765         }
 766
 767         /* in case any attribute of the message was indexed, we need
 768            to fetch the old record */
 769         ret = ltdb_search_dn1(module, dn, msg, LDB_UNPACK_DATA_FLAG_NO_DATA_ALLOC);
 770         if (ret != LDB_SUCCESS) {
 771                 /* not finding the old record is an error */
 772                 goto done;
 773         }
 774
 775         ret = ltdb_delete_noindex(module, msg);
 776         if (ret != LDB_SUCCESS) {
 777                 goto done;
 778         }
 779
 780         /* remove any indexed attributes */
 781         ret = ltdb_index_delete(module, msg);
 782         if (ret != LDB_SUCCESS) {
 783                 goto done;
 784         }
 785
 786         ret = ltdb_modified(module, dn);
 787         if (ret != LDB_SUCCESS) {
 788                 goto done;
 789         }
 790
 791 done:
 792         talloc_free(msg);
 793         return ret;
 794 }
 795
 796 /*
 797   delete a record from the database
 798 */
 799 static int ltdb_delete(struct ltdb_context *ctx)
 800 {
 801         struct ldb_module *module = ctx->module;
 802         struct ldb_request *req = ctx->req;
 803         int ret = LDB_SUCCESS;
 804
 805         ldb_request_set_state(req, LDB_ASYNC_PENDING);
 806
 807         if (ltdb_cache_load(module) != 0) {
 808                 return LDB_ERR_OPERATIONS_ERROR;
 809         }
 810
 811         ret = ltdb_delete_internal(module, req->op.del.dn);
 812
 813         return ret;
 814 }
 815
 816 /*
 817   find an element by attribute name. At the moment this does a linear search,
 818   it should be re-coded to use a binary search once all places that modify
 819   records guarantee sorted order
 820
 821   return the index of the first matching element if found, otherwise -1
 822 */
 823 static int find_element(const struct ldb_message *msg, const char *name)
 824 {
 825         unsigned int i;
 826         for (i=0;i<msg->num_elements;i++) {
 827                 if (ldb_attr_cmp(msg->elements[i].name, name) == 0) {
 828                         return i;
 829                 }
 830         }
 831         return -1;
 832 }
 833
 834
 835 /*
 836   add an element to an existing record. Assumes a elements array that we
 837   can call re-alloc on, and assumed that we can re-use the data pointers from
 838   the passed in additional values. Use with care!
 839
 840   returns 0 on success, -1 on failure (and sets errno)
 841 */
 842 static int ltdb_msg_add_element(struct ldb_message *msg,
 843                                 struct ldb_message_element *el)
 844 {
 845         struct ldb_message_element *e2;
 846         unsigned int i;
 847
 848         if (el->num_values == 0) {
 849                 /* nothing to do here - we don't add empty elements */
 850                 return 0;
 851         }
 852
 853         e2 = talloc_realloc(msg, msg->elements, struct ldb_message_element,
 854                               msg->num_elements+1);
 855         if (!e2) {
 856                 errno = ENOMEM;
 857                 return -1;
 858         }
 859
 860         msg->elements = e2;
 861
 862         e2 = &msg->elements[msg->num_elements];
 863
 864         e2->name = el->name;
 865         e2->flags = el->flags;
 866         e2->values = talloc_array(msg->elements,
 867                                   struct ldb_val, el->num_values);
 868         if (!e2->values) {
 869                 errno = ENOMEM;
 870                 return -1;
 871         }
 872         for (i=0;i<el->num_values;i++) {
 873                 e2->values[i] = el->values[i];
 874         }
 875         e2->num_values = el->num_values;
 876
 877         ++msg->num_elements;
 878
 879         return 0;
 880 }
 881
 882 /*
 883   delete all elements having a specified attribute name
 884 */
 885 static int msg_delete_attribute(struct ldb_module *module,
 886                                 struct ltdb_private *ltdb,
 887                                 struct ldb_message *msg, const char *name)
 888 {
 889         unsigned int i;
 890         int ret;
 891         struct ldb_message_element *el;
 892         bool is_special = ldb_dn_is_special(msg->dn);
 893
 894         if (!is_special
 895             && ltdb->cache->GUID_index_attribute != NULL
 896             && ldb_attr_cmp(name, ltdb->cache->GUID_index_attribute) == 0) {
 897                 struct ldb_context *ldb = ldb_module_get_ctx(module);
 898                 ldb_asprintf_errstring(ldb, "Must not modify GUID "
 899                                        "attribute %s (used as DB index)",
 900                                        ltdb->cache->GUID_index_attribute);
 901                 return LDB_ERR_CONSTRAINT_VIOLATION;
 902         }
 903
 904         el = ldb_msg_find_element(msg, name);
 905         if (el == NULL) {
 906                 return LDB_ERR_NO_SUCH_ATTRIBUTE;
 907         }
 908         i = el - msg->elements;
 909
 910         ret = ltdb_index_del_element(module, ltdb, msg, el);
 911         if (ret != LDB_SUCCESS) {
 912                 return ret;
 913         }
 914
 915         talloc_free(el->values);
 916         if (msg->num_elements > (i+1)) {
 917                 memmove(el, el+1, sizeof(*el) * (msg->num_elements - (i+1)));
 918         }
 919         msg->num_elements--;
 920         msg->elements = talloc_realloc(msg, msg->elements,
 921                                        struct ldb_message_element,
 922                                        msg->num_elements);
 923         return LDB_SUCCESS;
 924 }
 925
 926 /*
 927   delete all elements matching an attribute name/value
 928
 929   return LDB Error on failure
 930 */
 931 static int msg_delete_element(struct ldb_module *module,
 932                               struct ltdb_private *ltdb,
 933                               struct ldb_message *msg,
 934                               const char *name,
 935                               const struct ldb_val *val)
 936 {
 937         struct ldb_context *ldb = ldb_module_get_ctx(module);
 938         unsigned int i;
 939         int found, ret;
 940         struct ldb_message_element *el;
 941         const struct ldb_schema_attribute *a;
 942
 943         found = find_element(msg, name);
 944         if (found == -1) {
 945                 return LDB_ERR_NO_SUCH_ATTRIBUTE;
 946         }
 947
 948         i = (unsigned int) found;
 949         el = &(msg->elements[i]);
 950
 951         a = ldb_schema_attribute_by_name(ldb, el->name);
 952
 953         for (i=0;i<el->num_values;i++) {
 954                 bool matched;
 955                 if (a->syntax->operator_fn) {
 956                         ret = a->syntax->operator_fn(ldb, LDB_OP_EQUALITY, a,
 957                                                      &el->values[i], val, &matched);
 958                         if (ret != LDB_SUCCESS) return ret;
 959                 } else {
 960                         matched = (a->syntax->comparison_fn(ldb, ldb,
 961                                                             &el->values[i], val) == 0);
 962                 }
 963                 if (matched) {
 964                         if (el->num_values == 1) {
 965                                 return msg_delete_attribute(module,
 966                                                             ltdb, msg, name);
 967                         }
 968
 969                         ret = ltdb_index_del_value(module, ltdb, msg, el, i);
 970                         if (ret != LDB_SUCCESS) {
 971                                 return ret;
 972                         }
 973
 974                         if (i<el->num_values-1) {
 975                                 memmove(&el->values[i], &el->values[i+1],
 976                                         sizeof(el->values[i])*
 977                                                 (el->num_values-(i+1)));
 978                         }
 979                         el->num_values--;
 980
 981                         /* per definition we find in a canonicalised message an
 982                            attribute value only once. So we are finished here */
 983                         return LDB_SUCCESS;
 984                 }
 985         }
 986
 987         /* Not found */
 988         return LDB_ERR_NO_SUCH_ATTRIBUTE;
 989 }
 990
 991 /*
 992   modify a record - internal interface
 993
 994   yuck - this is O(n^2). Luckily n is usually small so we probably
 995   get away with it, but if we ever have really large attribute lists
 996   then we'll need to look at this again
 997
 998   'req' is optional, and is used to specify controls if supplied
 999 */
1000 int ltdb_modify_internal(struct ldb_module *module,
1001                          const struct ldb_message *msg,
1002                          struct ldb_request *req)
1003 {
1004         struct ldb_context *ldb = ldb_module_get_ctx(module);
1005         void *data = ldb_module_get_private(module);
1006         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
1007         struct ldb_message *msg2;
1008         unsigned int i, j;
1009         int ret = LDB_SUCCESS, idx;
1010         struct ldb_control *control_permissive = NULL;
1011         TALLOC_CTX *mem_ctx = talloc_new(req);
1012
1013         if (mem_ctx == NULL) {
1014                 return ldb_module_oom(module);
1015         }
1016
1017         if (req) {
1018                 control_permissive = ldb_request_get_control(req,
1019                                         LDB_CONTROL_PERMISSIVE_MODIFY_OID);
1020         }
1021
1022         msg2 = ldb_msg_new(mem_ctx);
1023         if (msg2 == NULL) {
1024                 ret = LDB_ERR_OTHER;
1025                 goto done;
1026         }
1027
1028         ret = ltdb_search_dn1(module, msg->dn,
1029                               msg2,
1030                               LDB_UNPACK_DATA_FLAG_NO_DATA_ALLOC);
1031         if (ret != LDB_SUCCESS) {
1032                 goto done;
1033         }
1034
1035         for (i=0; i<msg->num_elements; i++) {
1036                 struct ldb_message_element *el = &msg->elements[i], *el2;
1037                 struct ldb_val *vals;
1038                 const struct ldb_schema_attribute *a = ldb_schema_attribute_by_name(ldb, el->name);
1039                 const char *dn;
1040                 uint32_t options = 0;
1041                 if (control_permissive != NULL) {
1042                         options |= LDB_MSG_FIND_COMMON_REMOVE_DUPLICATES;
1043                 }
1044
1045                 switch (msg->elements[i].flags & LDB_FLAG_MOD_MASK) {
1046                 case LDB_FLAG_MOD_ADD:
1047
1048                         if (el->num_values == 0) {
1049                                 ldb_asprintf_errstring(ldb,
1050                                                        "attribute '%s': attribute on '%s' specified, but with 0 values (illegal)",
1051                                                        el->name, ldb_dn_get_linearized(msg2->dn));
1052                                 ret = LDB_ERR_CONSTRAINT_VIOLATION;
1053                                 goto done;
1054                         }
1055
1056                         /* make a copy of the array so that a permissive
1057                          * control can remove duplicates without changing the
1058                          * original values, but do not copy data as we do not
1059                          * need to keep it around once the operation is
1060                          * finished */
1061                         if (control_permissive) {
1062                                 el = talloc(msg2, struct ldb_message_element);
1063                                 if (!el) {
1064                                         ret = LDB_ERR_OTHER;
1065                                         goto done;
1066                                 }
1067                                 *el = msg->elements[i];
1068                                 el->values = talloc_array(el, struct ldb_val, el->num_values);
1069                                 if (el->values == NULL) {
1070                                         ret = LDB_ERR_OTHER;
1071                                         goto done;
1072                                 }
1073                                 for (j = 0; j < el->num_values; j++) {
1074                                         el->values[j] = msg->elements[i].values[j];
1075                                 }
1076                         }
1077
1078                         if (el->num_values > 1 && ldb_tdb_single_valued(a, el)) {
1079                                 ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
1080                                                        el->name, ldb_dn_get_linearized(msg2->dn));
1081                                 ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
1082                                 goto done;
1083                         }
1084
1085                         /* Checks if element already exists */
1086                         idx = find_element(msg2, el->name);
1087                         if (idx == -1) {
1088                                 if (ltdb_msg_add_element(msg2, el) != 0) {
1089                                         ret = LDB_ERR_OTHER;
1090                                         goto done;
1091                                 }
1092                                 ret = ltdb_index_add_element(module, ltdb,
1093                                                              msg2,
1094                                                              el);
1095                                 if (ret != LDB_SUCCESS) {
1096                                         goto done;
1097                                 }
1098                         } else {
1099                                 j = (unsigned int) idx;
1100                                 el2 = &(msg2->elements[j]);
1101
1102                                 /* We cannot add another value on a existing one
1103                                    if the attribute is single-valued */
1104                                 if (ldb_tdb_single_valued(a, el)) {
1105                                         ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
1106                                                                el->name, ldb_dn_get_linearized(msg2->dn));
1107                                         ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
1108                                         goto done;
1109                                 }
1110
1111                                 /* Check that values don't exist yet on multi-
1112                                    valued attributes or aren't provided twice */
1113                                 if (!(el->flags &
1114                                       LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK)) {
1115                                         struct ldb_val *duplicate = NULL;
1116                                         ret = ldb_msg_find_common_values(ldb,
1117                                                                          msg2,
1118                                                                          el,
1119                                                                          el2,
1120                                                                          options);
1121
1122                                         if (ret ==
1123                                             LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS) {
1124                                                 ldb_asprintf_errstring(ldb,
1125                                                         "attribute '%s': value "
1126                                                         "#%u on '%s' already "
1127                                                         "exists", el->name, j,
1128                                                         ldb_dn_get_linearized(msg2->dn));
1129                                                 goto done;
1130                                         } else if (ret != LDB_SUCCESS) {
1131                                                 goto done;
1132                                         }
1133
1134                                         ret = ldb_msg_find_duplicate_val(
1135                                                 ldb, msg2, el, &duplicate, 0);
1136                                         if (ret != LDB_SUCCESS) {
1137                                                 goto done;
1138                                         }
1139                                         if (duplicate != NULL) {
1140                                                 ldb_asprintf_errstring(
1141                                                         ldb,
1142                                                         "attribute '%s': value "
1143                                                         "'%.*s' on '%s' "
1144                                                         "provided more than "
1145                                                         "once in ADD",
1146                                                         el->name,
1147                                                         (int)duplicate->length,
1148                                                         duplicate->data,
1149                                                         ldb_dn_get_linearized(msg->dn));
1150                                                 ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
1151                                                 goto done;
1152                                         }
1153                                 }
1154
1155                                 /* Now combine existing and new values to a new
1156                                    attribute record */
1157                                 vals = talloc_realloc(msg2->elements,
1158                                                       el2->values, struct ldb_val,
1159                                                       el2->num_values + el->num_values);
1160                                 if (vals == NULL) {
1161                                         ldb_oom(ldb);
1162                                         ret = LDB_ERR_OTHER;
1163                                         goto done;
1164                                 }
1165
1166                                 for (j=0; j<el->num_values; j++) {
1167                                         vals[el2->num_values + j] =
1168                                                 ldb_val_dup(vals, &el->values[j]);
1169                                 }
1170
1171                                 el2->values = vals;
1172                                 el2->num_values += el->num_values;
1173
1174                                 ret = ltdb_index_add_element(module, ltdb,
1175                                                              msg2, el);
1176                                 if (ret != LDB_SUCCESS) {
1177                                         goto done;
1178                                 }
1179                         }
1180
1181                         break;
1182
1183                 case LDB_FLAG_MOD_REPLACE:
1184
1185                         if (el->num_values > 1 && ldb_tdb_single_valued(a, el)) {
1186                                 ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
1187                                                        el->name, ldb_dn_get_linearized(msg2->dn));
1188                                 ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
1189                                 goto done;
1190                         }
1191
1192                         /*
1193                          * We don't need to check this if we have been
1194                          * pre-screened by the repl_meta_data module
1195                          * in Samba, or someone else who can claim to
1196                          * know what they are doing.
1197                          */
1198                         if (!(el->flags & LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK)) {
1199                                 struct ldb_val *duplicate = NULL;
1200
1201                                 ret = ldb_msg_find_duplicate_val(ldb, msg2, el,
1202                                                                  &duplicate, 0);
1203                                 if (ret != LDB_SUCCESS) {
1204                                         goto done;
1205                                 }
1206                                 if (duplicate != NULL) {
1207                                         ldb_asprintf_errstring(
1208                                                 ldb,
1209                                                 "attribute '%s': value '%.*s' "
1210                                                 "on '%s' provided more than "
1211                                                 "once in REPLACE",
1212                                                 el->name,
1213                                                 (int)duplicate->length,
1214                                                 duplicate->data,
1215                                                 ldb_dn_get_linearized(msg2->dn));
1216                                         ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
1217                                         goto done;
1218                                 }
1219                         }
1220
1221                         /* Checks if element already exists */
1222                         idx = find_element(msg2, el->name);
1223                         if (idx != -1) {
1224                                 j = (unsigned int) idx;
1225                                 el2 = &(msg2->elements[j]);
1226
1227                                 /* we consider two elements to be
1228                                  * equal only if the order
1229                                  * matches. This allows dbcheck to
1230                                  * fix the ordering on attributes
1231                                  * where order matters, such as
1232                                  * objectClass
1233                                  */
1234                                 if (ldb_msg_element_equal_ordered(el, el2)) {
1235                                         continue;
1236                                 }
1237
1238                                 /* Delete the attribute if it exists in the DB */
1239                                 if (msg_delete_attribute(module, ltdb,
1240                                                          msg2,
1241                                                          el->name) != 0) {
1242                                         ret = LDB_ERR_OTHER;
1243                                         goto done;
1244                                 }
1245                         }
1246
1247                         /* Recreate it with the new values */
1248                         if (ltdb_msg_add_element(msg2, el) != 0) {
1249                                 ret = LDB_ERR_OTHER;
1250                                 goto done;
1251                         }
1252
1253                         ret = ltdb_index_add_element(module, ltdb,
1254                                                      msg2, el);
1255                         if (ret != LDB_SUCCESS) {
1256                                 goto done;
1257                         }
1258
1259                         break;
1260
1261                 case LDB_FLAG_MOD_DELETE:
1262                         dn = ldb_dn_get_linearized(msg2->dn);
1263                         if (dn == NULL) {
1264                                 ret = LDB_ERR_OTHER;
1265                                 goto done;
1266                         }
1267
1268                         if (msg->elements[i].num_values == 0) {
1269                                 /* Delete the whole attribute */
1270                                 ret = msg_delete_attribute(module,
1271                                                            ltdb,
1272                                                            msg2,
1273                                                            msg->elements[i].name);
1274                                 if (ret == LDB_ERR_NO_SUCH_ATTRIBUTE &&
1275                                     control_permissive) {
1276                                         ret = LDB_SUCCESS;
1277                                 } else {
1278                                         ldb_asprintf_errstring(ldb,
1279                                                                "attribute '%s': no such attribute for delete on '%s'",
1280                                                                msg->elements[i].name, dn);
1281                                 }
1282                                 if (ret != LDB_SUCCESS) {
1283                                         goto done;
1284                                 }
1285                         } else {
1286                                 /* Delete specified values from an attribute */
1287                                 for (j=0; j < msg->elements[i].num_values; j++) {
1288                                         ret = msg_delete_element(module,
1289                                                                  ltdb,
1290                                                                  msg2,
1291                                                                  msg->elements[i].name,
1292                                                                  &msg->elements[i].values[j]);
1293                                         if (ret == LDB_ERR_NO_SUCH_ATTRIBUTE &&
1294                                             control_permissive) {
1295                                                 ret = LDB_SUCCESS;
1296                                         } else if (ret == LDB_ERR_NO_SUCH_ATTRIBUTE) {
1297                                                 ldb_asprintf_errstring(ldb,
1298                                                                        "attribute '%s': no matching attribute value while deleting attribute on '%s'",
1299                                                                        msg->elements[i].name, dn);
1300                                         }
1301                                         if (ret != LDB_SUCCESS) {
1302                                                 goto done;
1303                                         }
1304                                 }
1305                         }
1306                         break;
1307                 default:
1308                         ldb_asprintf_errstring(ldb,
1309                                                "attribute '%s': invalid modify flags on '%s': 0x%x",
1310                                                msg->elements[i].name, ldb_dn_get_linearized(msg->dn),
1311                                                msg->elements[i].flags & LDB_FLAG_MOD_MASK);
1312                         ret = LDB_ERR_PROTOCOL_ERROR;
1313                         goto done;
1314                 }
1315         }
1316
1317         ret = ltdb_store(module, msg2, TDB_MODIFY);
1318         if (ret != LDB_SUCCESS) {
1319                 goto done;
1320         }
1321
1322         ret = ltdb_modified(module, msg2->dn);
1323         if (ret != LDB_SUCCESS) {
1324                 goto done;
1325         }
1326
1327 done:
1328         TALLOC_FREE(mem_ctx);
1329         return ret;
1330 }
1331
1332 /*
1333   modify a record
1334 */
1335 static int ltdb_modify(struct ltdb_context *ctx)
1336 {
1337         struct ldb_module *module = ctx->module;
1338         struct ldb_request *req = ctx->req;
1339         int ret = LDB_SUCCESS;
1340
1341         ret = ltdb_check_special_dn(module, req->op.mod.message);
1342         if (ret != LDB_SUCCESS) {
1343                 return ret;
1344         }
1345
1346         ldb_request_set_state(req, LDB_ASYNC_PENDING);
1347
1348         if (ltdb_cache_load(module) != 0) {
1349                 return LDB_ERR_OPERATIONS_ERROR;
1350         }
1351
1352         ret = ltdb_modify_internal(module, req->op.mod.message, req);
1353
1354         return ret;
1355 }
1356
1357 /*
1358   rename a record
1359 */
1360 static int ltdb_rename(struct ltdb_context *ctx)
1361 {
1362         struct ldb_module *module = ctx->module;
1363         void *data = ldb_module_get_private(module);
1364         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
1365         struct ldb_request *req = ctx->req;
1366         struct ldb_message *msg;
1367         int ret = LDB_SUCCESS;
1368         TDB_DATA tdb_key, tdb_key_old;
1369         struct ldb_dn *db_dn;
1370
1371         ldb_request_set_state(req, LDB_ASYNC_PENDING);
1372
1373         if (ltdb_cache_load(ctx->module) != 0) {
1374                 return LDB_ERR_OPERATIONS_ERROR;
1375         }
1376
1377         msg = ldb_msg_new(ctx);
1378         if (msg == NULL) {
1379                 return LDB_ERR_OPERATIONS_ERROR;
1380         }
1381
1382         /* we need to fetch the old record to re-add under the new name */
1383         ret = ltdb_search_dn1(module, req->op.rename.olddn, msg,
1384                               LDB_UNPACK_DATA_FLAG_NO_DATA_ALLOC);
1385         if (ret != LDB_SUCCESS) {
1386                 /* not finding the old record is an error */
1387                 return ret;
1388         }
1389
1390         /* We need to, before changing the DB, check if the new DN
1391          * exists, so we can return this error to the caller with an
1392          * unmodified DB
1393          *
1394          * Even in GUID index mode we use ltdb_key_dn() as we are
1395          * trying to figure out if this is just a case rename
1396          */
1397         tdb_key = ltdb_key_dn(module, msg, req->op.rename.newdn);
1398         if (!tdb_key.dptr) {
1399                 talloc_free(msg);
1400                 return LDB_ERR_OPERATIONS_ERROR;
1401         }
1402
1403         tdb_key_old = ltdb_key_dn(module, msg, req->op.rename.olddn);
1404         if (!tdb_key_old.dptr) {
1405                 talloc_free(msg);
1406                 talloc_free(tdb_key.dptr);
1407                 return LDB_ERR_OPERATIONS_ERROR;
1408         }
1409
1410         /*
1411          * Only declare a conflict if the new DN already exists,
1412          * and it isn't a case change on the old DN
1413          */
1414         if (tdb_key_old.dsize != tdb_key.dsize
1415             || memcmp(tdb_key.dptr, tdb_key_old.dptr, tdb_key.dsize) != 0) {
1416                 ret = ltdb_search_base(module, msg,
1417                                        req->op.rename.newdn,
1418                                        &db_dn);
1419                 if (ret == LDB_SUCCESS) {
1420                         ret = LDB_ERR_ENTRY_ALREADY_EXISTS;
1421                 } else if (ret == LDB_ERR_NO_SUCH_OBJECT) {
1422                         ret = LDB_SUCCESS;
1423                 }
1424         }
1425
1426         /* finding the new record already in the DB is an error */
1427
1428         if (ret == LDB_ERR_ENTRY_ALREADY_EXISTS) {
1429                 ldb_asprintf_errstring(ldb_module_get_ctx(module),
1430                                        "Entry %s already exists",
1431                                        ldb_dn_get_linearized(req->op.rename.newdn));
1432         }
1433         if (ret != LDB_SUCCESS) {
1434                 talloc_free(tdb_key_old.dptr);
1435                 talloc_free(tdb_key.dptr);
1436                 talloc_free(msg);
1437                 return ret;
1438         }
1439
1440         talloc_free(tdb_key_old.dptr);
1441         talloc_free(tdb_key.dptr);
1442
1443         /* Always delete first then add, to avoid conflicts with
1444          * unique indexes. We rely on the transaction to make this
1445          * atomic
1446          */
1447         ret = ltdb_delete_internal(module, msg->dn);
1448         if (ret != LDB_SUCCESS) {
1449                 talloc_free(msg);
1450                 return ret;
1451         }
1452
1453         msg->dn = ldb_dn_copy(msg, req->op.rename.newdn);
1454         if (msg->dn == NULL) {
1455                 talloc_free(msg);
1456                 return LDB_ERR_OPERATIONS_ERROR;
1457         }
1458
1459         /* We don't check single value as we can have more than 1 with
1460          * deleted attributes. We could go through all elements but that's
1461          * maybe not the most efficient way
1462          */
1463         ret = ltdb_add_internal(module, ltdb, msg, false);
1464
1465         talloc_free(msg);
1466
1467         return ret;
1468 }
1469
1470 static int ltdb_tdb_transaction_start(struct ltdb_private *ltdb)
1471 {
1472         pid_t pid = getpid();
1473
1474         if (ltdb->pid != pid) {
1475                 ldb_asprintf_errstring(
1476                         ldb_module_get_ctx(ltdb->module),
1477                         __location__": Reusing ldb opend by pid %d in "
1478                         "process %d\n",
1479                         ltdb->pid,
1480                         pid);
1481                 return LDB_ERR_PROTOCOL_ERROR;
1482         }
1483
1484         return tdb_transaction_start(ltdb->tdb);
1485 }
1486
1487 static int ltdb_tdb_transaction_cancel(struct ltdb_private *ltdb)
1488 {
1489         pid_t pid = getpid();
1490
1491         if (ltdb->pid != pid) {
1492                 ldb_asprintf_errstring(
1493                         ldb_module_get_ctx(ltdb->module),
1494                         __location__": Reusing ldb opend by pid %d in "
1495                         "process %d\n",
1496                         ltdb->pid,
1497                         pid);
1498                 return LDB_ERR_PROTOCOL_ERROR;
1499         }
1500
1501         return tdb_transaction_cancel(ltdb->tdb);
1502 }
1503
1504 static int ltdb_tdb_transaction_prepare_commit(struct ltdb_private *ltdb)
1505 {
1506         pid_t pid = getpid();
1507
1508         if (ltdb->pid != pid) {
1509                 ldb_asprintf_errstring(
1510                         ldb_module_get_ctx(ltdb->module),
1511                         __location__": Reusing ldb opend by pid %d in "
1512                         "process %d\n",
1513                         ltdb->pid,
1514                         pid);
1515                 return LDB_ERR_PROTOCOL_ERROR;
1516         }
1517
1518         return tdb_transaction_prepare_commit(ltdb->tdb);
1519 }
1520
1521 static int ltdb_tdb_transaction_commit(struct ltdb_private *ltdb)
1522 {
1523         pid_t pid = getpid();
1524
1525         if (ltdb->pid != pid) {
1526                 ldb_asprintf_errstring(
1527                         ldb_module_get_ctx(ltdb->module),
1528                         __location__": Reusing ldb opend by pid %d in "
1529                         "process %d\n",
1530                         ltdb->pid,
1531                         pid);
1532                 return LDB_ERR_PROTOCOL_ERROR;
1533         }
1534
1535         return tdb_transaction_commit(ltdb->tdb);
1536 }
1537
1538 static int ltdb_start_trans(struct ldb_module *module)
1539 {
1540         void *data = ldb_module_get_private(module);
1541         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
1542
1543         pid_t pid = getpid();
1544
1545         if (ltdb->pid != pid) {
1546                 ldb_asprintf_errstring(
1547                         ldb_module_get_ctx(ltdb->module),
1548                         __location__": Reusing ldb opend by pid %d in "
1549                         "process %d\n",
1550                         ltdb->pid,
1551                         pid);
1552                 return LDB_ERR_PROTOCOL_ERROR;
1553         }
1554
1555         /* Do not take out the transaction lock on a read-only DB */
1556         if (ltdb->read_only) {
1557                 return LDB_ERR_UNWILLING_TO_PERFORM;
1558         }
1559
1560         if (ltdb->kv_ops->begin_write(ltdb) != 0) {
1561                 return ltdb->kv_ops->error(ltdb);
1562         }
1563
1564
1565         ltdb_index_transaction_start(module);
1566
1567         ltdb->reindex_failed = false;
1568
1569         return LDB_SUCCESS;
1570 }
1571
1572 /*
1573  * Forward declaration to allow prepare_commit to in fact abort the
1574  * transaction
1575  */
1576 static int ltdb_del_trans(struct ldb_module *module);
1577
1578 static int ltdb_prepare_commit(struct ldb_module *module)
1579 {
1580         int ret;
1581         void *data = ldb_module_get_private(module);
1582         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
1583         pid_t pid = getpid();
1584
1585         if (ltdb->pid != pid) {
1586                 ldb_asprintf_errstring(
1587                         ldb_module_get_ctx(module),
1588                         __location__": Reusing ldb opend by pid %d in "
1589                         "process %d\n",
1590                         ltdb->pid,
1591                         pid);
1592                 return LDB_ERR_PROTOCOL_ERROR;
1593         }
1594
1595         if (!ltdb->kv_ops->transaction_active(ltdb)) {
1596                 ldb_set_errstring(ldb_module_get_ctx(module),
1597                                   "ltdb_prepare_commit() called "
1598                                   "without transaction active");
1599                 return LDB_ERR_OPERATIONS_ERROR;
1600         }
1601
1602         /*
1603          * Check if the last re-index failed.
1604          *
1605          * This can happen if for example a duplicate value was marked
1606          * unique.  We must not write a partial re-index into the DB.
1607          */
1608         if (ltdb->reindex_failed) {
1609                 /*
1610                  * We must instead abort the transaction so we get the
1611                  * old values and old index back
1612                  */
1613                 ltdb_del_trans(module);
1614                 ldb_set_errstring(ldb_module_get_ctx(module),
1615                                   "Failure during re-index, so "
1616                                   "transaction must be aborted.");
1617                 return LDB_ERR_OPERATIONS_ERROR;
1618         }
1619
1620         ret = ltdb_index_transaction_commit(module);
1621         if (ret != LDB_SUCCESS) {
1622                 ltdb->kv_ops->abort_write(ltdb);
1623                 return ret;
1624         }
1625
1626         if (ltdb->kv_ops->prepare_write(ltdb) != 0) {
1627                 ret = ltdb->kv_ops->error(ltdb);
1628                 ldb_debug_set(ldb_module_get_ctx(module),
1629                               LDB_DEBUG_FATAL,
1630                               "Failure during "
1631                               "prepare_write): %s -> %s",
1632                               ltdb->kv_ops->errorstr(ltdb),
1633                               ldb_strerror(ret));
1634                 return ret;
1635         }
1636
1637         ltdb->prepared_commit = true;
1638
1639         return LDB_SUCCESS;
1640 }
1641
1642 static int ltdb_end_trans(struct ldb_module *module)
1643 {
1644         int ret;
1645         void *data = ldb_module_get_private(module);
1646         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
1647
1648         if (!ltdb->prepared_commit) {
1649                 ret = ltdb_prepare_commit(module);
1650                 if (ret != LDB_SUCCESS) {
1651                         return ret;
1652                 }
1653         }
1654
1655         ltdb->prepared_commit = false;
1656
1657         if (ltdb->kv_ops->finish_write(ltdb) != 0) {
1658                 ret = ltdb->kv_ops->error(ltdb);
1659                 ldb_asprintf_errstring(ldb_module_get_ctx(module),
1660                                        "Failure during tdb_transaction_commit(): %s -> %s",
1661                                        ltdb->kv_ops->errorstr(ltdb),
1662                                        ldb_strerror(ret));
1663                 return ret;
1664         }
1665
1666         return LDB_SUCCESS;
1667 }
1668
1669 static int ltdb_del_trans(struct ldb_module *module)
1670 {
1671         void *data = ldb_module_get_private(module);
1672         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
1673
1674
1675         if (ltdb_index_transaction_cancel(module) != 0) {
1676                 ltdb->kv_ops->abort_write(ltdb);
1677                 return ltdb->kv_ops->error(ltdb);
1678         }
1679
1680         ltdb->kv_ops->abort_write(ltdb);
1681         return LDB_SUCCESS;
1682 }
1683
1684 /*
1685   return sequenceNumber from @BASEINFO
1686 */
1687 static int ltdb_sequence_number(struct ltdb_context *ctx,
1688                                 struct ldb_extended **ext)
1689 {
1690         struct ldb_context *ldb;
1691         struct ldb_module *module = ctx->module;
1692         struct ldb_request *req = ctx->req;
1693         void *data = ldb_module_get_private(module);
1694         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
1695         TALLOC_CTX *tmp_ctx = NULL;
1696         struct ldb_seqnum_request *seq;
1697         struct ldb_seqnum_result *res;
1698         struct ldb_message *msg = NULL;
1699         struct ldb_dn *dn;
1700         const char *date;
1701         int ret = LDB_SUCCESS;
1702
1703         ldb = ldb_module_get_ctx(module);
1704
1705         seq = talloc_get_type(req->op.extended.data,
1706                                 struct ldb_seqnum_request);
1707         if (seq == NULL) {
1708                 return LDB_ERR_OPERATIONS_ERROR;
1709         }
1710
1711         ldb_request_set_state(req, LDB_ASYNC_PENDING);
1712
1713         if (ltdb->kv_ops->lock_read(module) != 0) {
1714                 return LDB_ERR_OPERATIONS_ERROR;
1715         }
1716
1717         res = talloc_zero(req, struct ldb_seqnum_result);
1718         if (res == NULL) {
1719                 ret = LDB_ERR_OPERATIONS_ERROR;
1720                 goto done;
1721         }
1722
1723         tmp_ctx = talloc_new(req);
1724         if (tmp_ctx == NULL) {
1725                 ret = LDB_ERR_OPERATIONS_ERROR;
1726                 goto done;
1727         }
1728
1729         dn = ldb_dn_new(tmp_ctx, ldb, LTDB_BASEINFO);
1730         if (dn == NULL) {
1731                 ret = LDB_ERR_OPERATIONS_ERROR;
1732                 goto done;
1733         }
1734
1735         msg = ldb_msg_new(tmp_ctx);
1736         if (msg == NULL) {
1737                 ret = LDB_ERR_OPERATIONS_ERROR;
1738                 goto done;
1739         }
1740
1741         ret = ltdb_search_dn1(module, dn, msg, 0);
1742         if (ret != LDB_SUCCESS) {
1743                 goto done;
1744         }
1745
1746         switch (seq->type) {
1747         case LDB_SEQ_HIGHEST_SEQ:
1748                 res->seq_num = ldb_msg_find_attr_as_uint64(msg, LTDB_SEQUENCE_NUMBER, 0);
1749                 break;
1750         case LDB_SEQ_NEXT:
1751                 res->seq_num = ldb_msg_find_attr_as_uint64(msg, LTDB_SEQUENCE_NUMBER, 0);
1752                 res->seq_num++;
1753                 break;
1754         case LDB_SEQ_HIGHEST_TIMESTAMP:
1755                 date = ldb_msg_find_attr_as_string(msg, LTDB_MOD_TIMESTAMP, NULL);
1756                 if (date) {
1757                         res->seq_num = ldb_string_to_time(date);
1758                 } else {
1759                         res->seq_num = 0;
1760                         /* zero is as good as anything when we don't know */
1761                 }
1762                 break;
1763         }
1764
1765         *ext = talloc_zero(req, struct ldb_extended);
1766         if (*ext == NULL) {
1767                 ret = LDB_ERR_OPERATIONS_ERROR;
1768                 goto done;
1769         }
1770         (*ext)->oid = LDB_EXTENDED_SEQUENCE_NUMBER;
1771         (*ext)->data = talloc_steal(*ext, res);
1772
1773 done:
1774         talloc_free(tmp_ctx);
1775
1776         ltdb->kv_ops->unlock_read(module);
1777         return ret;
1778 }
1779
1780 static void ltdb_request_done(struct ltdb_context *ctx, int error)
1781 {
1782         struct ldb_context *ldb;
1783         struct ldb_request *req;
1784         struct ldb_reply *ares;
1785
1786         ldb = ldb_module_get_ctx(ctx->module);
1787         req = ctx->req;
1788
1789         /* if we already returned an error just return */
1790         if (ldb_request_get_status(req) != LDB_SUCCESS) {
1791                 return;
1792         }
1793
1794         ares = talloc_zero(req, struct ldb_reply);
1795         if (!ares) {
1796                 ldb_oom(ldb);
1797                 req->callback(req, NULL);
1798                 return;
1799         }
1800         ares->type = LDB_REPLY_DONE;
1801         ares->error = error;
1802
1803         req->callback(req, ares);
1804 }
1805
1806 static void ltdb_timeout(struct tevent_context *ev,
1807                           struct tevent_timer *te,
1808                           struct timeval t,
1809                           void *private_data)
1810 {
1811         struct ltdb_context *ctx;
1812         ctx = talloc_get_type(private_data, struct ltdb_context);
1813
1814         if (!ctx->request_terminated) {
1815                 /* request is done now */
1816                 ltdb_request_done(ctx, LDB_ERR_TIME_LIMIT_EXCEEDED);
1817         }
1818
1819         if (ctx->spy) {
1820                 /* neutralize the spy */
1821                 ctx->spy->ctx = NULL;
1822                 ctx->spy = NULL;
1823         }
1824         talloc_free(ctx);
1825 }
1826
1827 static void ltdb_request_extended_done(struct ltdb_context *ctx,
1828                                         struct ldb_extended *ext,
1829                                         int error)
1830 {
1831         struct ldb_context *ldb;
1832         struct ldb_request *req;
1833         struct ldb_reply *ares;
1834
1835         ldb = ldb_module_get_ctx(ctx->module);
1836         req = ctx->req;
1837
1838         /* if we already returned an error just return */
1839         if (ldb_request_get_status(req) != LDB_SUCCESS) {
1840                 return;
1841         }
1842
1843         ares = talloc_zero(req, struct ldb_reply);
1844         if (!ares) {
1845                 ldb_oom(ldb);
1846                 req->callback(req, NULL);
1847                 return;
1848         }
1849         ares->type = LDB_REPLY_DONE;
1850         ares->response = ext;
1851         ares->error = error;
1852
1853         req->callback(req, ares);
1854 }
1855
1856 static void ltdb_handle_extended(struct ltdb_context *ctx)
1857 {
1858         struct ldb_extended *ext = NULL;
1859         int ret;
1860
1861         if (strcmp(ctx->req->op.extended.oid,
1862                    LDB_EXTENDED_SEQUENCE_NUMBER) == 0) {
1863                 /* get sequence number */
1864                 ret = ltdb_sequence_number(ctx, &ext);
1865         } else {
1866                 /* not recognized */
1867                 ret = LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
1868         }
1869
1870         ltdb_request_extended_done(ctx, ext, ret);
1871 }
1872
1873 struct kv_ctx {
1874         ldb_kv_traverse_fn kv_traverse_fn;
1875         void *ctx;
1876         struct ltdb_private *ltdb;
1877         int (*parser)(struct ldb_val key,
1878                       struct ldb_val data,
1879                       void *private_data);
1880 };
1881
1882 static int ldb_tdb_traverse_fn_wrapper(struct tdb_context *tdb, TDB_DATA tdb_key, TDB_DATA tdb_data, void *ctx)
1883 {
1884         struct kv_ctx *kv_ctx = ctx;
1885         struct ldb_val key = {
1886                 .length = tdb_key.dsize,
1887                 .data = tdb_key.dptr,
1888         };
1889         struct ldb_val data = {
1890                 .length = tdb_data.dsize,
1891                 .data = tdb_data.dptr,
1892         };
1893         return kv_ctx->kv_traverse_fn(kv_ctx->ltdb, key, data, kv_ctx->ctx);
1894 }
1895
1896 static int ltdb_tdb_traverse_fn(struct ltdb_private *ltdb, ldb_kv_traverse_fn fn, void *ctx)
1897 {
1898         struct kv_ctx kv_ctx = {
1899                 .kv_traverse_fn = fn,
1900                 .ctx = ctx,
1901                 .ltdb = ltdb
1902         };
1903         if (tdb_transaction_active(ltdb->tdb)) {
1904                 return tdb_traverse(ltdb->tdb, ldb_tdb_traverse_fn_wrapper, &kv_ctx);
1905         } else {
1906                 return tdb_traverse_read(ltdb->tdb, ldb_tdb_traverse_fn_wrapper, &kv_ctx);
1907         }
1908 }
1909
1910 static int ltdb_tdb_update_in_iterate(struct ltdb_private *ltdb,
1911                                       struct ldb_val ldb_key,
1912                                       struct ldb_val ldb_key2,
1913                                       struct ldb_val ldb_data, void *state)
1914 {
1915         int tdb_ret;
1916         struct ldb_context *ldb;
1917         struct ltdb_reindex_context *ctx = (struct ltdb_reindex_context *)state;
1918         struct ldb_module *module = ctx->module;
1919         TDB_DATA key = {
1920                 .dptr = ldb_key.data,
1921                 .dsize = ldb_key.length
1922         };
1923         TDB_DATA key2 = {
1924                 .dptr = ldb_key2.data,
1925                 .dsize = ldb_key2.length
1926         };
1927         TDB_DATA data = {
1928                 .dptr = ldb_data.data,
1929                 .dsize = ldb_data.length
1930         };
1931
1932         ldb = ldb_module_get_ctx(module);
1933
1934         tdb_ret = tdb_delete(ltdb->tdb, key);
1935         if (tdb_ret != 0) {
1936                 ldb_debug(ldb, LDB_DEBUG_ERROR,
1937                           "Failed to delete %*.*s "
1938                           "for rekey as %*.*s: %s",
1939                           (int)key.dsize, (int)key.dsize,
1940                           (const char *)key.dptr,
1941                           (int)key2.dsize, (int)key2.dsize,
1942                           (const char *)key.dptr,
1943                           tdb_errorstr(ltdb->tdb));
1944                 ctx->error = ltdb_err_map(tdb_error(ltdb->tdb));
1945                 return -1;
1946         }
1947         tdb_ret = tdb_store(ltdb->tdb, key2, data, 0);
1948         if (tdb_ret != 0) {
1949                 ldb_debug(ldb, LDB_DEBUG_ERROR,
1950                           "Failed to rekey %*.*s as %*.*s: %s",
1951                           (int)key.dsize, (int)key.dsize,
1952                           (const char *)key.dptr,
1953                           (int)key2.dsize, (int)key2.dsize,
1954                           (const char *)key.dptr,
1955                           tdb_errorstr(ltdb->tdb));
1956                 ctx->error = ltdb_err_map(tdb_error(ltdb->tdb));
1957                 return -1;
1958         }
1959         return tdb_ret;
1960 }
1961
1962 static int ltdb_tdb_parse_record_wrapper(TDB_DATA tdb_key, TDB_DATA tdb_data,
1963                                          void *ctx)
1964 {
1965         struct kv_ctx *kv_ctx = ctx;
1966         struct ldb_val key = {
1967                 .length = tdb_key.dsize,
1968                 .data = tdb_key.dptr,
1969         };
1970         struct ldb_val data = {
1971                 .length = tdb_data.dsize,
1972                 .data = tdb_data.dptr,
1973         };
1974
1975         return kv_ctx->parser(key, data, kv_ctx->ctx);
1976 }
1977
1978 static int ltdb_tdb_parse_record(struct ltdb_private *ltdb,
1979                                  struct ldb_val ldb_key,
1980                                  int (*parser)(struct ldb_val key,
1981                                                struct ldb_val data,
1982                                                void *private_data),
1983                                  void *ctx)
1984 {
1985         struct kv_ctx kv_ctx = {
1986                 .parser = parser,
1987                 .ctx = ctx,
1988                 .ltdb = ltdb
1989         };
1990         TDB_DATA key = {
1991                 .dptr = ldb_key.data,
1992                 .dsize = ldb_key.length
1993         };
1994         int ret;
1995
1996         if (tdb_transaction_active(ltdb->tdb) == false &&
1997             ltdb->read_lock_count == 0) {
1998                 return LDB_ERR_PROTOCOL_ERROR;
1999         }
2000
2001         ret = tdb_parse_record(ltdb->tdb, key, ltdb_tdb_parse_record_wrapper,
2002                                &kv_ctx);
2003         if (ret == 0) {
2004                 return LDB_SUCCESS;
2005         }
2006         return ltdb_err_map(tdb_error(ltdb->tdb));
2007 }
2008
2009 static const char * ltdb_tdb_name(struct ltdb_private *ltdb)
2010 {
2011         return tdb_name(ltdb->tdb);
2012 }
2013
2014 static bool ltdb_tdb_changed(struct ltdb_private *ltdb)
2015 {
2016         int seq = tdb_get_seqnum(ltdb->tdb);
2017         bool has_changed = (seq != ltdb->tdb_seqnum);
2018
2019         ltdb->tdb_seqnum = seq;
2020
2021         return has_changed;
2022 }
2023
2024 static bool ltdb_transaction_active(struct ltdb_private *ltdb)
2025 {
2026         return tdb_transaction_active(ltdb->tdb);
2027 }
2028
2029 static const struct kv_db_ops key_value_ops = {
2030         .store = ltdb_tdb_store,
2031         .delete = ltdb_tdb_delete,
2032         .iterate = ltdb_tdb_traverse_fn,
2033         .update_in_iterate = ltdb_tdb_update_in_iterate,
2034         .fetch_and_parse = ltdb_tdb_parse_record,
2035         .lock_read = ltdb_lock_read,
2036         .unlock_read = ltdb_unlock_read,
2037         .begin_write = ltdb_tdb_transaction_start,
2038         .prepare_write = ltdb_tdb_transaction_prepare_commit,
2039         .finish_write = ltdb_tdb_transaction_commit,
2040         .abort_write = ltdb_tdb_transaction_cancel,
2041         .error = ltdb_error,
2042         .errorstr = ltdb_errorstr,
2043         .name = ltdb_tdb_name,
2044         .has_changed = ltdb_tdb_changed,
2045         .transaction_active = ltdb_transaction_active,
2046 };
2047
2048 static void ltdb_callback(struct tevent_context *ev,
2049                           struct tevent_timer *te,
2050                           struct timeval t,
2051                           void *private_data)
2052 {
2053         struct ltdb_context *ctx;
2054         int ret;
2055
2056         ctx = talloc_get_type(private_data, struct ltdb_context);
2057
2058         if (ctx->request_terminated) {
2059                 goto done;
2060         }
2061
2062         switch (ctx->req->operation) {
2063         case LDB_SEARCH:
2064                 ret = ltdb_search(ctx);
2065                 break;
2066         case LDB_ADD:
2067                 ret = ltdb_add(ctx);
2068                 break;
2069         case LDB_MODIFY:
2070                 ret = ltdb_modify(ctx);
2071                 break;
2072         case LDB_DELETE:
2073                 ret = ltdb_delete(ctx);
2074                 break;
2075         case LDB_RENAME:
2076                 ret = ltdb_rename(ctx);
2077                 break;
2078         case LDB_EXTENDED:
2079                 ltdb_handle_extended(ctx);
2080                 goto done;
2081         default:
2082                 /* no other op supported */
2083                 ret = LDB_ERR_PROTOCOL_ERROR;
2084         }
2085
2086         if (!ctx->request_terminated) {
2087                 /* request is done now */
2088                 ltdb_request_done(ctx, ret);
2089         }
2090
2091 done:
2092         if (ctx->spy) {
2093                 /* neutralize the spy */
2094                 ctx->spy->ctx = NULL;
2095                 ctx->spy = NULL;
2096         }
2097         talloc_free(ctx);
2098 }
2099
2100 static int ltdb_request_destructor(void *ptr)
2101 {
2102         struct ltdb_req_spy *spy = talloc_get_type(ptr, struct ltdb_req_spy);
2103
2104         if (spy->ctx != NULL) {
2105                 spy->ctx->spy = NULL;
2106                 spy->ctx->request_terminated = true;
2107                 spy->ctx = NULL;
2108         }
2109
2110         return 0;
2111 }
2112
2113 static int ltdb_handle_request(struct ldb_module *module,
2114                                struct ldb_request *req)
2115 {
2116         struct ldb_control *control_permissive;
2117         struct ldb_context *ldb;
2118         struct tevent_context *ev;
2119         struct ltdb_context *ac;
2120         struct tevent_timer *te;
2121         struct timeval tv;
2122         unsigned int i;
2123
2124         ldb = ldb_module_get_ctx(module);
2125
2126         control_permissive = ldb_request_get_control(req,
2127                                         LDB_CONTROL_PERMISSIVE_MODIFY_OID);
2128
2129         for (i = 0; req->controls && req->controls[i]; i++) {
2130                 if (req->controls[i]->critical &&
2131                     req->controls[i] != control_permissive) {
2132                         ldb_asprintf_errstring(ldb, "Unsupported critical extension %s",
2133                                                req->controls[i]->oid);
2134                         return LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
2135                 }
2136         }
2137
2138         if (req->starttime == 0 || req->timeout == 0) {
2139                 ldb_set_errstring(ldb, "Invalid timeout settings");
2140                 return LDB_ERR_TIME_LIMIT_EXCEEDED;
2141         }
2142
2143         ev = ldb_handle_get_event_context(req->handle);
2144
2145         ac = talloc_zero(ldb, struct ltdb_context);
2146         if (ac == NULL) {
2147                 ldb_oom(ldb);
2148                 return LDB_ERR_OPERATIONS_ERROR;
2149         }
2150
2151         ac->module = module;
2152         ac->req = req;
2153
2154         tv.tv_sec = 0;
2155         tv.tv_usec = 0;
2156         te = tevent_add_timer(ev, ac, tv, ltdb_callback, ac);
2157         if (NULL == te) {
2158                 talloc_free(ac);
2159                 return LDB_ERR_OPERATIONS_ERROR;
2160         }
2161
2162         if (req->timeout > 0) {
2163                 tv.tv_sec = req->starttime + req->timeout;
2164                 tv.tv_usec = 0;
2165                 ac->timeout_event = tevent_add_timer(ev, ac, tv,
2166                                                      ltdb_timeout, ac);
2167                 if (NULL == ac->timeout_event) {
2168                         talloc_free(ac);
2169                         return LDB_ERR_OPERATIONS_ERROR;
2170                 }
2171         }
2172
2173         /* set a spy so that we do not try to use the request context
2174          * if it is freed before ltdb_callback fires */
2175         ac->spy = talloc(req, struct ltdb_req_spy);
2176         if (NULL == ac->spy) {
2177                 talloc_free(ac);
2178                 return LDB_ERR_OPERATIONS_ERROR;
2179         }
2180         ac->spy->ctx = ac;
2181
2182         talloc_set_destructor((TALLOC_CTX *)ac->spy, ltdb_request_destructor);
2183
2184         return LDB_SUCCESS;
2185 }
2186
2187 static int ltdb_init_rootdse(struct ldb_module *module)
2188 {
2189         /* ignore errors on this - we expect it for non-sam databases */
2190         ldb_mod_register_control(module, LDB_CONTROL_PERMISSIVE_MODIFY_OID);
2191
2192         /* there can be no module beyond the backend, just return */
2193         return LDB_SUCCESS;
2194 }
2195
2196
2197 static int generic_lock_read(struct ldb_module *module)
2198 {
2199         void *data = ldb_module_get_private(module);
2200         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
2201         return ltdb->kv_ops->lock_read(module);
2202 }
2203
2204 static int generic_unlock_read(struct ldb_module *module)
2205 {
2206         void *data = ldb_module_get_private(module);
2207         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
2208         return ltdb->kv_ops->unlock_read(module);
2209 }
2210
2211 static const struct ldb_module_ops ltdb_ops = {
2212         .name              = "tdb",
2213         .init_context      = ltdb_init_rootdse,
2214         .search            = ltdb_handle_request,
2215         .add               = ltdb_handle_request,
2216         .modify            = ltdb_handle_request,
2217         .del               = ltdb_handle_request,
2218         .rename            = ltdb_handle_request,
2219         .extended          = ltdb_handle_request,
2220         .start_transaction = ltdb_start_trans,
2221         .end_transaction   = ltdb_end_trans,
2222         .prepare_commit    = ltdb_prepare_commit,
2223         .del_transaction   = ltdb_del_trans,
2224         .read_lock         = generic_lock_read,
2225         .read_unlock       = generic_unlock_read,
2226 };
2227
2228 int init_store(struct ltdb_private *ltdb,
2229                       const char *name,
2230                       struct ldb_context *ldb,
2231                       const char *options[],
2232                       struct ldb_module **_module)
2233 {
2234         if (getenv("LDB_WARN_UNINDEXED")) {
2235                 ltdb->warn_unindexed = true;
2236         }
2237
2238         if (getenv("LDB_WARN_REINDEX")) {
2239                 ltdb->warn_reindex = true;
2240         }
2241
2242         ltdb->sequence_number = 0;
2243
2244         ltdb->pid = getpid();
2245
2246         ltdb->module = ldb_module_new(ldb, ldb, name, &ltdb_ops);
2247         if (!ltdb->module) {
2248                 ldb_oom(ldb);
2249                 talloc_free(ltdb);
2250                 return LDB_ERR_OPERATIONS_ERROR;
2251         }
2252         ldb_module_set_private(ltdb->module, ltdb);
2253         talloc_steal(ltdb->module, ltdb);
2254
2255         if (ltdb_cache_load(ltdb->module) != 0) {
2256                 ldb_asprintf_errstring(ldb, "Unable to load ltdb cache "
2257                                        "records for backend '%s'", name);
2258                 talloc_free(ltdb->module);
2259                 return LDB_ERR_OPERATIONS_ERROR;
2260         }
2261
2262         *_module = ltdb->module;
2263         /*
2264          * Set or override the maximum key length
2265          *
2266          * The ldb_mdb code will have set this to 511, but our tests
2267          * set this even smaller (to make the tests more practical).
2268          *
2269          * This must only be used for the selftest as the length
2270          * becomes encoded in the index keys.
2271          */
2272         {
2273                 const char *len_str =
2274                         ldb_options_find(ldb, options,
2275                                          "max_key_len_for_self_test");
2276                 if (len_str != NULL) {
2277                         unsigned len = strtoul(len_str, NULL, 0);
2278                         ltdb->max_key_length = len;
2279                 }
2280         }
2281
2282         /*
2283          * Override full DB scans
2284          *
2285          * A full DB scan is expensive on a large database.  This
2286          * option is for testing to show that the full DB scan is not
2287          * triggered.
2288          */
2289         {
2290                 const char *len_str =
2291                         ldb_options_find(ldb, options,
2292                                          "disable_full_db_scan_for_self_test");
2293                 if (len_str != NULL) {
2294                         ltdb->disable_full_db_scan = true;
2295                 }
2296         }
2297
2298         return LDB_SUCCESS;
2299 }
2300
2301 /*
2302   connect to the database
2303 */
2304 int ltdb_connect(struct ldb_context *ldb, const char *url,
2305                  unsigned int flags, const char *options[],
2306                  struct ldb_module **_module)
2307 {
2308         const char *path;
2309         int tdb_flags, open_flags;
2310         struct ltdb_private *ltdb;
2311
2312         /*
2313          * We hold locks, so we must use a private event context
2314          * on each returned handle
2315          */
2316         ldb_set_require_private_event_context(ldb);
2317
2318         /* parse the url */
2319         if (strchr(url, ':')) {
2320                 if (strncmp(url, "tdb://", 6) != 0) {
2321                         ldb_debug(ldb, LDB_DEBUG_ERROR,
2322                                   "Invalid tdb URL '%s'", url);
2323                         return LDB_ERR_OPERATIONS_ERROR;
2324                 }
2325                 path = url+6;
2326         } else {
2327                 path = url;
2328         }
2329
2330         tdb_flags = TDB_DEFAULT | TDB_SEQNUM | TDB_DISALLOW_NESTING;
2331
2332         /* check for the 'nosync' option */
2333         if (flags & LDB_FLG_NOSYNC) {
2334                 tdb_flags |= TDB_NOSYNC;
2335         }
2336
2337         /* and nommap option */
2338         if (flags & LDB_FLG_NOMMAP) {
2339                 tdb_flags |= TDB_NOMMAP;
2340         }
2341
2342         ltdb = talloc_zero(ldb, struct ltdb_private);
2343         if (!ltdb) {
2344                 ldb_oom(ldb);
2345                 return LDB_ERR_OPERATIONS_ERROR;
2346         }
2347
2348         if (flags & LDB_FLG_RDONLY) {
2349                 /*
2350                  * This is weird, but because we can only have one tdb
2351                  * in this process, and the other one could be
2352                  * read-write, we can't use the tdb readonly.  Plus a
2353                  * read only tdb prohibits the all-record lock.
2354                  */
2355                 open_flags = O_RDWR;
2356
2357                 ltdb->read_only = true;
2358
2359         } else if (flags & LDB_FLG_DONT_CREATE_DB) {
2360                 /*
2361                  * This is used by ldbsearch to prevent creation of the database
2362                  * if the name is wrong
2363                  */
2364                 open_flags = O_RDWR;
2365         } else {
2366                 /*
2367                  * This is the normal case
2368                  */
2369                 open_flags = O_CREAT | O_RDWR;
2370         }
2371
2372         ltdb->kv_ops = &key_value_ops;
2373
2374         errno = 0;
2375         /* note that we use quite a large default hash size */
2376         ltdb->tdb = ltdb_wrap_open(ltdb, path, 10000,
2377                                    tdb_flags, open_flags,
2378                                    ldb_get_create_perms(ldb), ldb);
2379         if (!ltdb->tdb) {
2380                 ldb_asprintf_errstring(ldb,
2381                                        "Unable to open tdb '%s': %s", path, strerror(errno));
2382                 ldb_debug(ldb, LDB_DEBUG_ERROR,
2383                           "Unable to open tdb '%s': %s", path, strerror(errno));
2384                 talloc_free(ltdb);
2385                 if (errno == EACCES || errno == EPERM) {
2386                         return LDB_ERR_INSUFFICIENT_ACCESS_RIGHTS;
2387                 }
2388                 return LDB_ERR_OPERATIONS_ERROR;
2389         }
2390
2391         return init_store(ltdb, "ldb_tdb backend", ldb, options, _module);
2392 }