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