source4/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
   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 asyncronous 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
  54
  55 /*
  56   map a tdb error code to a ldb error code
  57 */
  58 int ltdb_err_map(enum TDB_ERROR tdb_code)
  59 {
  60         switch (tdb_code) {
  61         case TDB_SUCCESS:
  62                 return LDB_SUCCESS;
  63         case TDB_ERR_CORRUPT:
  64         case TDB_ERR_OOM:
  65         case TDB_ERR_EINVAL:
  66                 return LDB_ERR_OPERATIONS_ERROR;
  67         case TDB_ERR_IO:
  68                 return LDB_ERR_PROTOCOL_ERROR;
  69         case TDB_ERR_LOCK:
  70         case TDB_ERR_NOLOCK:
  71                 return LDB_ERR_BUSY;
  72         case TDB_ERR_LOCK_TIMEOUT:
  73                 return LDB_ERR_TIME_LIMIT_EXCEEDED;
  74         case TDB_ERR_EXISTS:
  75                 return LDB_ERR_ENTRY_ALREADY_EXISTS;
  76         case TDB_ERR_NOEXIST:
  77                 return LDB_ERR_NO_SUCH_OBJECT;
  78         case TDB_ERR_RDONLY:
  79                 return LDB_ERR_INSUFFICIENT_ACCESS_RIGHTS;
  80         }
  81         return LDB_ERR_OTHER;
  82 }
  83
  84 /*
  85   lock the database for read - use by ltdb_search and ltdb_sequence_number
  86 */
  87 int ltdb_lock_read(struct ldb_module *module)
  88 {
  89         void *data = ldb_module_get_private(module);
  90         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
  91         if (ltdb->in_transaction == 0) {
  92                 return tdb_lockall_read(ltdb->tdb);
  93         }
  94         return 0;
  95 }
  96
  97 /*
  98   unlock the database after a ltdb_lock_read()
  99 */
 100 int ltdb_unlock_read(struct ldb_module *module)
 101 {
 102         void *data = ldb_module_get_private(module);
 103         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 104         if (ltdb->in_transaction == 0) {
 105                 return tdb_unlockall_read(ltdb->tdb);
 106         }
 107         return 0;
 108 }
 109
 110
 111 /*
 112   form a TDB_DATA for a record key
 113   caller frees
 114
 115   note that the key for a record can depend on whether the
 116   dn refers to a case sensitive index record or not
 117 */
 118 struct TDB_DATA ltdb_key(struct ldb_module *module, struct ldb_dn *dn)
 119 {
 120         struct ldb_context *ldb = ldb_module_get_ctx(module);
 121         TDB_DATA key;
 122         char *key_str = NULL;
 123         const char *dn_folded = NULL;
 124
 125         /*
 126           most DNs are case insensitive. The exception is index DNs for
 127           case sensitive attributes
 128
 129           there are 3 cases dealt with in this code:
 130
 131           1) if the dn doesn't start with @ then uppercase the attribute
 132              names and the attributes values of case insensitive attributes
 133           2) if the dn starts with @ then leave it alone -
 134              the indexing code handles the rest
 135         */
 136
 137         dn_folded = ldb_dn_get_casefold(dn);
 138         if (!dn_folded) {
 139                 goto failed;
 140         }
 141
 142         key_str = talloc_strdup(ldb, "DN=");
 143         if (!key_str) {
 144                 goto failed;
 145         }
 146
 147         key_str = talloc_strdup_append_buffer(key_str, dn_folded);
 148         if (!key_str) {
 149                 goto failed;
 150         }
 151
 152         key.dptr = (uint8_t *)key_str;
 153         key.dsize = strlen(key_str) + 1;
 154
 155         return key;
 156
 157 failed:
 158         errno = ENOMEM;
 159         key.dptr = NULL;
 160         key.dsize = 0;
 161         return key;
 162 }
 163
 164 /*
 165   check special dn's have valid attributes
 166   currently only @ATTRIBUTES is checked
 167 */
 168 static int ltdb_check_special_dn(struct ldb_module *module,
 169                           const struct ldb_message *msg)
 170 {
 171         struct ldb_context *ldb = ldb_module_get_ctx(module);
 172         int i, j;
 173
 174         if (! ldb_dn_is_special(msg->dn) ||
 175             ! ldb_dn_check_special(msg->dn, LTDB_ATTRIBUTES)) {
 176                 return LDB_SUCCESS;
 177         }
 178
 179         /* we have @ATTRIBUTES, let's check attributes are fine */
 180         /* should we check that we deny multivalued attributes ? */
 181         for (i = 0; i < msg->num_elements; i++) {
 182                 if (ldb_attr_cmp(msg->elements[i].name, "distinguishedName") == 0) continue;
 183
 184                 for (j = 0; j < msg->elements[i].num_values; j++) {
 185                         if (ltdb_check_at_attributes_values(&msg->elements[i].values[j]) != 0) {
 186                                 ldb_set_errstring(ldb, "Invalid attribute value in an @ATTRIBUTES entry");
 187                                 return LDB_ERR_INVALID_ATTRIBUTE_SYNTAX;
 188                         }
 189                 }
 190         }
 191
 192         return LDB_SUCCESS;
 193 }
 194
 195
 196 /*
 197   we've made a modification to a dn - possibly reindex and
 198   update sequence number
 199 */
 200 static int ltdb_modified(struct ldb_module *module, struct ldb_dn *dn)
 201 {
 202         int ret = LDB_SUCCESS;
 203         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(module), struct ltdb_private);
 204
 205         /* only allow modifies inside a transaction, otherwise the
 206          * ldb is unsafe */
 207         if (ltdb->in_transaction == 0) {
 208                 ldb_set_errstring(ldb_module_get_ctx(module), "ltdb modify without transaction");
 209                 return LDB_ERR_OPERATIONS_ERROR;
 210         }
 211
 212         if (ldb_dn_is_special(dn) &&
 213             (ldb_dn_check_special(dn, LTDB_INDEXLIST) ||
 214              ldb_dn_check_special(dn, LTDB_ATTRIBUTES)) ) {
 215                 ret = ltdb_reindex(module);
 216         }
 217
 218         /* If the modify was to a normal record, or any special except @BASEINFO, update the seq number */
 219         if (ret == LDB_SUCCESS &&
 220             !(ldb_dn_is_special(dn) &&
 221               ldb_dn_check_special(dn, LTDB_BASEINFO)) ) {
 222                 ret = ltdb_increase_sequence_number(module);
 223         }
 224
 225         /* If the modify was to @OPTIONS, reload the cache */
 226         if (ldb_dn_is_special(dn) &&
 227             (ldb_dn_check_special(dn, LTDB_OPTIONS)) ) {
 228                 ret = ltdb_cache_reload(module);
 229         }
 230
 231         return ret;
 232 }
 233
 234 /*
 235   store a record into the db
 236 */
 237 int ltdb_store(struct ldb_module *module, const struct ldb_message *msg, int flgs)
 238 {
 239         void *data = ldb_module_get_private(module);
 240         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 241         TDB_DATA tdb_key, tdb_data;
 242         int ret = LDB_SUCCESS;
 243
 244         tdb_key = ltdb_key(module, msg->dn);
 245         if (tdb_key.dptr == NULL) {
 246                 return LDB_ERR_OTHER;
 247         }
 248
 249         ret = ltdb_pack_data(module, msg, &tdb_data);
 250         if (ret == -1) {
 251                 talloc_free(tdb_key.dptr);
 252                 return LDB_ERR_OTHER;
 253         }
 254
 255         ret = tdb_store(ltdb->tdb, tdb_key, tdb_data, flgs);
 256         if (ret == -1) {
 257                 ret = ltdb_err_map(tdb_error(ltdb->tdb));
 258                 goto done;
 259         }
 260
 261 done:
 262         talloc_free(tdb_key.dptr);
 263         talloc_free(tdb_data.dptr);
 264
 265         return ret;
 266 }
 267
 268
 269 static int ltdb_add_internal(struct ldb_module *module,
 270                              const struct ldb_message *msg)
 271 {
 272         struct ldb_context *ldb = ldb_module_get_ctx(module);
 273         int ret = LDB_SUCCESS, i;
 274
 275         ret = ltdb_check_special_dn(module, msg);
 276         if (ret != LDB_SUCCESS) {
 277                 return ret;
 278         }
 279
 280         if (ltdb_cache_load(module) != 0) {
 281                 return LDB_ERR_OPERATIONS_ERROR;
 282         }
 283
 284         for (i=0;i<msg->num_elements;i++) {
 285                 struct ldb_message_element *el = &msg->elements[i];
 286                 const struct ldb_schema_attribute *a = ldb_schema_attribute_by_name(ldb, el->name);
 287
 288                 if (el->num_values == 0) {
 289                         ldb_asprintf_errstring(ldb, "attribute %s on %s specified, but with 0 values (illegal)",
 290                                                el->name, ldb_dn_get_linearized(msg->dn));
 291                         return LDB_ERR_CONSTRAINT_VIOLATION;
 292                 }
 293                 if (a && a->flags & LDB_ATTR_FLAG_SINGLE_VALUE) {
 294                         if (el->num_values > 1) {
 295                                 ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
 296                                                        el->name, ldb_dn_get_linearized(msg->dn));
 297                                 return LDB_ERR_CONSTRAINT_VIOLATION;
 298                         }
 299                 }
 300         }
 301
 302         ret = ltdb_store(module, msg, TDB_INSERT);
 303         if (ret != LDB_SUCCESS) {
 304                 if (ret == LDB_ERR_ENTRY_ALREADY_EXISTS) {
 305                         ldb_asprintf_errstring(ldb,
 306                                                "Entry %s already exists",
 307                                                ldb_dn_get_linearized(msg->dn));
 308                 }
 309                 return ret;
 310         }
 311
 312         ret = ltdb_index_add_new(module, msg);
 313         if (ret != LDB_SUCCESS) {
 314                 return ret;
 315         }
 316
 317         ret = ltdb_modified(module, msg->dn);
 318
 319         return ret;
 320 }
 321
 322 /*
 323   add a record to the database
 324 */
 325 static int ltdb_add(struct ltdb_context *ctx)
 326 {
 327         struct ldb_module *module = ctx->module;
 328         struct ldb_request *req = ctx->req;
 329         int ret = LDB_SUCCESS;
 330
 331         ldb_request_set_state(req, LDB_ASYNC_PENDING);
 332
 333         if (ltdb_cache_load(module) != 0) {
 334                 return LDB_ERR_OPERATIONS_ERROR;
 335         }
 336
 337         ret = ltdb_add_internal(module, req->op.add.message);
 338
 339         return ret;
 340 }
 341
 342 /*
 343   delete a record from the database, not updating indexes (used for deleting
 344   index records)
 345 */
 346 int ltdb_delete_noindex(struct ldb_module *module, struct ldb_dn *dn)
 347 {
 348         void *data = ldb_module_get_private(module);
 349         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 350         TDB_DATA tdb_key;
 351         int ret;
 352
 353         tdb_key = ltdb_key(module, dn);
 354         if (!tdb_key.dptr) {
 355                 return LDB_ERR_OTHER;
 356         }
 357
 358         ret = tdb_delete(ltdb->tdb, tdb_key);
 359         talloc_free(tdb_key.dptr);
 360
 361         if (ret != 0) {
 362                 ret = ltdb_err_map(tdb_error(ltdb->tdb));
 363         }
 364
 365         return ret;
 366 }
 367
 368 static int ltdb_delete_internal(struct ldb_module *module, struct ldb_dn *dn)
 369 {
 370         struct ldb_message *msg;
 371         int ret = LDB_SUCCESS;
 372
 373         msg = talloc(module, struct ldb_message);
 374         if (msg == NULL) {
 375                 return LDB_ERR_OPERATIONS_ERROR;
 376         }
 377
 378         /* in case any attribute of the message was indexed, we need
 379            to fetch the old record */
 380         ret = ltdb_search_dn1(module, dn, msg);
 381         if (ret != LDB_SUCCESS) {
 382                 /* not finding the old record is an error */
 383                 goto done;
 384         }
 385
 386         ret = ltdb_delete_noindex(module, dn);
 387         if (ret != LDB_SUCCESS) {
 388                 goto done;
 389         }
 390
 391         /* remove any indexed attributes */
 392         ret = ltdb_index_delete(module, msg);
 393         if (ret != LDB_SUCCESS) {
 394                 goto done;
 395         }
 396
 397         ret = ltdb_modified(module, dn);
 398         if (ret != LDB_SUCCESS) {
 399                 goto done;
 400         }
 401
 402 done:
 403         talloc_free(msg);
 404         return ret;
 405 }
 406
 407 /*
 408   delete a record from the database
 409 */
 410 static int ltdb_delete(struct ltdb_context *ctx)
 411 {
 412         struct ldb_module *module = ctx->module;
 413         struct ldb_request *req = ctx->req;
 414         int ret = LDB_SUCCESS;
 415
 416         ldb_request_set_state(req, LDB_ASYNC_PENDING);
 417
 418         if (ltdb_cache_load(module) != 0) {
 419                 return LDB_ERR_OPERATIONS_ERROR;
 420         }
 421
 422         ret = ltdb_delete_internal(module, req->op.del.dn);
 423
 424         return ret;
 425 }
 426
 427 /*
 428   find an element by attribute name. At the moment this does a linear search,
 429   it should be re-coded to use a binary search once all places that modify
 430   records guarantee sorted order
 431
 432   return the index of the first matching element if found, otherwise -1
 433 */
 434 static int find_element(const struct ldb_message *msg, const char *name)
 435 {
 436         unsigned int i;
 437         for (i=0;i<msg->num_elements;i++) {
 438                 if (ldb_attr_cmp(msg->elements[i].name, name) == 0) {
 439                         return i;
 440                 }
 441         }
 442         return -1;
 443 }
 444
 445
 446 /*
 447   add an element to an existing record. Assumes a elements array that we
 448   can call re-alloc on, and assumed that we can re-use the data pointers from
 449   the passed in additional values. Use with care!
 450
 451   returns 0 on success, -1 on failure (and sets errno)
 452 */
 453 static int ltdb_msg_add_element(struct ldb_context *ldb,
 454                                 struct ldb_message *msg,
 455                                 struct ldb_message_element *el)
 456 {
 457         struct ldb_message_element *e2;
 458         unsigned int i;
 459
 460         if (el->num_values == 0) {
 461                 /* nothing to do here - we don't add empty elements */
 462                 return 0;
 463         }
 464
 465         e2 = talloc_realloc(msg, msg->elements, struct ldb_message_element,
 466                               msg->num_elements+1);
 467         if (!e2) {
 468                 errno = ENOMEM;
 469                 return -1;
 470         }
 471
 472         msg->elements = e2;
 473
 474         e2 = &msg->elements[msg->num_elements];
 475
 476         e2->name = el->name;
 477         e2->flags = el->flags;
 478         e2->values = talloc_array(msg->elements,
 479                                   struct ldb_val, el->num_values);
 480         if (!e2->values) {
 481                 errno = ENOMEM;
 482                 return -1;
 483         }
 484         for (i=0;i<el->num_values;i++) {
 485                 e2->values[i] = el->values[i];
 486         }
 487         e2->num_values = el->num_values;
 488
 489         ++msg->num_elements;
 490
 491         return 0;
 492 }
 493
 494 /*
 495   delete all elements having a specified attribute name
 496 */
 497 static int msg_delete_attribute(struct ldb_module *module,
 498                                 struct ldb_context *ldb,
 499                                 struct ldb_message *msg, const char *name)
 500 {
 501         unsigned int i;
 502         int ret;
 503         struct ldb_message_element *el;
 504
 505         el = ldb_msg_find_element(msg, name);
 506         if (el == NULL) {
 507                 return -1;
 508         }
 509         i = el - msg->elements;
 510
 511         ret = ltdb_index_del_element(module, msg->dn, el);
 512         if (ret != LDB_SUCCESS) {
 513                 return ret;
 514         }
 515
 516         talloc_free(el->values);
 517         if (msg->num_elements > (i+1)) {
 518                 memmove(el, el+1, sizeof(*el) * (msg->num_elements - (i+1)));
 519         }
 520         msg->num_elements--;
 521         msg->elements = talloc_realloc(msg, msg->elements,
 522                                        struct ldb_message_element,
 523                                        msg->num_elements);
 524         return 0;
 525 }
 526
 527 /*
 528   delete all elements matching an attribute name/value
 529
 530   return 0 on success, -1 on failure
 531 */
 532 static int msg_delete_element(struct ldb_module *module,
 533                               struct ldb_message *msg,
 534                               const char *name,
 535                               const struct ldb_val *val)
 536 {
 537         struct ldb_context *ldb = ldb_module_get_ctx(module);
 538         unsigned int i;
 539         int found, ret;
 540         struct ldb_message_element *el;
 541         const struct ldb_schema_attribute *a;
 542
 543         found = find_element(msg, name);
 544         if (found == -1) {
 545                 return -1;
 546         }
 547
 548         el = &msg->elements[found];
 549
 550         a = ldb_schema_attribute_by_name(ldb, el->name);
 551
 552         for (i=0;i<el->num_values;i++) {
 553                 if (a->syntax->comparison_fn(ldb, ldb,
 554                                              &el->values[i], val) == 0) {
 555                         if (el->num_values == 1) {
 556                                 return msg_delete_attribute(module, ldb, msg, name);
 557                         }
 558
 559                         ret = ltdb_index_del_value(module, msg->dn, el, i);
 560                         if (ret != LDB_SUCCESS) {
 561                                 return -1;
 562                         }
 563
 564                         if (i<el->num_values-1) {
 565                                 memmove(&el->values[i], &el->values[i+1],
 566                                         sizeof(el->values[i])*
 567                                                 (el->num_values-(i+1)));
 568                         }
 569                         el->num_values--;
 570
 571                         /* per definition we find in a canonicalised message an
 572                            attribute value only once. So we are finished here */
 573                         return 0;
 574                 }
 575         }
 576
 577         /* Not found */
 578         return -1;
 579 }
 580
 581
 582 /*
 583   modify a record - internal interface
 584
 585   yuck - this is O(n^2). Luckily n is usually small so we probably
 586   get away with it, but if we ever have really large attribute lists
 587   then we'll need to look at this again
 588 */
 589 int ltdb_modify_internal(struct ldb_module *module,
 590                          const struct ldb_message *msg)
 591 {
 592         struct ldb_context *ldb = ldb_module_get_ctx(module);
 593         void *data = ldb_module_get_private(module);
 594         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 595         TDB_DATA tdb_key, tdb_data;
 596         struct ldb_message *msg2;
 597         unsigned i, j;
 598         int ret = LDB_SUCCESS, idx;
 599
 600         tdb_key = ltdb_key(module, msg->dn);
 601         if (!tdb_key.dptr) {
 602                 return LDB_ERR_OTHER;
 603         }
 604
 605         tdb_data = tdb_fetch(ltdb->tdb, tdb_key);
 606         if (!tdb_data.dptr) {
 607                 talloc_free(tdb_key.dptr);
 608                 return ltdb_err_map(tdb_error(ltdb->tdb));
 609         }
 610
 611         msg2 = talloc(tdb_key.dptr, struct ldb_message);
 612         if (msg2 == NULL) {
 613                 free(tdb_data.dptr);
 614                 ret = LDB_ERR_OTHER;
 615                 goto done;
 616         }
 617
 618         ret = ltdb_unpack_data(module, &tdb_data, msg2);
 619         free(tdb_data.dptr);
 620         if (ret == -1) {
 621                 ret = LDB_ERR_OTHER;
 622                 goto done;
 623         }
 624
 625         if (!msg2->dn) {
 626                 msg2->dn = msg->dn;
 627         }
 628
 629         for (i=0; i<msg->num_elements; i++) {
 630                 struct ldb_message_element *el = &msg->elements[i], *el2;
 631                 struct ldb_val *vals;
 632                 const struct ldb_schema_attribute *a = ldb_schema_attribute_by_name(ldb, el->name);
 633                 const char *dn;
 634
 635                 if (ldb_attr_cmp(el->name, "distinguishedName") == 0) {
 636                         ldb_asprintf_errstring(ldb, "it is not permitted to perform a modify on 'distinguishedName' (use rename instead): %s",
 637                                                ldb_dn_get_linearized(msg2->dn));
 638                         ret = LDB_ERR_CONSTRAINT_VIOLATION;
 639                         goto done;
 640                 }
 641
 642                 switch (msg->elements[i].flags & LDB_FLAG_MOD_MASK) {
 643                 case LDB_FLAG_MOD_ADD:
 644                         if (el->num_values == 0) {
 645                                 ldb_asprintf_errstring(ldb, "attribute %s on %s specified, but with 0 values (illigal)",
 646                                                        el->name, ldb_dn_get_linearized(msg2->dn));
 647                                 ret = LDB_ERR_CONSTRAINT_VIOLATION;
 648                                 goto done;
 649                         }
 650
 651                         if (a && a->flags & LDB_ATTR_FLAG_SINGLE_VALUE) {
 652                                 if (el->num_values > 1) {
 653                                         ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
 654                                                                el->name, ldb_dn_get_linearized(msg2->dn));
 655                                         ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
 656                                         goto done;
 657                                 }
 658                         }
 659
 660                         /* Checks if element already exists */
 661                         idx = find_element(msg2, el->name);
 662                         if (idx == -1) {
 663                                 if (ltdb_msg_add_element(ldb, msg2, el) != 0) {
 664                                         ret = LDB_ERR_OTHER;
 665                                         goto done;
 666                                 }
 667                                 ret = ltdb_index_add_element(module, msg2->dn, el);
 668                                 if (ret != LDB_SUCCESS) {
 669                                         goto done;
 670                                 }
 671                         } else {
 672                                 /* We cannot add another value on a existing one
 673                                    if the attribute is single-valued */
 674                                 if (a && a->flags & LDB_ATTR_FLAG_SINGLE_VALUE) {
 675                                         ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
 676                                                                el->name, ldb_dn_get_linearized(msg2->dn));
 677                                         ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
 678                                         goto done;
 679                                 }
 680
 681                                 el2 = &(msg2->elements[idx]);
 682
 683                                 /* Check that values don't exist yet on multi-
 684                                    valued attributes or aren't provided twice */
 685                                 for (j=0; j<el->num_values; j++) {
 686                                         if (ldb_msg_find_val(el2, &el->values[j]) != NULL) {
 687                                                 ldb_asprintf_errstring(ldb, "%s: value #%d already exists", el->name, j);
 688                                                 ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
 689                                                 goto done;
 690                                         }
 691                                         if (ldb_msg_find_val(el, &el->values[j]) != &el->values[j]) {
 692                                                 ldb_asprintf_errstring(ldb, "%s: value #%d provided more than once", el->name, j);
 693                                                 ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
 694                                                 goto done;
 695                                         }
 696                                 }
 697
 698                                 /* Now combine existing and new values to a new
 699                                    attribute record */
 700                                 vals = talloc_realloc(msg2->elements,
 701                                                       el2->values, struct ldb_val,
 702                                                       el2->num_values + el->num_values);
 703                                 if (vals == NULL) {
 704                                         ldb_oom(ldb);
 705                                         ret = LDB_ERR_OTHER;
 706                                         goto done;
 707                                 }
 708
 709                                 for (j=0; j<el->num_values; j++) {
 710                                         vals[el2->num_values + j] =
 711                                                 ldb_val_dup(vals, &el->values[j]);
 712                                 }
 713
 714                                 el2->values = vals;
 715                                 el2->num_values += el->num_values;
 716
 717                                 ret = ltdb_index_add_element(module, msg2->dn, el);
 718                                 if (ret != LDB_SUCCESS) {
 719                                         goto done;
 720                                 }
 721                         }
 722
 723                         break;
 724
 725                 case LDB_FLAG_MOD_REPLACE:
 726                         if (a && a->flags & LDB_ATTR_FLAG_SINGLE_VALUE) {
 727                                 if (el->num_values > 1) {
 728                                         ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
 729                                                                el->name, ldb_dn_get_linearized(msg2->dn));
 730                                         ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
 731                                         goto done;
 732                                 }
 733                         }
 734
 735                         /* TODO: This is O(n^2) - replace with more efficient check */
 736                         for (j=0; j<el->num_values; j++) {
 737                                 if (ldb_msg_find_val(el, &el->values[j]) != &el->values[j]) {
 738                                         ldb_asprintf_errstring(ldb, "%s: value #%d provided more than once", el->name, j);
 739                                         ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
 740                                         goto done;
 741                                 }
 742                         }
 743
 744                         idx = find_element(msg2, el->name);
 745                         if (idx != -1) {
 746                                 el2 = &(msg2->elements[idx]);
 747                                 if (ldb_msg_element_compare(el, el2) == 0) {
 748                                         /* we are replacing with the same values */
 749                                         continue;
 750                                 }
 751
 752                                 /* Delete the attribute if it exists in the DB */
 753                                 if (msg_delete_attribute(module, ldb, msg2, el->name) != 0) {
 754                                         ret = LDB_ERR_OTHER;
 755                                         goto done;
 756                                 }
 757                         }
 758
 759                         /* Recreate it with the new values */
 760                         if (ltdb_msg_add_element(ldb, msg2, el) != 0) {
 761                                 ret = LDB_ERR_OTHER;
 762                                 goto done;
 763                         }
 764
 765                         ret = ltdb_index_add_element(module, msg2->dn, el);
 766                         if (ret != LDB_SUCCESS) {
 767                                 goto done;
 768                         }
 769
 770                         break;
 771
 772                 case LDB_FLAG_MOD_DELETE:
 773                         dn = ldb_dn_get_linearized(msg2->dn);
 774                         if (dn == NULL) {
 775                                 ret = LDB_ERR_OTHER;
 776                                 goto done;
 777                         }
 778
 779                         if (msg->elements[i].num_values == 0) {
 780                                 /* Delete the whole attribute */
 781                                 if (msg_delete_attribute(module, ldb, msg2,
 782                                                          msg->elements[i].name) != 0) {
 783                                         ldb_asprintf_errstring(ldb, "No such attribute: %s for delete on %s",
 784                                                                msg->elements[i].name, dn);
 785                                         ret = LDB_ERR_NO_SUCH_ATTRIBUTE;
 786                                         goto done;
 787                                 }
 788                         } else {
 789                                 /* Delete specified values from an attribute */
 790                                 for (j=0; j < msg->elements[i].num_values; j++) {
 791                                         if (msg_delete_element(module,
 792                                                                msg2,
 793                                                                msg->elements[i].name,
 794                                                                &msg->elements[i].values[j]) != 0) {
 795                                                 ldb_asprintf_errstring(ldb, "No matching attribute value when deleting attribute: %s on %s",
 796                                                                        msg->elements[i].name, dn);
 797                                                 ret = LDB_ERR_NO_SUCH_ATTRIBUTE;
 798                                                 goto done;
 799                                         }
 800                                 }
 801                         }
 802                         break;
 803                 default:
 804                         ldb_asprintf_errstring(ldb,
 805                                 "Invalid ldb_modify flags on %s: 0x%x",
 806                                 msg->elements[i].name,
 807                                 msg->elements[i].flags & LDB_FLAG_MOD_MASK);
 808                         ret = LDB_ERR_PROTOCOL_ERROR;
 809                         goto done;
 810                 }
 811         }
 812
 813         ret = ltdb_store(module, msg2, TDB_MODIFY);
 814         if (ret != LDB_SUCCESS) {
 815                 goto done;
 816         }
 817
 818         ret = ltdb_modified(module, msg2->dn);
 819         if (ret != LDB_SUCCESS) {
 820                 goto done;
 821         }
 822
 823 done:
 824         talloc_free(tdb_key.dptr);
 825         return ret;
 826 }
 827
 828 /*
 829   modify a record
 830 */
 831 static int ltdb_modify(struct ltdb_context *ctx)
 832 {
 833         struct ldb_module *module = ctx->module;
 834         struct ldb_request *req = ctx->req;
 835         int ret = LDB_SUCCESS;
 836
 837         ret = ltdb_check_special_dn(module, req->op.mod.message);
 838         if (ret != LDB_SUCCESS) {
 839                 return ret;
 840         }
 841
 842         ldb_request_set_state(req, LDB_ASYNC_PENDING);
 843
 844         if (ltdb_cache_load(module) != 0) {
 845                 return LDB_ERR_OPERATIONS_ERROR;
 846         }
 847
 848         ret = ltdb_modify_internal(module, req->op.mod.message);
 849
 850         return ret;
 851 }
 852
 853 /*
 854   rename a record
 855 */
 856 static int ltdb_rename(struct ltdb_context *ctx)
 857 {
 858         struct ldb_module *module = ctx->module;
 859         struct ldb_request *req = ctx->req;
 860         struct ldb_message *msg;
 861         int ret = LDB_SUCCESS;
 862
 863         ldb_request_set_state(req, LDB_ASYNC_PENDING);
 864
 865         if (ltdb_cache_load(ctx->module) != 0) {
 866                 return LDB_ERR_OPERATIONS_ERROR;
 867         }
 868
 869         msg = talloc(ctx, struct ldb_message);
 870         if (msg == NULL) {
 871                 return LDB_ERR_OPERATIONS_ERROR;
 872         }
 873
 874         /* in case any attribute of the message was indexed, we need
 875            to fetch the old record */
 876         ret = ltdb_search_dn1(module, req->op.rename.olddn, msg);
 877         if (ret != LDB_SUCCESS) {
 878                 /* not finding the old record is an error */
 879                 return ret;
 880         }
 881
 882         /* Always delete first then add, to avoid conflicts with
 883          * unique indexes. We rely on the transaction to make this
 884          * atomic
 885          */
 886         ret = ltdb_delete_internal(module, msg->dn);
 887         if (ret != LDB_SUCCESS) {
 888                 return ret;
 889         }
 890
 891         msg->dn = ldb_dn_copy(msg, req->op.rename.newdn);
 892         if (msg->dn == NULL) {
 893                 return LDB_ERR_OPERATIONS_ERROR;
 894         }
 895
 896         ret = ltdb_add_internal(module, msg);
 897
 898         return ret;
 899 }
 900
 901 static int ltdb_start_trans(struct ldb_module *module)
 902 {
 903         void *data = ldb_module_get_private(module);
 904         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 905
 906         if (tdb_transaction_start(ltdb->tdb) != 0) {
 907                 return ltdb_err_map(tdb_error(ltdb->tdb));
 908         }
 909
 910         ltdb->in_transaction++;
 911
 912         ltdb_index_transaction_start(module);
 913
 914         return LDB_SUCCESS;
 915 }
 916
 917 static int ltdb_prepare_commit(struct ldb_module *module)
 918 {
 919         void *data = ldb_module_get_private(module);
 920         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 921
 922         if (ltdb->in_transaction != 1) {
 923                 return LDB_SUCCESS;
 924         }
 925
 926         if (ltdb_index_transaction_commit(module) != 0) {
 927                 tdb_transaction_cancel(ltdb->tdb);
 928                 ltdb->in_transaction--;
 929                 return ltdb_err_map(tdb_error(ltdb->tdb));
 930         }
 931
 932         if (tdb_transaction_prepare_commit(ltdb->tdb) != 0) {
 933                 ltdb->in_transaction--;
 934                 return ltdb_err_map(tdb_error(ltdb->tdb));
 935         }
 936
 937         ltdb->prepared_commit = true;
 938
 939         return LDB_SUCCESS;
 940 }
 941
 942 static int ltdb_end_trans(struct ldb_module *module)
 943 {
 944         void *data = ldb_module_get_private(module);
 945         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 946
 947         if (!ltdb->prepared_commit) {
 948                 int ret = ltdb_prepare_commit(module);
 949                 if (ret != LDB_SUCCESS) {
 950                         return ret;
 951                 }
 952         }
 953
 954         ltdb->in_transaction--;
 955         ltdb->prepared_commit = false;
 956
 957         if (tdb_transaction_commit(ltdb->tdb) != 0) {
 958                 return ltdb_err_map(tdb_error(ltdb->tdb));
 959         }
 960
 961         return LDB_SUCCESS;
 962 }
 963
 964 static int ltdb_del_trans(struct ldb_module *module)
 965 {
 966         void *data = ldb_module_get_private(module);
 967         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 968
 969         ltdb->in_transaction--;
 970
 971         if (ltdb_index_transaction_cancel(module) != 0) {
 972                 tdb_transaction_cancel(ltdb->tdb);
 973                 return ltdb_err_map(tdb_error(ltdb->tdb));
 974         }
 975
 976         if (tdb_transaction_cancel(ltdb->tdb) != 0) {
 977                 return ltdb_err_map(tdb_error(ltdb->tdb));
 978         }
 979
 980         return LDB_SUCCESS;
 981 }
 982
 983 /*
 984   return sequenceNumber from @BASEINFO
 985 */
 986 static int ltdb_sequence_number(struct ltdb_context *ctx,
 987                                 struct ldb_extended **ext)
 988 {
 989         struct ldb_context *ldb;
 990         struct ldb_module *module = ctx->module;
 991         struct ldb_request *req = ctx->req;
 992         TALLOC_CTX *tmp_ctx;
 993         struct ldb_seqnum_request *seq;
 994         struct ldb_seqnum_result *res;
 995         struct ldb_message *msg = NULL;
 996         struct ldb_dn *dn;
 997         const char *date;
 998         int ret = LDB_SUCCESS;
 999
1000         ldb = ldb_module_get_ctx(module);
1001
1002         seq = talloc_get_type(req->op.extended.data,
1003                                 struct ldb_seqnum_request);
1004         if (seq == NULL) {
1005                 return LDB_ERR_OPERATIONS_ERROR;
1006         }
1007
1008         ldb_request_set_state(req, LDB_ASYNC_PENDING);
1009
1010         if (ltdb_lock_read(module) != 0) {
1011                 return LDB_ERR_OPERATIONS_ERROR;
1012         }
1013
1014         res = talloc_zero(req, struct ldb_seqnum_result);
1015         if (res == NULL) {
1016                 ret = LDB_ERR_OPERATIONS_ERROR;
1017                 goto done;
1018         }
1019         tmp_ctx = talloc_new(req);
1020         if (tmp_ctx == NULL) {
1021                 ret = LDB_ERR_OPERATIONS_ERROR;
1022                 goto done;
1023         }
1024
1025         dn = ldb_dn_new(tmp_ctx, ldb, LTDB_BASEINFO);
1026
1027         msg = talloc(tmp_ctx, struct ldb_message);
1028         if (msg == NULL) {
1029                 ret = LDB_ERR_OPERATIONS_ERROR;
1030                 goto done;
1031         }
1032
1033         ret = ltdb_search_dn1(module, dn, msg);
1034         if (ret != LDB_SUCCESS) {
1035                 goto done;
1036         }
1037
1038         switch (seq->type) {
1039         case LDB_SEQ_HIGHEST_SEQ:
1040                 res->seq_num = ldb_msg_find_attr_as_uint64(msg, LTDB_SEQUENCE_NUMBER, 0);
1041                 break;
1042         case LDB_SEQ_NEXT:
1043                 res->seq_num = ldb_msg_find_attr_as_uint64(msg, LTDB_SEQUENCE_NUMBER, 0);
1044                 res->seq_num++;
1045                 break;
1046         case LDB_SEQ_HIGHEST_TIMESTAMP:
1047                 date = ldb_msg_find_attr_as_string(msg, LTDB_MOD_TIMESTAMP, NULL);
1048                 if (date) {
1049                         res->seq_num = ldb_string_to_time(date);
1050                 } else {
1051                         res->seq_num = 0;
1052                         /* zero is as good as anything when we don't know */
1053                 }
1054                 break;
1055         }
1056
1057         *ext = talloc_zero(req, struct ldb_extended);
1058         if (*ext == NULL) {
1059                 ret = LDB_ERR_OPERATIONS_ERROR;
1060                 goto done;
1061         }
1062         (*ext)->oid = LDB_EXTENDED_SEQUENCE_NUMBER;
1063         (*ext)->data = talloc_steal(*ext, res);
1064
1065 done:
1066         talloc_free(tmp_ctx);
1067         ltdb_unlock_read(module);
1068         return ret;
1069 }
1070
1071 static void ltdb_request_done(struct ltdb_context *ctx, int error)
1072 {
1073         struct ldb_context *ldb;
1074         struct ldb_request *req;
1075         struct ldb_reply *ares;
1076
1077         ldb = ldb_module_get_ctx(ctx->module);
1078         req = ctx->req;
1079
1080         /* if we already returned an error just return */
1081         if (ldb_request_get_status(req) != LDB_SUCCESS) {
1082                 return;
1083         }
1084
1085         ares = talloc_zero(req, struct ldb_reply);
1086         if (!ares) {
1087                 ldb_oom(ldb);
1088                 req->callback(req, NULL);
1089                 return;
1090         }
1091         ares->type = LDB_REPLY_DONE;
1092         ares->error = error;
1093
1094         req->callback(req, ares);
1095 }
1096
1097 static void ltdb_timeout(struct tevent_context *ev,
1098                           struct tevent_timer *te,
1099                           struct timeval t,
1100                           void *private_data)
1101 {
1102         struct ltdb_context *ctx;
1103         ctx = talloc_get_type(private_data, struct ltdb_context);
1104
1105         if (!ctx->request_terminated) {
1106                 /* request is done now */
1107                 ltdb_request_done(ctx, LDB_ERR_TIME_LIMIT_EXCEEDED);
1108         }
1109
1110         if (!ctx->request_terminated) {
1111                 /* neutralize the spy */
1112                 ctx->spy->ctx = NULL;
1113         }
1114         talloc_free(ctx);
1115 }
1116
1117 static void ltdb_request_extended_done(struct ltdb_context *ctx,
1118                                         struct ldb_extended *ext,
1119                                         int error)
1120 {
1121         struct ldb_context *ldb;
1122         struct ldb_request *req;
1123         struct ldb_reply *ares;
1124
1125         ldb = ldb_module_get_ctx(ctx->module);
1126         req = ctx->req;
1127
1128         /* if we already returned an error just return */
1129         if (ldb_request_get_status(req) != LDB_SUCCESS) {
1130                 return;
1131         }
1132
1133         ares = talloc_zero(req, struct ldb_reply);
1134         if (!ares) {
1135                 ldb_oom(ldb);
1136                 req->callback(req, NULL);
1137                 return;
1138         }
1139         ares->type = LDB_REPLY_DONE;
1140         ares->response = ext;
1141         ares->error = error;
1142
1143         req->callback(req, ares);
1144 }
1145
1146 static void ltdb_handle_extended(struct ltdb_context *ctx)
1147 {
1148         struct ldb_extended *ext = NULL;
1149         int ret;
1150
1151         if (strcmp(ctx->req->op.extended.oid,
1152                    LDB_EXTENDED_SEQUENCE_NUMBER) == 0) {
1153                 /* get sequence number */
1154                 ret = ltdb_sequence_number(ctx, &ext);
1155         } else {
1156                 /* not recognized */
1157                 ret = LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
1158         }
1159
1160         ltdb_request_extended_done(ctx, ext, ret);
1161 }
1162
1163 static void ltdb_callback(struct tevent_context *ev,
1164                           struct tevent_timer *te,
1165                           struct timeval t,
1166                           void *private_data)
1167 {
1168         struct ltdb_context *ctx;
1169         int ret;
1170
1171         ctx = talloc_get_type(private_data, struct ltdb_context);
1172
1173         if (ctx->request_terminated) {
1174                 goto done;
1175         }
1176
1177         switch (ctx->req->operation) {
1178         case LDB_SEARCH:
1179                 ret = ltdb_search(ctx);
1180                 break;
1181         case LDB_ADD:
1182                 ret = ltdb_add(ctx);
1183                 break;
1184         case LDB_MODIFY:
1185                 ret = ltdb_modify(ctx);
1186                 break;
1187         case LDB_DELETE:
1188                 ret = ltdb_delete(ctx);
1189                 break;
1190         case LDB_RENAME:
1191                 ret = ltdb_rename(ctx);
1192                 break;
1193         case LDB_EXTENDED:
1194                 ltdb_handle_extended(ctx);
1195                 goto done;
1196         default:
1197                 /* no other op supported */
1198                 ret = LDB_ERR_UNWILLING_TO_PERFORM;
1199         }
1200
1201         if (!ctx->request_terminated) {
1202                 /* request is done now */
1203                 ltdb_request_done(ctx, ret);
1204         }
1205
1206 done:
1207         if (!ctx->request_terminated) {
1208                 /* neutralize the spy */
1209                 ctx->spy->ctx = NULL;
1210         }
1211         talloc_free(ctx);
1212 }
1213
1214 static int ltdb_request_destructor(void *ptr)
1215 {
1216         struct ltdb_req_spy *spy = talloc_get_type(ptr, struct ltdb_req_spy);
1217
1218         if (spy->ctx != NULL) {
1219                 spy->ctx->request_terminated = true;
1220         }
1221
1222         return 0;
1223 }
1224
1225 static int ltdb_handle_request(struct ldb_module *module,
1226                                struct ldb_request *req)
1227 {
1228         struct ldb_context *ldb;
1229         struct tevent_context *ev;
1230         struct ltdb_context *ac;
1231         struct tevent_timer *te;
1232         struct timeval tv;
1233
1234         if (check_critical_controls(req->controls)) {
1235                 return LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
1236         }
1237
1238         ldb = ldb_module_get_ctx(module);
1239
1240         if (req->starttime == 0 || req->timeout == 0) {
1241                 ldb_set_errstring(ldb, "Invalid timeout settings");
1242                 return LDB_ERR_TIME_LIMIT_EXCEEDED;
1243         }
1244
1245         ev = ldb_get_event_context(ldb);
1246
1247         ac = talloc_zero(ldb, struct ltdb_context);
1248         if (ac == NULL) {
1249                 ldb_oom(ldb);
1250                 return LDB_ERR_OPERATIONS_ERROR;
1251         }
1252
1253         ac->module = module;
1254         ac->req = req;
1255
1256         tv.tv_sec = 0;
1257         tv.tv_usec = 0;
1258         te = tevent_add_timer(ev, ac, tv, ltdb_callback, ac);
1259         if (NULL == te) {
1260                 talloc_free(ac);
1261                 return LDB_ERR_OPERATIONS_ERROR;
1262         }
1263
1264         tv.tv_sec = req->starttime + req->timeout;
1265         ac->timeout_event = tevent_add_timer(ev, ac, tv, ltdb_timeout, ac);
1266         if (NULL == ac->timeout_event) {
1267                 talloc_free(ac);
1268                 return LDB_ERR_OPERATIONS_ERROR;
1269         }
1270
1271         /* set a spy so that we do not try to use the request context
1272          * if it is freed before ltdb_callback fires */
1273         ac->spy = talloc(req, struct ltdb_req_spy);
1274         if (NULL == ac->spy) {
1275                 talloc_free(ac);
1276                 return LDB_ERR_OPERATIONS_ERROR;
1277         }
1278         ac->spy->ctx = ac;
1279
1280         talloc_set_destructor((TALLOC_CTX *)ac->spy, ltdb_request_destructor);
1281
1282         return LDB_SUCCESS;
1283 }
1284
1285 static const struct ldb_module_ops ltdb_ops = {
1286         .name              = "tdb",
1287         .search            = ltdb_handle_request,
1288         .add               = ltdb_handle_request,
1289         .modify            = ltdb_handle_request,
1290         .del               = ltdb_handle_request,
1291         .rename            = ltdb_handle_request,
1292         .extended          = ltdb_handle_request,
1293         .start_transaction = ltdb_start_trans,
1294         .end_transaction   = ltdb_end_trans,
1295         .prepare_commit    = ltdb_prepare_commit,
1296         .del_transaction   = ltdb_del_trans,
1297 };
1298
1299 /*
1300   connect to the database
1301 */
1302 static int ltdb_connect(struct ldb_context *ldb, const char *url,
1303                         unsigned int flags, const char *options[],
1304                         struct ldb_module **_module)
1305 {
1306         struct ldb_module *module;
1307         const char *path;
1308         int tdb_flags, open_flags;
1309         struct ltdb_private *ltdb;
1310
1311         /* parse the url */
1312         if (strchr(url, ':')) {
1313                 if (strncmp(url, "tdb://", 6) != 0) {
1314                         ldb_debug(ldb, LDB_DEBUG_ERROR,
1315                                   "Invalid tdb URL '%s'", url);
1316                         return LDB_ERR_OPERATIONS_ERROR;
1317                 }
1318                 path = url+6;
1319         } else {
1320                 path = url;
1321         }
1322
1323         tdb_flags = TDB_DEFAULT | TDB_SEQNUM;
1324
1325         /* check for the 'nosync' option */
1326         if (flags & LDB_FLG_NOSYNC) {
1327                 tdb_flags |= TDB_NOSYNC;
1328         }
1329
1330         /* and nommap option */
1331         if (flags & LDB_FLG_NOMMAP) {
1332                 tdb_flags |= TDB_NOMMAP;
1333         }
1334
1335         if (flags & LDB_FLG_RDONLY) {
1336                 open_flags = O_RDONLY;
1337         } else {
1338                 open_flags = O_CREAT | O_RDWR;
1339         }
1340
1341         ltdb = talloc_zero(ldb, struct ltdb_private);
1342         if (!ltdb) {
1343                 ldb_oom(ldb);
1344                 return LDB_ERR_OPERATIONS_ERROR;
1345         }
1346
1347         /* note that we use quite a large default hash size */
1348         ltdb->tdb = ltdb_wrap_open(ltdb, path, 10000,
1349                                    tdb_flags, open_flags,
1350                                    ldb_get_create_perms(ldb), ldb);
1351         if (!ltdb->tdb) {
1352                 ldb_debug(ldb, LDB_DEBUG_ERROR,
1353                           "Unable to open tdb '%s'", path);
1354                 talloc_free(ltdb);
1355                 return LDB_ERR_OPERATIONS_ERROR;
1356         }
1357
1358         ltdb->sequence_number = 0;
1359
1360         module = ldb_module_new(ldb, ldb, "ldb_tdb backend", &ltdb_ops);
1361         if (!module) {
1362                 talloc_free(ltdb);
1363                 return LDB_ERR_OPERATIONS_ERROR;
1364         }
1365         ldb_module_set_private(module, ltdb);
1366         talloc_steal(module, ltdb);
1367
1368         if (ltdb_cache_load(module) != 0) {
1369                 talloc_free(module);
1370                 talloc_free(ltdb);
1371                 return LDB_ERR_OPERATIONS_ERROR;
1372         }
1373
1374         *_module = module;
1375         return LDB_SUCCESS;
1376 }
1377
1378 const struct ldb_backend_ops ldb_tdb_backend_ops = {
1379         .name = "tdb",
1380         .connect_fn = ltdb_connect
1381 };