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