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
   7
   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
  44 #include "ldb_includes.h"
  45
  46 #include "ldb_tdb.h"
  47
  48
  49 /*
  50   map a tdb error code to a ldb error code
  51 */
  52 static int ltdb_err_map(enum TDB_ERROR tdb_code)
  53 {
  54         switch (tdb_code) {
  55         case TDB_SUCCESS:
  56                 return LDB_SUCCESS;
  57         case TDB_ERR_CORRUPT:
  58         case TDB_ERR_OOM:
  59         case TDB_ERR_EINVAL:
  60                 return LDB_ERR_OPERATIONS_ERROR;
  61         case TDB_ERR_IO:
  62                 return LDB_ERR_PROTOCOL_ERROR;
  63         case TDB_ERR_LOCK:
  64         case TDB_ERR_NOLOCK:
  65                 return LDB_ERR_BUSY;
  66         case TDB_ERR_LOCK_TIMEOUT:
  67                 return LDB_ERR_TIME_LIMIT_EXCEEDED;
  68         case TDB_ERR_EXISTS:
  69                 return LDB_ERR_ENTRY_ALREADY_EXISTS;
  70         case TDB_ERR_NOEXIST:
  71                 return LDB_ERR_NO_SUCH_OBJECT;
  72         case TDB_ERR_RDONLY:
  73                 return LDB_ERR_INSUFFICIENT_ACCESS_RIGHTS;
  74         }
  75         return LDB_ERR_OTHER;
  76 }
  77
  78
  79 struct ldb_handle *init_ltdb_handle(struct ltdb_private *ltdb,
  80                                     struct ldb_module *module,
  81                                     struct ldb_request *req)
  82 {
  83         struct ltdb_context *ac;
  84         struct ldb_handle *h;
  85
  86         h = talloc_zero(req, struct ldb_handle);
  87         if (h == NULL) {
  88                 ldb_set_errstring(module->ldb, "Out of Memory");
  89                 return NULL;
  90         }
  91
  92         h->module = module;
  93
  94         ac = talloc_zero(h, struct ltdb_context);
  95         if (ac == NULL) {
  96                 ldb_set_errstring(module->ldb, "Out of Memory");
  97                 talloc_free(h);
  98                 return NULL;
  99         }
 100
 101         h->private_data = (void *)ac;
 102
 103         h->state = LDB_ASYNC_INIT;
 104         h->status = LDB_SUCCESS;
 105
 106         ac->module = module;
 107         ac->context = req->context;
 108         ac->callback = req->callback;
 109
 110         return h;
 111 }
 112
 113 /*
 114   form a TDB_DATA for a record key
 115   caller frees
 116
 117   note that the key for a record can depend on whether the
 118   dn refers to a case sensitive index record or not
 119 */
 120 struct TDB_DATA ltdb_key(struct ldb_module *module, struct ldb_dn *dn)
 121 {
 122         struct ldb_context *ldb = module->ldb;
 123         TDB_DATA key;
 124         char *key_str = NULL;
 125         const char *dn_folded = NULL;
 126
 127         /*
 128           most DNs are case insensitive. The exception is index DNs for
 129           case sensitive attributes
 130
 131           there are 3 cases dealt with in this code:
 132
 133           1) if the dn doesn't start with @ then uppercase the attribute
 134              names and the attributes values of case insensitive attributes
 135           2) if the dn starts with @ then leave it alone -
 136              the indexing code handles the rest
 137         */
 138
 139         dn_folded = ldb_dn_get_casefold(dn);
 140         if (!dn_folded) {
 141                 goto failed;
 142         }
 143
 144         key_str = talloc_strdup(ldb, "DN=");
 145         if (!key_str) {
 146                 goto failed;
 147         }
 148
 149         key_str = talloc_strdup_append_buffer(key_str, dn_folded);
 150         if (!key_str) {
 151                 goto failed;
 152         }
 153
 154         key.dptr = (uint8_t *)key_str;
 155         key.dsize = strlen(key_str) + 1;
 156
 157         return key;
 158
 159 failed:
 160         errno = ENOMEM;
 161         key.dptr = NULL;
 162         key.dsize = 0;
 163         return key;
 164 }
 165
 166 /*
 167   check special dn's have valid attributes
 168   currently only @ATTRIBUTES is checked
 169 */
 170 int ltdb_check_special_dn(struct ldb_module *module,
 171                           const struct ldb_message *msg)
 172 {
 173         int i, j;
 174
 175         if (! ldb_dn_is_special(msg->dn) ||
 176             ! ldb_dn_check_special(msg->dn, LTDB_ATTRIBUTES)) {
 177                 return 0;
 178         }
 179
 180         /* we have @ATTRIBUTES, let's check attributes are fine */
 181         /* should we check that we deny multivalued attributes ? */
 182         for (i = 0; i < msg->num_elements; i++) {
 183                 for (j = 0; j < msg->elements[i].num_values; j++) {
 184                         if (ltdb_check_at_attributes_values(&msg->elements[i].values[j]) != 0) {
 185                                 ldb_set_errstring(module->ldb, "Invalid attribute value in an @ATTRIBUTES entry");
 186                                 return LDB_ERR_INVALID_ATTRIBUTE_SYNTAX;
 187                         }
 188                 }
 189         }
 190
 191         return 0;
 192 }
 193
 194
 195 /*
 196   we've made a modification to a dn - possibly reindex and
 197   update sequence number
 198 */
 199 static int ltdb_modified(struct ldb_module *module, struct ldb_dn *dn)
 200 {
 201         int ret = LDB_SUCCESS;
 202
 203         if (ldb_dn_is_special(dn) &&
 204             (ldb_dn_check_special(dn, LTDB_INDEXLIST) ||
 205              ldb_dn_check_special(dn, LTDB_ATTRIBUTES)) ) {
 206                 ret = ltdb_reindex(module);
 207         }
 208
 209         if (ret == LDB_SUCCESS &&
 210             !(ldb_dn_is_special(dn) &&
 211               ldb_dn_check_special(dn, LTDB_BASEINFO)) ) {
 212                 ret = ltdb_increase_sequence_number(module);
 213         }
 214
 215         return ret;
 216 }
 217
 218 /*
 219   store a record into the db
 220 */
 221 int ltdb_store(struct ldb_module *module, const struct ldb_message *msg, int flgs)
 222 {
 223         struct ltdb_private *ltdb =
 224                 talloc_get_type(module->private_data, struct ltdb_private);
 225         TDB_DATA tdb_key, tdb_data;
 226         int ret;
 227
 228         tdb_key = ltdb_key(module, msg->dn);
 229         if (!tdb_key.dptr) {
 230                 return LDB_ERR_OTHER;
 231         }
 232
 233         ret = ltdb_pack_data(module, msg, &tdb_data);
 234         if (ret == -1) {
 235                 talloc_free(tdb_key.dptr);
 236                 return LDB_ERR_OTHER;
 237         }
 238
 239         ret = tdb_store(ltdb->tdb, tdb_key, tdb_data, flgs);
 240         if (ret == -1) {
 241                 ret = ltdb_err_map(tdb_error(ltdb->tdb));
 242                 goto done;
 243         }
 244
 245         ret = ltdb_index_add(module, msg);
 246         if (ret != LDB_SUCCESS) {
 247                 tdb_delete(ltdb->tdb, tdb_key);
 248         }
 249
 250 done:
 251         talloc_free(tdb_key.dptr);
 252         talloc_free(tdb_data.dptr);
 253
 254         return ret;
 255 }
 256
 257
 258 static int ltdb_add_internal(struct ldb_module *module,
 259                              const struct ldb_message *msg)
 260 {
 261         int ret;
 262
 263         ret = ltdb_check_special_dn(module, msg);
 264         if (ret != LDB_SUCCESS) {
 265                 return ret;
 266         }
 267
 268         if (ltdb_cache_load(module) != 0) {
 269                 return LDB_ERR_OPERATIONS_ERROR;
 270         }
 271
 272         ret = ltdb_store(module, msg, TDB_INSERT);
 273
 274         if (ret == LDB_ERR_ENTRY_ALREADY_EXISTS) {
 275                 ldb_asprintf_errstring(module->ldb,
 276                                         "Entry %s already exists",
 277                                         ldb_dn_get_linearized(msg->dn));
 278                 return ret;
 279         }
 280
 281         if (ret == LDB_SUCCESS) {
 282                 ret = ltdb_index_one(module, msg, 1);
 283                 if (ret != LDB_SUCCESS) {
 284                         return ret;
 285                 }
 286
 287                 ret = ltdb_modified(module, msg->dn);
 288                 if (ret != LDB_SUCCESS) {
 289                         return ret;
 290                 }
 291         }
 292
 293         return ret;
 294 }
 295
 296 /*
 297   add a record to the database
 298 */
 299 static int ltdb_add(struct ldb_module *module, struct ldb_request *req)
 300 {
 301         struct ltdb_private *ltdb;
 302         struct ltdb_context *ltdb_ac;
 303         int tret, ret = LDB_SUCCESS;
 304
 305         ltdb = talloc_get_type(module->private_data, struct ltdb_private);
 306
 307         if (check_critical_controls(req->controls)) {
 308                 return LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
 309         }
 310
 311         req->handle = init_ltdb_handle(ltdb, module, req);
 312         if (req->handle == NULL) {
 313                 return LDB_ERR_OPERATIONS_ERROR;
 314         }
 315         ltdb_ac = talloc_get_type(req->handle->private_data, struct ltdb_context);
 316
 317         tret = ltdb_add_internal(module, req->op.add.message);
 318         if (tret != LDB_SUCCESS) {
 319                 req->handle->status = tret;
 320                 goto done;
 321         }
 322
 323         if (ltdb_ac->callback) {
 324                 ret = ltdb_ac->callback(module->ldb, ltdb_ac->context, NULL);
 325         }
 326 done:
 327         req->handle->state = LDB_ASYNC_DONE;
 328         return ret;
 329 }
 330
 331 /*
 332   delete a record from the database, not updating indexes (used for deleting
 333   index records)
 334 */
 335 int ltdb_delete_noindex(struct ldb_module *module, struct ldb_dn *dn)
 336 {
 337         struct ltdb_private *ltdb =
 338                 talloc_get_type(module->private_data, struct ltdb_private);
 339         TDB_DATA tdb_key;
 340         int ret;
 341
 342         tdb_key = ltdb_key(module, dn);
 343         if (!tdb_key.dptr) {
 344                 return LDB_ERR_OTHER;
 345         }
 346
 347         ret = tdb_delete(ltdb->tdb, tdb_key);
 348         talloc_free(tdb_key.dptr);
 349
 350         if (ret != 0) {
 351                 ret = ltdb_err_map(tdb_error(ltdb->tdb));
 352         }
 353
 354         return ret;
 355 }
 356
 357 static int ltdb_delete_internal(struct ldb_module *module, struct ldb_dn *dn)
 358 {
 359         struct ldb_message *msg;
 360         int ret;
 361
 362         msg = talloc(module, struct ldb_message);
 363         if (msg == NULL) {
 364                 return LDB_ERR_OPERATIONS_ERROR;
 365         }
 366
 367         /* in case any attribute of the message was indexed, we need
 368            to fetch the old record */
 369         ret = ltdb_search_dn1(module, dn, msg);
 370         if (ret != LDB_SUCCESS) {
 371                 /* not finding the old record is an error */
 372                 goto done;
 373         }
 374
 375         ret = ltdb_delete_noindex(module, dn);
 376         if (ret != LDB_SUCCESS) {
 377                 goto done;
 378         }
 379
 380         /* remove one level attribute */
 381         ret = ltdb_index_one(module, msg, 0);
 382         if (ret != LDB_SUCCESS) {
 383                 goto done;
 384         }
 385
 386         /* remove any indexed attributes */
 387         ret = ltdb_index_del(module, msg);
 388         if (ret != LDB_SUCCESS) {
 389                 goto done;
 390         }
 391
 392         ret = ltdb_modified(module, dn);
 393         if (ret != LDB_SUCCESS) {
 394                 goto done;
 395         }
 396
 397 done:
 398         talloc_free(msg);
 399         return ret;
 400 }
 401
 402 /*
 403   delete a record from the database
 404 */
 405 static int ltdb_delete(struct ldb_module *module, struct ldb_request *req)
 406 {
 407         struct ltdb_private *ltdb;
 408         struct ltdb_context *ltdb_ac;
 409         int tret, ret = LDB_SUCCESS;
 410
 411         ltdb = talloc_get_type(module->private_data, struct ltdb_private);
 412
 413         if (check_critical_controls(req->controls)) {
 414                 return LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
 415         }
 416
 417         req->handle = NULL;
 418
 419         if (ltdb_cache_load(module) != 0) {
 420                 return LDB_ERR_OPERATIONS_ERROR;
 421         }
 422
 423         req->handle = init_ltdb_handle(ltdb, module, req);
 424         if (req->handle == NULL) {
 425                 return LDB_ERR_OPERATIONS_ERROR;
 426         }
 427         ltdb_ac = talloc_get_type(req->handle->private_data, struct ltdb_context);
 428
 429         tret = ltdb_delete_internal(module, req->op.del.dn);
 430         if (tret != LDB_SUCCESS) {
 431                 req->handle->status = tret;
 432                 goto done;
 433         }
 434
 435         if (ltdb_ac->callback) {
 436                 ret = ltdb_ac->callback(module->ldb, ltdb_ac->context, NULL);
 437         }
 438 done:
 439         req->handle->state = LDB_ASYNC_DONE;
 440         return ret;
 441 }
 442
 443 /*
 444   find an element by attribute name. At the moment this does a linear search,
 445   it should be re-coded to use a binary search once all places that modify
 446   records guarantee sorted order
 447
 448   return the index of the first matching element if found, otherwise -1
 449 */
 450 static int find_element(const struct ldb_message *msg, const char *name)
 451 {
 452         unsigned int i;
 453         for (i=0;i<msg->num_elements;i++) {
 454                 if (ldb_attr_cmp(msg->elements[i].name, name) == 0) {
 455                         return i;
 456                 }
 457         }
 458         return -1;
 459 }
 460
 461
 462 /*
 463   add an element to an existing record. Assumes a elements array that we
 464   can call re-alloc on, and assumed that we can re-use the data pointers from
 465   the passed in additional values. Use with care!
 466
 467   returns 0 on success, -1 on failure (and sets errno)
 468 */
 469 static int msg_add_element(struct ldb_context *ldb,
 470                            struct ldb_message *msg,
 471                            struct ldb_message_element *el)
 472 {
 473         struct ldb_message_element *e2;
 474         unsigned int i;
 475
 476         e2 = talloc_realloc(msg, msg->elements, struct ldb_message_element,
 477                               msg->num_elements+1);
 478         if (!e2) {
 479                 errno = ENOMEM;
 480                 return -1;
 481         }
 482
 483         msg->elements = e2;
 484
 485         e2 = &msg->elements[msg->num_elements];
 486
 487         e2->name = el->name;
 488         e2->flags = el->flags;
 489         e2->values = NULL;
 490         if (el->num_values != 0) {
 491                 e2->values = talloc_array(msg->elements,
 492                                           struct ldb_val, el->num_values);
 493                 if (!e2->values) {
 494                         errno = ENOMEM;
 495                         return -1;
 496                 }
 497         }
 498         for (i=0;i<el->num_values;i++) {
 499                 e2->values[i] = el->values[i];
 500         }
 501         e2->num_values = el->num_values;
 502
 503         msg->num_elements++;
 504
 505         return 0;
 506 }
 507
 508 /*
 509   delete all elements having a specified attribute name
 510 */
 511 static int msg_delete_attribute(struct ldb_module *module,
 512                                 struct ldb_context *ldb,
 513                                 struct ldb_message *msg, const char *name)
 514 {
 515         const char *dn;
 516         unsigned int i, j;
 517
 518         dn = ldb_dn_get_linearized(msg->dn);
 519         if (dn == NULL) {
 520                 return -1;
 521         }
 522
 523         for (i=0;i<msg->num_elements;i++) {
 524                 if (ldb_attr_cmp(msg->elements[i].name, name) == 0) {
 525                         for (j=0;j<msg->elements[i].num_values;j++) {
 526                                 ltdb_index_del_value(module, dn,
 527                                                      &msg->elements[i], j);
 528                         }
 529                         talloc_free(msg->elements[i].values);
 530                         if (msg->num_elements > (i+1)) {
 531                                 memmove(&msg->elements[i],
 532                                         &msg->elements[i+1],
 533                                         sizeof(struct ldb_message_element)*
 534                                         (msg->num_elements - (i+1)));
 535                         }
 536                         msg->num_elements--;
 537                         i--;
 538                         msg->elements = talloc_realloc(msg, msg->elements,
 539                                                         struct ldb_message_element,
 540                                                         msg->num_elements);
 541                 }
 542         }
 543
 544         return 0;
 545 }
 546
 547 /*
 548   delete all elements matching an attribute name/value
 549
 550   return 0 on success, -1 on failure
 551 */
 552 static int msg_delete_element(struct ldb_module *module,
 553                               struct ldb_message *msg,
 554                               const char *name,
 555                               const struct ldb_val *val)
 556 {
 557         struct ldb_context *ldb = module->ldb;
 558         unsigned int i;
 559         int found;
 560         struct ldb_message_element *el;
 561         const struct ldb_schema_attribute *a;
 562
 563         found = find_element(msg, name);
 564         if (found == -1) {
 565                 return -1;
 566         }
 567
 568         el = &msg->elements[found];
 569
 570         a = ldb_schema_attribute_by_name(ldb, el->name);
 571
 572         for (i=0;i<el->num_values;i++) {
 573                 if (a->syntax->comparison_fn(ldb, ldb,
 574                                                 &el->values[i], val) == 0) {
 575                         if (i<el->num_values-1) {
 576                                 memmove(&el->values[i], &el->values[i+1],
 577                                         sizeof(el->values[i])*
 578                                                 (el->num_values-(i+1)));
 579                         }
 580                         el->num_values--;
 581                         if (el->num_values == 0) {
 582                                 return msg_delete_attribute(module, ldb,
 583                                                             msg, name);
 584                         }
 585                         return 0;
 586                 }
 587         }
 588
 589         return -1;
 590 }
 591
 592
 593 /*
 594   modify a record - internal interface
 595
 596   yuck - this is O(n^2). Luckily n is usually small so we probably
 597   get away with it, but if we ever have really large attribute lists
 598   then we'll need to look at this again
 599 */
 600 int ltdb_modify_internal(struct ldb_module *module,
 601                          const struct ldb_message *msg)
 602 {
 603         struct ldb_context *ldb = module->ldb;
 604         struct ltdb_private *ltdb =
 605                 talloc_get_type(module->private_data, struct ltdb_private);
 606         TDB_DATA tdb_key, tdb_data;
 607         struct ldb_message *msg2;
 608         unsigned i, j;
 609         int ret, idx;
 610
 611         tdb_key = ltdb_key(module, msg->dn);
 612         if (!tdb_key.dptr) {
 613                 return LDB_ERR_OTHER;
 614         }
 615
 616         tdb_data = tdb_fetch(ltdb->tdb, tdb_key);
 617         if (!tdb_data.dptr) {
 618                 talloc_free(tdb_key.dptr);
 619                 return ltdb_err_map(tdb_error(ltdb->tdb));
 620         }
 621
 622         msg2 = talloc(tdb_key.dptr, struct ldb_message);
 623         if (msg2 == NULL) {
 624                 talloc_free(tdb_key.dptr);
 625                 return LDB_ERR_OTHER;
 626         }
 627
 628         ret = ltdb_unpack_data(module, &tdb_data, msg2);
 629         if (ret == -1) {
 630                 ret = LDB_ERR_OTHER;
 631                 goto failed;
 632         }
 633
 634         if (!msg2->dn) {
 635                 msg2->dn = msg->dn;
 636         }
 637
 638         for (i=0;i<msg->num_elements;i++) {
 639                 struct ldb_message_element *el = &msg->elements[i];
 640                 struct ldb_message_element *el2;
 641                 struct ldb_val *vals;
 642                 const char *dn;
 643
 644                 switch (msg->elements[i].flags & LDB_FLAG_MOD_MASK) {
 645
 646                 case LDB_FLAG_MOD_ADD:
 647                         /* add this element to the message. fail if it
 648                            already exists */
 649                         idx = find_element(msg2, el->name);
 650
 651                         if (idx == -1) {
 652                                 if (msg_add_element(ldb, msg2, el) != 0) {
 653                                         ret = LDB_ERR_OTHER;
 654                                         goto failed;
 655                                 }
 656                                 continue;
 657                         }
 658
 659                         el2 = &msg2->elements[idx];
 660
 661                         /* An attribute with this name already exists,
 662                          * add all values if they don't already exist
 663                          * (check both the other elements to be added,
 664                          * and those already in the db). */
 665
 666                         for (j=0;j<el->num_values;j++) {
 667                                 if (ldb_msg_find_val(el2, &el->values[j])) {
 668                                         ldb_asprintf_errstring(module->ldb, "%s: value #%d already exists", el->name, j);
 669                                         ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
 670                                         goto failed;
 671                                 }
 672                                 if (ldb_msg_find_val(el, &el->values[j]) != &el->values[j]) {
 673                                         ldb_asprintf_errstring(module->ldb, "%s: value #%d provided more than once", el->name, j);
 674                                         ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
 675                                         goto failed;
 676                                 }
 677                         }
 678
 679                         vals = talloc_realloc(msg2->elements, el2->values, struct ldb_val,
 680                                                 el2->num_values + el->num_values);
 681
 682                         if (vals == NULL) {
 683                                 ret = LDB_ERR_OTHER;
 684                                 goto failed;
 685                         }
 686
 687                         for (j=0;j<el->num_values;j++) {
 688                                 vals[el2->num_values + j] =
 689                                         ldb_val_dup(vals, &el->values[j]);
 690                         }
 691
 692                         el2->values = vals;
 693                         el2->num_values += el->num_values;
 694
 695                         break;
 696
 697                 case LDB_FLAG_MOD_REPLACE:
 698                         /* replace all elements of this attribute name with the elements
 699                            listed. The attribute not existing is not an error */
 700                         msg_delete_attribute(module, ldb, msg2, el->name);
 701
 702                         for (j=0;j<el->num_values;j++) {
 703                                 if (ldb_msg_find_val(el, &el->values[j]) != &el->values[j]) {
 704                                         ldb_asprintf_errstring(module->ldb, "%s: value #%d provided more than once", el->name, j);
 705                                         ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
 706                                         goto failed;
 707                                 }
 708                         }
 709
 710                         /* add the replacement element, if not empty */
 711                         if (el->num_values != 0 &&
 712                             msg_add_element(ldb, msg2, el) != 0) {
 713                                 ret = LDB_ERR_OTHER;
 714                                 goto failed;
 715                         }
 716                         break;
 717
 718                 case LDB_FLAG_MOD_DELETE:
 719
 720                         dn = ldb_dn_get_linearized(msg->dn);
 721                         if (dn == NULL) {
 722                                 ret = LDB_ERR_OTHER;
 723                                 goto failed;
 724                         }
 725
 726                         /* we could be being asked to delete all
 727                            values or just some values */
 728                         if (msg->elements[i].num_values == 0) {
 729                                 if (msg_delete_attribute(module, ldb, msg2,
 730                                                          msg->elements[i].name) != 0) {
 731                                         ldb_asprintf_errstring(module->ldb, "No such attribute: %s for delete on %s", msg->elements[i].name, dn);
 732                                         ret = LDB_ERR_NO_SUCH_ATTRIBUTE;
 733                                         goto failed;
 734                                 }
 735                                 break;
 736                         }
 737                         for (j=0;j<msg->elements[i].num_values;j++) {
 738                                 if (msg_delete_element(module,
 739                                                        msg2,
 740                                                        msg->elements[i].name,
 741                                                        &msg->elements[i].values[j]) != 0) {
 742                                         ldb_asprintf_errstring(module->ldb, "No matching attribute value when deleting attribute: %s on %s", msg->elements[i].name, dn);
 743                                         ret = LDB_ERR_NO_SUCH_ATTRIBUTE;
 744                                         goto failed;
 745                                 }
 746                                 ret = ltdb_index_del_value(module, dn, &msg->elements[i], j);
 747                                 if (ret != LDB_SUCCESS) {
 748                                         goto failed;
 749                                 }
 750                         }
 751                         break;
 752                 default:
 753                         ldb_asprintf_errstring(module->ldb,
 754                                 "Invalid ldb_modify flags on %s: 0x%x",
 755                                 msg->elements[i].name,
 756                                 msg->elements[i].flags & LDB_FLAG_MOD_MASK);
 757                         ret = LDB_ERR_PROTOCOL_ERROR;
 758                         goto failed;
 759                 }
 760         }
 761
 762         /* we've made all the mods
 763          * save the modified record back into the database */
 764         ret = ltdb_store(module, msg2, TDB_MODIFY);
 765         if (ret != LDB_SUCCESS) {
 766                 goto failed;
 767         }
 768
 769         ret = ltdb_modified(module, msg->dn);
 770         if (ret != LDB_SUCCESS) {
 771                 goto failed;
 772         }
 773
 774         talloc_free(tdb_key.dptr);
 775         free(tdb_data.dptr);
 776         return ret;
 777
 778 failed:
 779         talloc_free(tdb_key.dptr);
 780         free(tdb_data.dptr);
 781         return ret;
 782 }
 783
 784 /*
 785   modify a record
 786 */
 787 static int ltdb_modify(struct ldb_module *module, struct ldb_request *req)
 788 {
 789         struct ltdb_private *ltdb;
 790         struct ltdb_context *ltdb_ac;
 791         int tret, ret = LDB_SUCCESS;
 792
 793         ltdb = talloc_get_type(module->private_data, struct ltdb_private);
 794
 795         if (check_critical_controls(req->controls)) {
 796                 return LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
 797         }
 798
 799         req->handle = NULL;
 800
 801         req->handle = init_ltdb_handle(ltdb, module, req);
 802         if (req->handle == NULL) {
 803                 return LDB_ERR_OPERATIONS_ERROR;
 804         }
 805         ltdb_ac = talloc_get_type(req->handle->private_data, struct ltdb_context);
 806
 807         tret = ltdb_check_special_dn(module, req->op.mod.message);
 808         if (tret != LDB_SUCCESS) {
 809                 req->handle->status = tret;
 810                 goto done;
 811         }
 812
 813         if (ltdb_cache_load(module) != 0) {
 814                 ret = LDB_ERR_OPERATIONS_ERROR;
 815                 goto done;
 816         }
 817
 818         tret = ltdb_modify_internal(module, req->op.mod.message);
 819         if (tret != LDB_SUCCESS) {
 820                 req->handle->status = tret;
 821                 goto done;
 822         }
 823
 824         if (ltdb_ac->callback) {
 825                 ret = ltdb_ac->callback(module->ldb, ltdb_ac->context, NULL);
 826         }
 827 done:
 828         req->handle->state = LDB_ASYNC_DONE;
 829         return ret;
 830 }
 831
 832 /*
 833   rename a record
 834 */
 835 static int ltdb_rename(struct ldb_module *module, struct ldb_request *req)
 836 {
 837         struct ltdb_private *ltdb;
 838         struct ltdb_context *ltdb_ac;
 839         struct ldb_message *msg;
 840         int tret, ret = LDB_SUCCESS;
 841
 842         ltdb = talloc_get_type(module->private_data, struct ltdb_private);
 843
 844         if (check_critical_controls(req->controls)) {
 845                 return LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
 846         }
 847
 848         req->handle = NULL;
 849
 850         if (ltdb_cache_load(module) != 0) {
 851                 return LDB_ERR_OPERATIONS_ERROR;
 852         }
 853
 854         req->handle = init_ltdb_handle(ltdb, module, req);
 855         if (req->handle == NULL) {
 856                 return LDB_ERR_OPERATIONS_ERROR;
 857         }
 858         ltdb_ac = talloc_get_type(req->handle->private_data, struct ltdb_context);
 859
 860         msg = talloc(ltdb_ac, struct ldb_message);
 861         if (msg == NULL) {
 862                 ret = LDB_ERR_OPERATIONS_ERROR;
 863                 goto done;
 864         }
 865
 866         /* in case any attribute of the message was indexed, we need
 867            to fetch the old record */
 868         tret = ltdb_search_dn1(module, req->op.rename.olddn, msg);
 869         if (tret != LDB_SUCCESS) {
 870                 /* not finding the old record is an error */
 871                 req->handle->status = tret;
 872                 goto done;
 873         }
 874
 875         msg->dn = ldb_dn_copy(msg, req->op.rename.newdn);
 876         if (!msg->dn) {
 877                 ret = LDB_ERR_OPERATIONS_ERROR;
 878                 goto done;
 879         }
 880
 881         if (ldb_dn_compare(req->op.rename.olddn, req->op.rename.newdn) == 0) {
 882                 /* The rename operation is apparently only changing case -
 883                    the DNs are the same.  Delete the old DN before adding
 884                    the new one to avoid a TDB_ERR_EXISTS error.
 885
 886                    The only drawback to this is that if the delete
 887                    succeeds but the add fails, we rely on the
 888                    transaction to roll this all back. */
 889                 ret = ltdb_delete_internal(module, req->op.rename.olddn);
 890                 if (ret != LDB_SUCCESS) {
 891                         goto done;
 892                 }
 893
 894                 ret = ltdb_add_internal(module, msg);
 895                 if (ret != LDB_SUCCESS) {
 896                         goto done;
 897                 }
 898         } else {
 899                 /* The rename operation is changing DNs.  Try to add the new
 900                    DN first to avoid clobbering another DN not related to
 901                    this rename operation. */
 902                 ret = ltdb_add_internal(module, msg);
 903                 if (ret != LDB_SUCCESS) {
 904                         goto done;
 905                 }
 906
 907                 tret = ltdb_delete_internal(module, req->op.rename.olddn);
 908                 if (tret != LDB_SUCCESS) {
 909                         ltdb_delete_internal(module, req->op.rename.newdn);
 910                         ret = LDB_ERR_OPERATIONS_ERROR;
 911                         goto done;
 912                 }
 913         }
 914
 915         if (ltdb_ac->callback) {
 916                 ret = ltdb_ac->callback(module->ldb, ltdb_ac->context, NULL);
 917         }
 918 done:
 919         req->handle->state = LDB_ASYNC_DONE;
 920         return ret;
 921 }
 922
 923 static int ltdb_start_trans(struct ldb_module *module)
 924 {
 925         struct ltdb_private *ltdb =
 926                 talloc_get_type(module->private_data, struct ltdb_private);
 927
 928         if (tdb_transaction_start(ltdb->tdb) != 0) {
 929                 return ltdb_err_map(tdb_error(ltdb->tdb));
 930         }
 931
 932         ltdb->in_transaction++;
 933
 934         return LDB_SUCCESS;
 935 }
 936
 937 static int ltdb_end_trans(struct ldb_module *module)
 938 {
 939         struct ltdb_private *ltdb =
 940                 talloc_get_type(module->private_data, struct ltdb_private);
 941
 942         ltdb->in_transaction--;
 943
 944         if (tdb_transaction_commit(ltdb->tdb) != 0) {
 945                 return ltdb_err_map(tdb_error(ltdb->tdb));
 946         }
 947
 948         return LDB_SUCCESS;
 949 }
 950
 951 static int ltdb_del_trans(struct ldb_module *module)
 952 {
 953         struct ltdb_private *ltdb =
 954                 talloc_get_type(module->private_data, struct ltdb_private);
 955
 956         ltdb->in_transaction--;
 957
 958         if (tdb_transaction_cancel(ltdb->tdb) != 0) {
 959                 return ltdb_err_map(tdb_error(ltdb->tdb));
 960         }
 961
 962         return LDB_SUCCESS;
 963 }
 964
 965 static int ltdb_wait(struct ldb_handle *handle, enum ldb_wait_type type)
 966 {
 967         return handle->status;
 968 }
 969
 970 static int ltdb_request(struct ldb_module *module, struct ldb_request *req)
 971 {
 972         /* check for oustanding critical controls
 973          * and return an error if found */
 974         if (check_critical_controls(req->controls)) {
 975                 return LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
 976         }
 977
 978         /* search, add, modify, delete, rename are handled by their own,
 979          * no other op supported */
 980         return LDB_ERR_OPERATIONS_ERROR;
 981 }
 982
 983 /*
 984   return sequenceNumber from @BASEINFO
 985 */
 986 static int ltdb_sequence_number(struct ldb_module *module,
 987                                 struct ldb_request *req)
 988 {
 989         TALLOC_CTX *tmp_ctx;
 990         struct ldb_message *msg = NULL;
 991         struct ldb_dn *dn;
 992         const char *date;
 993         int tret;
 994
 995         tmp_ctx = talloc_new(req);
 996         if (tmp_ctx == NULL) {
 997                 talloc_free(tmp_ctx);
 998                 return LDB_ERR_OPERATIONS_ERROR;
 999         }
1000
1001         dn = ldb_dn_new(tmp_ctx, module->ldb, LTDB_BASEINFO);
1002
1003         msg = talloc(tmp_ctx, struct ldb_message);
1004         if (msg == NULL) {
1005                 talloc_free(tmp_ctx);
1006                 return LDB_ERR_OPERATIONS_ERROR;
1007         }
1008
1009         req->op.seq_num.flags = 0;
1010
1011         tret = ltdb_search_dn1(module, dn, msg);
1012         if (tret != LDB_SUCCESS) {
1013                 talloc_free(tmp_ctx);
1014                 /* zero is as good as anything when we don't know */
1015                 req->op.seq_num.seq_num = 0;
1016                 return LDB_SUCCESS;
1017         }
1018
1019         switch (req->op.seq_num.type) {
1020         case LDB_SEQ_HIGHEST_SEQ:
1021                 req->op.seq_num.seq_num = ldb_msg_find_attr_as_uint64(msg, LTDB_SEQUENCE_NUMBER, 0);
1022                 break;
1023         case LDB_SEQ_NEXT:
1024                 req->op.seq_num.seq_num = ldb_msg_find_attr_as_uint64(msg, LTDB_SEQUENCE_NUMBER, 0);
1025                 req->op.seq_num.seq_num++;
1026                 break;
1027         case LDB_SEQ_HIGHEST_TIMESTAMP:
1028                 date = ldb_msg_find_attr_as_string(msg, LTDB_MOD_TIMESTAMP, NULL);
1029                 if (date) {
1030                         req->op.seq_num.seq_num = ldb_string_to_time(date);
1031                 } else {
1032                         req->op.seq_num.seq_num = 0;
1033                         /* zero is as good as anything when we don't know */
1034                 }
1035                 break;
1036         }
1037         talloc_free(tmp_ctx);
1038         return LDB_SUCCESS;
1039 }
1040
1041 static const struct ldb_module_ops ltdb_ops = {
1042         .name              = "tdb",
1043         .search            = ltdb_search,
1044         .add               = ltdb_add,
1045         .modify            = ltdb_modify,
1046         .del               = ltdb_delete,
1047         .rename            = ltdb_rename,
1048         .request           = ltdb_request,
1049         .start_transaction = ltdb_start_trans,
1050         .end_transaction   = ltdb_end_trans,
1051         .del_transaction   = ltdb_del_trans,
1052         .wait              = ltdb_wait,
1053         .sequence_number   = ltdb_sequence_number
1054 };
1055
1056 /*
1057   connect to the database
1058 */
1059 static int ltdb_connect(struct ldb_context *ldb, const char *url,
1060                         unsigned int flags, const char *options[],
1061                         struct ldb_module **module)
1062 {
1063         const char *path;
1064         int tdb_flags, open_flags;
1065         struct ltdb_private *ltdb;
1066
1067         /* parse the url */
1068         if (strchr(url, ':')) {
1069                 if (strncmp(url, "tdb://", 6) != 0) {
1070                         ldb_debug(ldb, LDB_DEBUG_ERROR,
1071                                   "Invalid tdb URL '%s'", url);
1072                         return -1;
1073                 }
1074                 path = url+6;
1075         } else {
1076                 path = url;
1077         }
1078
1079         tdb_flags = TDB_DEFAULT | TDB_SEQNUM;
1080
1081         /* check for the 'nosync' option */
1082         if (flags & LDB_FLG_NOSYNC) {
1083                 tdb_flags |= TDB_NOSYNC;
1084         }
1085
1086         /* and nommap option */
1087         if (flags & LDB_FLG_NOMMAP) {
1088                 tdb_flags |= TDB_NOMMAP;
1089         }
1090
1091         if (flags & LDB_FLG_RDONLY) {
1092                 open_flags = O_RDONLY;
1093         } else {
1094                 open_flags = O_CREAT | O_RDWR;
1095         }
1096
1097         ltdb = talloc_zero(ldb, struct ltdb_private);
1098         if (!ltdb) {
1099                 ldb_oom(ldb);
1100                 return -1;
1101         }
1102
1103         /* note that we use quite a large default hash size */
1104         ltdb->tdb = ltdb_wrap_open(ltdb, path, 10000,
1105                                    tdb_flags, open_flags,
1106                                    ldb->create_perms, ldb);
1107         if (!ltdb->tdb) {
1108                 ldb_debug(ldb, LDB_DEBUG_ERROR,
1109                           "Unable to open tdb '%s'\n", path);
1110                 talloc_free(ltdb);
1111                 return -1;
1112         }
1113
1114         ltdb->sequence_number = 0;
1115
1116         *module = talloc(ldb, struct ldb_module);
1117         if (!module) {
1118                 ldb_oom(ldb);
1119                 talloc_free(ltdb);
1120                 return -1;
1121         }
1122         talloc_set_name_const(*module, "ldb_tdb backend");
1123         (*module)->ldb = ldb;
1124         (*module)->prev = (*module)->next = NULL;
1125         (*module)->private_data = ltdb;
1126         (*module)->ops = &ltdb_ops;
1127
1128         if (ltdb_cache_load(*module) != 0) {
1129                 talloc_free(*module);
1130                 talloc_free(ltdb);
1131                 return -1;
1132         }
1133
1134         return 0;
1135 }
1136
1137 const struct ldb_backend_ops ldb_tdb_backend_ops = {
1138         .name = "tdb",
1139         .connect_fn = ltdb_connect
1140 };