source4/lib/ldb/ldb_tdb/ldb_index.c

   1 /*
   2    ldb database library
   3
   4    Copyright (C) Andrew Tridgell  2004
   5
   6      ** NOTE! The following LGPL license applies to the ldb
   7      ** library. This does NOT imply that all of Samba is released
   8      ** under the LGPL
   9
  10    This library is free software; you can redistribute it and/or
  11    modify it under the terms of the GNU Lesser General Public
  12    License as published by the Free Software Foundation; either
  13    version 3 of the License, or (at your option) any later version.
  14
  15    This library is distributed in the hope that it will be useful,
  16    but WITHOUT ANY WARRANTY; without even the implied warranty of
  17    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  18    Lesser General Public License for more details.
  19
  20    You should have received a copy of the GNU Lesser General Public
  21    License along with this library; if not, see <http://www.gnu.org/licenses/>.
  22 */
  23
  24 /*
  25  *  Name: ldb
  26  *
  27  *  Component: ldb tdb backend - indexing
  28  *
  29  *  Description: indexing routines for ldb tdb backend
  30  *
  31  *  Author: Andrew Tridgell
  32  */
  33
  34 #include "ldb_includes.h"
  35
  36 #include "ldb_tdb.h"
  37
  38 /*
  39   find an element in a list, using the given comparison function and
  40   assuming that the list is already sorted using comp_fn
  41
  42   return -1 if not found, or the index of the first occurance of needle if found
  43 */
  44 static int ldb_list_find(const void *needle,
  45                          const void *base, size_t nmemb, size_t size,
  46                          comparison_fn_t comp_fn)
  47 {
  48         const char *base_p = (const char *)base;
  49         size_t min_i, max_i, test_i;
  50
  51         if (nmemb == 0) {
  52                 return -1;
  53         }
  54
  55         min_i = 0;
  56         max_i = nmemb-1;
  57
  58         while (min_i < max_i) {
  59                 int r;
  60
  61                 test_i = (min_i + max_i) / 2;
  62                 /* the following cast looks strange, but is
  63                  correct. The key to understanding it is that base_p
  64                  is a pointer to an array of pointers, so we have to
  65                  dereference it after casting to void **. The strange
  66                  const in the middle gives us the right type of pointer
  67                  after the dereference  (tridge) */
  68                 r = comp_fn(needle, *(void * const *)(base_p + (size * test_i)));
  69                 if (r == 0) {
  70                         /* scan back for first element */
  71                         while (test_i > 0 &&
  72                                comp_fn(needle, *(void * const *)(base_p + (size * (test_i-1)))) == 0) {
  73                                 test_i--;
  74                         }
  75                         return test_i;
  76                 }
  77                 if (r < 0) {
  78                         if (test_i == 0) {
  79                                 return -1;
  80                         }
  81                         max_i = test_i - 1;
  82                 }
  83                 if (r > 0) {
  84                         min_i = test_i + 1;
  85                 }
  86         }
  87
  88         if (comp_fn(needle, *(void * const *)(base_p + (size * min_i))) == 0) {
  89                 return min_i;
  90         }
  91
  92         return -1;
  93 }
  94
  95 struct dn_list {
  96         unsigned int count;
  97         char **dn;
  98 };
  99
 100 /*
 101   return the dn key to be used for an index
 102   caller frees
 103 */
 104 static struct ldb_dn *ltdb_index_key(struct ldb_context *ldb,
 105                                      const char *attr, const struct ldb_val *value)
 106 {
 107         struct ldb_dn *ret;
 108         struct ldb_val v;
 109         const struct ldb_schema_attribute *a;
 110         char *attr_folded;
 111         int r;
 112
 113         attr_folded = ldb_attr_casefold(ldb, attr);
 114         if (!attr_folded) {
 115                 return NULL;
 116         }
 117
 118         a = ldb_schema_attribute_by_name(ldb, attr);
 119         r = a->syntax->canonicalise_fn(ldb, ldb, value, &v);
 120         if (r != LDB_SUCCESS) {
 121                 const char *errstr = ldb_errstring(ldb);
 122                 /* canonicalisation can be refused. For example,
 123                    a attribute that takes wildcards will refuse to canonicalise
 124                    if the value contains a wildcard */
 125                 ldb_asprintf_errstring(ldb, "Failed to create index key for attribute '%s':%s%s%s",
 126                                        attr, ldb_strerror(r), (errstr?":":""), (errstr?errstr:""));
 127                 talloc_free(attr_folded);
 128                 return NULL;
 129         }
 130         if (ldb_should_b64_encode(&v)) {
 131                 char *vstr = ldb_base64_encode(ldb, (char *)v.data, v.length);
 132                 if (!vstr) return NULL;
 133                 ret = ldb_dn_new_fmt(ldb, ldb, "%s:%s::%s", LTDB_INDEX, attr_folded, vstr);
 134                 talloc_free(vstr);
 135         } else {
 136                 ret = ldb_dn_new_fmt(ldb, ldb, "%s:%s:%.*s", LTDB_INDEX, attr_folded, (int)v.length, (char *)v.data);
 137         }
 138
 139         if (v.data != value->data) {
 140                 talloc_free(v.data);
 141         }
 142         talloc_free(attr_folded);
 143
 144         return ret;
 145 }
 146
 147 /*
 148   see if a attribute value is in the list of indexed attributes
 149 */
 150 static int ldb_msg_find_idx(const struct ldb_message *msg, const char *attr,
 151                             unsigned int *v_idx, const char *key)
 152 {
 153         unsigned int i, j;
 154         for (i=0;i<msg->num_elements;i++) {
 155                 if (ldb_attr_cmp(msg->elements[i].name, key) == 0) {
 156                         const struct ldb_message_element *el = &msg->elements[i];
 157
 158                         if (attr == NULL) {
 159                                 /* in this case we are just looking to see if key is present,
 160                                    we are not spearching for a specific index */
 161                                 return 0;
 162                         }
 163
 164                         for (j=0;j<el->num_values;j++) {
 165                                 if (ldb_attr_cmp((char *)el->values[j].data, attr) == 0) {
 166                                         if (v_idx) {
 167                                                 *v_idx = j;
 168                                         }
 169                                         return i;
 170                                 }
 171                         }
 172                 }
 173         }
 174         return -1;
 175 }
 176
 177 /* used in sorting dn lists */
 178 static int list_cmp(const char **s1, const char **s2)
 179 {
 180         return strcmp(*s1, *s2);
 181 }
 182
 183 /*
 184   return a list of dn's that might match a simple indexed search or
 185  */
 186 static int ltdb_index_dn_simple(struct ldb_module *module,
 187                                 const struct ldb_parse_tree *tree,
 188                                 const struct ldb_message *index_list,
 189                                 struct dn_list *list)
 190 {
 191         struct ldb_context *ldb = module->ldb;
 192         struct ldb_dn *dn;
 193         int ret;
 194         unsigned int i, j;
 195         struct ldb_message *msg;
 196
 197         list->count = 0;
 198         list->dn = NULL;
 199
 200         /* if the attribute isn't in the list of indexed attributes then
 201            this node needs a full search */
 202         if (ldb_msg_find_idx(index_list, tree->u.equality.attr, NULL, LTDB_IDXATTR) == -1) {
 203                 return LDB_ERR_OPERATIONS_ERROR;
 204         }
 205
 206         /* the attribute is indexed. Pull the list of DNs that match the
 207            search criterion */
 208         dn = ltdb_index_key(ldb, tree->u.equality.attr, &tree->u.equality.value);
 209         if (!dn) return LDB_ERR_OPERATIONS_ERROR;
 210
 211         msg = talloc(list, struct ldb_message);
 212         if (msg == NULL) {
 213                 return LDB_ERR_OPERATIONS_ERROR;
 214         }
 215
 216         ret = ltdb_search_dn1(module, dn, msg);
 217         talloc_free(dn);
 218         if (ret != LDB_SUCCESS) {
 219                 return ret;
 220         }
 221
 222         for (i=0;i<msg->num_elements;i++) {
 223                 struct ldb_message_element *el;
 224
 225                 if (strcmp(msg->elements[i].name, LTDB_IDX) != 0) {
 226                         continue;
 227                 }
 228
 229                 el = &msg->elements[i];
 230
 231                 list->dn = talloc_array(list, char *, el->num_values);
 232                 if (!list->dn) {
 233                         talloc_free(msg);
 234                         return LDB_ERR_OPERATIONS_ERROR;
 235                 }
 236
 237                 for (j=0;j<el->num_values;j++) {
 238                         list->dn[list->count] =
 239                                 talloc_strdup(list->dn, (char *)el->values[j].data);
 240                         if (!list->dn[list->count]) {
 241                                 talloc_free(msg);
 242                                 return LDB_ERR_OPERATIONS_ERROR;
 243                         }
 244                         list->count++;
 245                 }
 246         }
 247
 248         talloc_free(msg);
 249
 250         if (list->count > 1) {
 251                 qsort(list->dn, list->count, sizeof(char *), (comparison_fn_t) list_cmp);
 252         }
 253
 254         return LDB_SUCCESS;
 255 }
 256
 257
 258 static int list_union(struct ldb_context *, struct dn_list *, const struct dn_list *);
 259
 260 /*
 261   return a list of dn's that might match a leaf indexed search
 262  */
 263 static int ltdb_index_dn_leaf(struct ldb_module *module,
 264                               const struct ldb_parse_tree *tree,
 265                               const struct ldb_message *index_list,
 266                               struct dn_list *list)
 267 {
 268         if (ldb_attr_dn(tree->u.equality.attr) == 0) {
 269                 list->dn = talloc_array(list, char *, 1);
 270                 if (list->dn == NULL) {
 271                         ldb_oom(module->ldb);
 272                         return LDB_ERR_OPERATIONS_ERROR;
 273                 }
 274                 list->dn[0] = talloc_strdup(list->dn, (char *)tree->u.equality.value.data);
 275                 if (list->dn[0] == NULL) {
 276                         ldb_oom(module->ldb);
 277                         return LDB_ERR_OPERATIONS_ERROR;
 278                 }
 279                 list->count = 1;
 280                 return LDB_SUCCESS;
 281         }
 282         return ltdb_index_dn_simple(module, tree, index_list, list);
 283 }
 284
 285
 286 /*
 287   list intersection
 288   list = list & list2
 289   relies on the lists being sorted
 290 */
 291 static int list_intersect(struct ldb_context *ldb,
 292                           struct dn_list *list, const struct dn_list *list2)
 293 {
 294         struct dn_list *list3;
 295         unsigned int i;
 296
 297         if (list->count == 0 || list2->count == 0) {
 298                 /* 0 & X == 0 */
 299                 return LDB_ERR_NO_SUCH_OBJECT;
 300         }
 301
 302         list3 = talloc(ldb, struct dn_list);
 303         if (list3 == NULL) {
 304                 return LDB_ERR_OPERATIONS_ERROR;
 305         }
 306
 307         list3->dn = talloc_array(list3, char *, list->count);
 308         if (!list3->dn) {
 309                 talloc_free(list3);
 310                 return LDB_ERR_OPERATIONS_ERROR;
 311         }
 312         list3->count = 0;
 313
 314         for (i=0;i<list->count;i++) {
 315                 if (ldb_list_find(list->dn[i], list2->dn, list2->count,
 316                               sizeof(char *), (comparison_fn_t)strcmp) != -1) {
 317                         list3->dn[list3->count] = talloc_move(list3->dn, &list->dn[i]);
 318                         list3->count++;
 319                 } else {
 320                         talloc_free(list->dn[i]);
 321                 }
 322         }
 323
 324         talloc_free(list->dn);
 325         list->dn = talloc_move(list, &list3->dn);
 326         list->count = list3->count;
 327         talloc_free(list3);
 328
 329         return LDB_ERR_NO_SUCH_OBJECT;
 330 }
 331
 332
 333 /*
 334   list union
 335   list = list | list2
 336   relies on the lists being sorted
 337 */
 338 static int list_union(struct ldb_context *ldb,
 339                       struct dn_list *list, const struct dn_list *list2)
 340 {
 341         unsigned int i;
 342         char **d;
 343         unsigned int count = list->count;
 344
 345         if (list->count == 0 && list2->count == 0) {
 346                 /* 0 | 0 == 0 */
 347                 return LDB_ERR_NO_SUCH_OBJECT;
 348         }
 349
 350         d = talloc_realloc(list, list->dn, char *, list->count + list2->count);
 351         if (!d) {
 352                 return LDB_ERR_OPERATIONS_ERROR;
 353         }
 354         list->dn = d;
 355
 356         for (i=0;i<list2->count;i++) {
 357                 if (ldb_list_find(list2->dn[i], list->dn, count,
 358                               sizeof(char *), (comparison_fn_t)strcmp) == -1) {
 359                         list->dn[list->count] = talloc_strdup(list->dn, list2->dn[i]);
 360                         if (!list->dn[list->count]) {
 361                                 return LDB_ERR_OPERATIONS_ERROR;
 362                         }
 363                         list->count++;
 364                 }
 365         }
 366
 367         if (list->count != count) {
 368                 qsort(list->dn, list->count, sizeof(char *), (comparison_fn_t)list_cmp);
 369         }
 370
 371         return LDB_ERR_NO_SUCH_OBJECT;
 372 }
 373
 374 static int ltdb_index_dn(struct ldb_module *module,
 375                          const struct ldb_parse_tree *tree,
 376                          const struct ldb_message *index_list,
 377                          struct dn_list *list);
 378
 379
 380 /*
 381   OR two index results
 382  */
 383 static int ltdb_index_dn_or(struct ldb_module *module,
 384                             const struct ldb_parse_tree *tree,
 385                             const struct ldb_message *index_list,
 386                             struct dn_list *list)
 387 {
 388         struct ldb_context *ldb = module->ldb;
 389         unsigned int i;
 390         int ret;
 391
 392         ret = LDB_ERR_OPERATIONS_ERROR;
 393         list->dn = NULL;
 394         list->count = 0;
 395
 396         for (i=0;i<tree->u.list.num_elements;i++) {
 397                 struct dn_list *list2;
 398                 int v;
 399
 400                 list2 = talloc(module, struct dn_list);
 401                 if (list2 == NULL) {
 402                         return LDB_ERR_OPERATIONS_ERROR;
 403                 }
 404
 405                 v = ltdb_index_dn(module, tree->u.list.elements[i], index_list, list2);
 406
 407                 if (v == LDB_ERR_NO_SUCH_OBJECT) {
 408                         /* 0 || X == X */
 409                         if (ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
 410                                 ret = v;
 411                         }
 412                         talloc_free(list2);
 413                         continue;
 414                 }
 415
 416                 if (v != LDB_SUCCESS && v != LDB_ERR_NO_SUCH_OBJECT) {
 417                         /* 1 || X == 1 */
 418                         talloc_free(list->dn);
 419                         talloc_free(list2);
 420                         return v;
 421                 }
 422
 423                 if (ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
 424                         ret = LDB_SUCCESS;
 425                         list->dn = talloc_move(list, &list2->dn);
 426                         list->count = list2->count;
 427                 } else {
 428                         if (list_union(ldb, list, list2) == -1) {
 429                                 talloc_free(list2);
 430                                 return LDB_ERR_OPERATIONS_ERROR;
 431                         }
 432                         ret = LDB_SUCCESS;
 433                 }
 434                 talloc_free(list2);
 435         }
 436
 437         if (list->count == 0) {
 438                 return LDB_ERR_NO_SUCH_OBJECT;
 439         }
 440
 441         return ret;
 442 }
 443
 444
 445 /*
 446   NOT an index results
 447  */
 448 static int ltdb_index_dn_not(struct ldb_module *module,
 449                              const struct ldb_parse_tree *tree,
 450                              const struct ldb_message *index_list,
 451                              struct dn_list *list)
 452 {
 453         /* the only way to do an indexed not would be if we could
 454            negate the not via another not or if we knew the total
 455            number of database elements so we could know that the
 456            existing expression covered the whole database.
 457
 458            instead, we just give up, and rely on a full index scan
 459            (unless an outer & manages to reduce the list)
 460         */
 461         return LDB_ERR_OPERATIONS_ERROR;
 462 }
 463
 464 /*
 465   AND two index results
 466  */
 467 static int ltdb_index_dn_and(struct ldb_module *module,
 468                              const struct ldb_parse_tree *tree,
 469                              const struct ldb_message *index_list,
 470                              struct dn_list *list)
 471 {
 472         struct ldb_context *ldb = module->ldb;
 473         unsigned int i;
 474         int ret;
 475
 476         ret = LDB_ERR_OPERATIONS_ERROR;
 477         list->dn = NULL;
 478         list->count = 0;
 479
 480         for (i=0;i<tree->u.list.num_elements;i++) {
 481                 struct dn_list *list2;
 482                 int v;
 483
 484                 list2 = talloc(module, struct dn_list);
 485                 if (list2 == NULL) {
 486                         return LDB_ERR_OPERATIONS_ERROR;
 487                 }
 488
 489                 v = ltdb_index_dn(module, tree->u.list.elements[i], index_list, list2);
 490
 491                 if (v == LDB_ERR_NO_SUCH_OBJECT) {
 492                         /* 0 && X == 0 */
 493                         talloc_free(list->dn);
 494                         talloc_free(list2);
 495                         return LDB_ERR_NO_SUCH_OBJECT;
 496                 }
 497
 498                 if (v != LDB_SUCCESS && v != LDB_ERR_NO_SUCH_OBJECT) {
 499                         talloc_free(list2);
 500                         continue;
 501                 }
 502
 503                 if (ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
 504                         ret = LDB_SUCCESS;
 505                         talloc_free(list->dn);
 506                         list->dn = talloc_move(list, &list2->dn);
 507                         list->count = list2->count;
 508                 } else {
 509                         if (list_intersect(ldb, list, list2) == -1) {
 510                                 talloc_free(list2);
 511                                 return LDB_ERR_OPERATIONS_ERROR;
 512                         }
 513                 }
 514
 515                 talloc_free(list2);
 516
 517                 if (list->count == 0) {
 518                         talloc_free(list->dn);
 519                         return LDB_ERR_NO_SUCH_OBJECT;
 520                 }
 521         }
 522
 523         return ret;
 524 }
 525
 526 /*
 527   AND index results and ONE level special index
 528  */
 529 static int ltdb_index_dn_one(struct ldb_module *module,
 530                              struct ldb_dn *parent_dn,
 531                              struct dn_list *list)
 532 {
 533         struct ldb_context *ldb = module->ldb;
 534         struct dn_list *list2;
 535         struct ldb_message *msg;
 536         struct ldb_dn *key;
 537         struct ldb_val val;
 538         unsigned int i, j;
 539         int ret;
 540
 541         list2 = talloc_zero(module, struct dn_list);
 542         if (list2 == NULL) {
 543                 return LDB_ERR_OPERATIONS_ERROR;
 544         }
 545
 546         /* the attribute is indexed. Pull the list of DNs that match the
 547            search criterion */
 548         val.data = (uint8_t *)((uintptr_t)ldb_dn_get_casefold(parent_dn));
 549         val.length = strlen((char *)val.data);
 550         key = ltdb_index_key(ldb, LTDB_IDXONE, &val);
 551         if (!key) {
 552                 talloc_free(list2);
 553                 return LDB_ERR_OPERATIONS_ERROR;
 554         }
 555
 556         msg = talloc(list2, struct ldb_message);
 557         if (msg == NULL) {
 558                 talloc_free(list2);
 559                 return LDB_ERR_OPERATIONS_ERROR;
 560         }
 561
 562         ret = ltdb_search_dn1(module, key, msg);
 563         talloc_free(key);
 564         if (ret != LDB_SUCCESS) {
 565                 return ret;
 566         }
 567
 568         for (i = 0; i < msg->num_elements; i++) {
 569                 struct ldb_message_element *el;
 570
 571                 if (strcmp(msg->elements[i].name, LTDB_IDX) != 0) {
 572                         continue;
 573                 }
 574
 575                 el = &msg->elements[i];
 576
 577                 list2->dn = talloc_array(list2, char *, el->num_values);
 578                 if (!list2->dn) {
 579                         talloc_free(list2);
 580                         return LDB_ERR_OPERATIONS_ERROR;
 581                 }
 582
 583                 for (j = 0; j < el->num_values; j++) {
 584                         list2->dn[list2->count] = talloc_strdup(list2->dn, (char *)el->values[j].data);
 585                         if (!list2->dn[list2->count]) {
 586                                 talloc_free(list2);
 587                                 return LDB_ERR_OPERATIONS_ERROR;
 588                         }
 589                         list2->count++;
 590                 }
 591         }
 592
 593         if (list2->count == 0) {
 594                 talloc_free(list2);
 595                 return LDB_ERR_NO_SUCH_OBJECT;
 596         }
 597
 598         if (list2->count > 1) {
 599                 qsort(list2->dn, list2->count, sizeof(char *), (comparison_fn_t) list_cmp);
 600         }
 601
 602         if (list->count > 0) {
 603                 if (list_intersect(ldb, list, list2) == -1) {
 604                         talloc_free(list2);
 605                         return LDB_ERR_OPERATIONS_ERROR;
 606                 }
 607
 608                 if (list->count == 0) {
 609                         talloc_free(list->dn);
 610                         talloc_free(list2);
 611                         return LDB_ERR_NO_SUCH_OBJECT;
 612                 }
 613         } else {
 614                 list->dn = talloc_move(list, &list2->dn);
 615                 list->count = list2->count;
 616         }
 617
 618         talloc_free(list2);
 619
 620         return LDB_SUCCESS;
 621 }
 622
 623 /*
 624   return a list of dn's that might match a indexed search or
 625   an error. return LDB_ERR_NO_SUCH_OBJECT for no matches, or LDB_SUCCESS for matches
 626  */
 627 static int ltdb_index_dn(struct ldb_module *module,
 628                          const struct ldb_parse_tree *tree,
 629                          const struct ldb_message *index_list,
 630                          struct dn_list *list)
 631 {
 632         int ret = LDB_ERR_OPERATIONS_ERROR;
 633
 634         switch (tree->operation) {
 635         case LDB_OP_AND:
 636                 ret = ltdb_index_dn_and(module, tree, index_list, list);
 637                 break;
 638
 639         case LDB_OP_OR:
 640                 ret = ltdb_index_dn_or(module, tree, index_list, list);
 641                 break;
 642
 643         case LDB_OP_NOT:
 644                 ret = ltdb_index_dn_not(module, tree, index_list, list);
 645                 break;
 646
 647         case LDB_OP_EQUALITY:
 648                 ret = ltdb_index_dn_leaf(module, tree, index_list, list);
 649                 break;
 650
 651         case LDB_OP_SUBSTRING:
 652         case LDB_OP_GREATER:
 653         case LDB_OP_LESS:
 654         case LDB_OP_PRESENT:
 655         case LDB_OP_APPROX:
 656         case LDB_OP_EXTENDED:
 657                 /* we can't index with fancy bitops yet */
 658                 ret = LDB_ERR_OPERATIONS_ERROR;
 659                 break;
 660         }
 661
 662         return ret;
 663 }
 664
 665 /*
 666   filter a candidate dn_list from an indexed search into a set of results
 667   extracting just the given attributes
 668 */
 669 static int ltdb_index_filter(const struct dn_list *dn_list,
 670                              struct ltdb_context *ac)
 671 {
 672         struct ldb_message *msg;
 673         unsigned int i;
 674
 675         for (i = 0; i < dn_list->count; i++) {
 676                 struct ldb_dn *dn;
 677                 int ret;
 678
 679                 msg = ldb_msg_new(ac);
 680                 if (!msg) {
 681                         return LDB_ERR_OPERATIONS_ERROR;
 682                 }
 683
 684                 dn = ldb_dn_new(msg, ac->module->ldb, dn_list->dn[i]);
 685                 if (dn == NULL) {
 686                         talloc_free(msg);
 687                         return LDB_ERR_OPERATIONS_ERROR;
 688                 }
 689
 690                 ret = ltdb_search_dn1(ac->module, dn, msg);
 691                 talloc_free(dn);
 692                 if (ret == LDB_ERR_NO_SUCH_OBJECT) {
 693                         /* the record has disappeared? yes, this can happen */
 694                         talloc_free(msg);
 695                         continue;
 696                 }
 697
 698                 if (ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
 699                         /* an internal error */
 700                         talloc_free(msg);
 701                         return LDB_ERR_OPERATIONS_ERROR;
 702                 }
 703
 704                 if (!ldb_match_msg(ac->module->ldb, msg,
 705                                    ac->tree, ac->base, ac->scope)) {
 706                         talloc_free(msg);
 707                         continue;
 708                 }
 709
 710                 /* filter the attributes that the user wants */
 711                 ret = ltdb_filter_attrs(msg, ac->attrs);
 712
 713                 if (ret == -1) {
 714                         talloc_free(msg);
 715                         return LDB_ERR_OPERATIONS_ERROR;
 716                 }
 717
 718                 ret = ldb_module_send_entry(ac->req, msg);
 719                 if (ret != LDB_SUCCESS) {
 720                         ac->callback_failed = true;
 721                         return ret;
 722                 }
 723         }
 724
 725         return LDB_SUCCESS;
 726 }
 727
 728 /*
 729   search the database with a LDAP-like expression using indexes
 730   returns -1 if an indexed search is not possible, in which
 731   case the caller should call ltdb_search_full()
 732 */
 733 int ltdb_search_indexed(struct ltdb_context *ac)
 734 {
 735         struct ltdb_private *ltdb = talloc_get_type(ac->module->private_data, struct ltdb_private);
 736         struct dn_list *dn_list;
 737         int ret, idxattr, idxone;
 738
 739         idxattr = idxone = 0;
 740         ret = ldb_msg_find_idx(ltdb->cache->indexlist, NULL, NULL, LTDB_IDXATTR);
 741         if (ret == 0 ) {
 742                 idxattr = 1;
 743         }
 744
 745         /* We do one level indexing only if requested */
 746         ret = ldb_msg_find_idx(ltdb->cache->indexlist, NULL, NULL, LTDB_IDXONE);
 747         if (ret == 0 ) {
 748                 idxone = 1;
 749         }
 750
 751         if ((ac->scope == LDB_SCOPE_ONELEVEL && (idxattr+idxone == 0)) ||
 752             (ac->scope == LDB_SCOPE_SUBTREE && idxattr == 0)) {
 753                 /* no indexes? must do full search */
 754                 return LDB_ERR_OPERATIONS_ERROR;
 755         }
 756
 757         ret = LDB_ERR_OPERATIONS_ERROR;
 758
 759         dn_list = talloc_zero(ac, struct dn_list);
 760         if (dn_list == NULL) {
 761                 return LDB_ERR_OPERATIONS_ERROR;
 762         }
 763
 764         if (ac->scope == LDB_SCOPE_BASE) {
 765                 /* with BASE searches only one DN can match */
 766                 dn_list->dn = talloc_array(dn_list, char *, 1);
 767                 if (dn_list->dn == NULL) {
 768                         ldb_oom(ac->module->ldb);
 769                         return LDB_ERR_OPERATIONS_ERROR;
 770                 }
 771                 dn_list->dn[0] = ldb_dn_alloc_linearized(dn_list, ac->base);
 772                 if (dn_list->dn[0] == NULL) {
 773                         ldb_oom(ac->module->ldb);
 774                         return LDB_ERR_OPERATIONS_ERROR;
 775                 }
 776                 dn_list->count = 1;
 777                 ret = LDB_SUCCESS;
 778         }
 779
 780         if (ac->scope != LDB_SCOPE_BASE && idxattr == 1) {
 781                 ret = ltdb_index_dn(ac->module, ac->tree, ltdb->cache->indexlist, dn_list);
 782
 783                 if (ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
 784                         talloc_free(dn_list);
 785                         return ret;
 786                 }
 787         }
 788
 789         if (ac->scope == LDB_SCOPE_ONELEVEL && idxone == 1) {
 790                 ret = ltdb_index_dn_one(ac->module, ac->base, dn_list);
 791         }
 792
 793         if (ret == LDB_SUCCESS) {
 794                 /* we've got a candidate list - now filter by the full tree
 795                    and extract the needed attributes */
 796                 ret = ltdb_index_filter(dn_list, ac);
 797         }
 798
 799         talloc_free(dn_list);
 800
 801         return ret;
 802 }
 803
 804 /*
 805   add a index element where this is the first indexed DN for this value
 806 */
 807 static int ltdb_index_add1_new(struct ldb_context *ldb,
 808                                struct ldb_message *msg,
 809                                const char *dn)
 810 {
 811         struct ldb_message_element *el;
 812
 813         /* add another entry */
 814         el = talloc_realloc(msg, msg->elements,
 815                                struct ldb_message_element, msg->num_elements+1);
 816         if (!el) {
 817                 return LDB_ERR_OPERATIONS_ERROR;
 818         }
 819
 820         msg->elements = el;
 821         msg->elements[msg->num_elements].name = talloc_strdup(msg->elements, LTDB_IDX);
 822         if (!msg->elements[msg->num_elements].name) {
 823                 return LDB_ERR_OPERATIONS_ERROR;
 824         }
 825         msg->elements[msg->num_elements].num_values = 0;
 826         msg->elements[msg->num_elements].values = talloc(msg->elements, struct ldb_val);
 827         if (!msg->elements[msg->num_elements].values) {
 828                 return LDB_ERR_OPERATIONS_ERROR;
 829         }
 830         msg->elements[msg->num_elements].values[0].length = strlen(dn);
 831         msg->elements[msg->num_elements].values[0].data = discard_const_p(uint8_t, dn);
 832         msg->elements[msg->num_elements].num_values = 1;
 833         msg->num_elements++;
 834
 835         return LDB_SUCCESS;
 836 }
 837
 838
 839 /*
 840   add a index element where this is not the first indexed DN for this
 841   value
 842 */
 843 static int ltdb_index_add1_add(struct ldb_context *ldb,
 844                                struct ldb_message *msg,
 845                                int idx,
 846                                const char *dn)
 847 {
 848         struct ldb_val *v2;
 849         unsigned int i;
 850
 851         /* for multi-valued attributes we can end up with repeats */
 852         for (i=0;i<msg->elements[idx].num_values;i++) {
 853                 if (strcmp(dn, (char *)msg->elements[idx].values[i].data) == 0) {
 854                         return LDB_SUCCESS;
 855                 }
 856         }
 857
 858         v2 = talloc_realloc(msg->elements, msg->elements[idx].values,
 859                               struct ldb_val,
 860                               msg->elements[idx].num_values+1);
 861         if (!v2) {
 862                 return LDB_ERR_OPERATIONS_ERROR;
 863         }
 864         msg->elements[idx].values = v2;
 865
 866         msg->elements[idx].values[msg->elements[idx].num_values].length = strlen(dn);
 867         msg->elements[idx].values[msg->elements[idx].num_values].data = discard_const_p(uint8_t, dn);
 868         msg->elements[idx].num_values++;
 869
 870         return LDB_SUCCESS;
 871 }
 872
 873 /*
 874   add an index entry for one message element
 875 */
 876 static int ltdb_index_add1(struct ldb_module *module, const char *dn,
 877                            struct ldb_message_element *el, int v_idx)
 878 {
 879         struct ldb_context *ldb = module->ldb;
 880         struct ldb_message *msg;
 881         struct ldb_dn *dn_key;
 882         int ret;
 883         unsigned int i;
 884
 885         msg = talloc(module, struct ldb_message);
 886         if (msg == NULL) {
 887                 errno = ENOMEM;
 888                 return LDB_ERR_OPERATIONS_ERROR;
 889         }
 890
 891         dn_key = ltdb_index_key(ldb, el->name, &el->values[v_idx]);
 892         if (!dn_key) {
 893                 talloc_free(msg);
 894                 return LDB_ERR_OPERATIONS_ERROR;
 895         }
 896         talloc_steal(msg, dn_key);
 897
 898         ret = ltdb_search_dn1(module, dn_key, msg);
 899         if (ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
 900                 talloc_free(msg);
 901                 return ret;
 902         }
 903
 904         if (ret == LDB_ERR_NO_SUCH_OBJECT) {
 905                 msg->dn = dn_key;
 906                 msg->num_elements = 0;
 907                 msg->elements = NULL;
 908         }
 909
 910         for (i=0;i<msg->num_elements;i++) {
 911                 if (strcmp(LTDB_IDX, msg->elements[i].name) == 0) {
 912                         break;
 913                 }
 914         }
 915
 916         if (i == msg->num_elements) {
 917                 ret = ltdb_index_add1_new(ldb, msg, dn);
 918         } else {
 919                 ret = ltdb_index_add1_add(ldb, msg, i, dn);
 920         }
 921
 922         if (ret == LDB_SUCCESS) {
 923                 ret = ltdb_store(module, msg, TDB_REPLACE);
 924         }
 925
 926         talloc_free(msg);
 927
 928         return ret;
 929 }
 930
 931 static int ltdb_index_add0(struct ldb_module *module, const char *dn,
 932                            struct ldb_message_element *elements, int num_el)
 933 {
 934         struct ltdb_private *ltdb = (struct ltdb_private *)module->private_data;
 935         int ret;
 936         unsigned int i, j;
 937
 938         if (dn[0] == '@') {
 939                 return LDB_SUCCESS;
 940         }
 941
 942         if (ltdb->cache->indexlist->num_elements == 0) {
 943                 /* no indexed fields */
 944                 return LDB_SUCCESS;
 945         }
 946
 947         for (i = 0; i < num_el; i++) {
 948                 ret = ldb_msg_find_idx(ltdb->cache->indexlist, elements[i].name,
 949                                        NULL, LTDB_IDXATTR);
 950                 if (ret == -1) {
 951                         continue;
 952                 }
 953                 for (j = 0; j < elements[i].num_values; j++) {
 954                         ret = ltdb_index_add1(module, dn, &elements[i], j);
 955                         if (ret != LDB_SUCCESS) {
 956                                 return ret;
 957                         }
 958                 }
 959         }
 960
 961         return LDB_SUCCESS;
 962 }
 963
 964 /*
 965   add the index entries for a new record
 966 */
 967 int ltdb_index_add(struct ldb_module *module, const struct ldb_message *msg)
 968 {
 969         const char *dn;
 970         int ret;
 971
 972         dn = ldb_dn_get_linearized(msg->dn);
 973         if (dn == NULL) {
 974                 return LDB_ERR_OPERATIONS_ERROR;
 975         }
 976
 977         ret = ltdb_index_add0(module, dn, msg->elements, msg->num_elements);
 978
 979         return ret;
 980 }
 981
 982
 983 /*
 984   delete an index entry for one message element
 985 */
 986 int ltdb_index_del_value(struct ldb_module *module, const char *dn,
 987                          struct ldb_message_element *el, int v_idx)
 988 {
 989         struct ldb_context *ldb = module->ldb;
 990         struct ldb_message *msg;
 991         struct ldb_dn *dn_key;
 992         int ret, i;
 993         unsigned int j;
 994
 995         if (dn[0] == '@') {
 996                 return LDB_SUCCESS;
 997         }
 998
 999         dn_key = ltdb_index_key(ldb, el->name, &el->values[v_idx]);
1000         if (!dn_key) {
1001                 return LDB_ERR_OPERATIONS_ERROR;
1002         }
1003
1004         msg = talloc(dn_key, struct ldb_message);
1005         if (msg == NULL) {
1006                 talloc_free(dn_key);
1007                 return LDB_ERR_OPERATIONS_ERROR;
1008         }
1009
1010         ret = ltdb_search_dn1(module, dn_key, msg);
1011         if (ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
1012                 talloc_free(dn_key);
1013                 return ret;
1014         }
1015
1016         if (ret == LDB_ERR_NO_SUCH_OBJECT) {
1017                 /* it wasn't indexed. Did we have an earlier error? If we did then
1018                    its gone now */
1019                 talloc_free(dn_key);
1020                 return LDB_SUCCESS;
1021         }
1022
1023         i = ldb_msg_find_idx(msg, dn, &j, LTDB_IDX);
1024         if (i == -1) {
1025                 ldb_debug(ldb, LDB_DEBUG_ERROR,
1026                                 "ERROR: dn %s not found in %s\n", dn,
1027                                 ldb_dn_get_linearized(dn_key));
1028                 /* it ain't there. hmmm */
1029                 talloc_free(dn_key);
1030                 return LDB_SUCCESS;
1031         }
1032
1033         if (j != msg->elements[i].num_values - 1) {
1034                 memmove(&msg->elements[i].values[j],
1035                         &msg->elements[i].values[j+1],
1036                         (msg->elements[i].num_values-(j+1)) *
1037                         sizeof(msg->elements[i].values[0]));
1038         }
1039         msg->elements[i].num_values--;
1040
1041         if (msg->elements[i].num_values == 0) {
1042                 ret = ltdb_delete_noindex(module, dn_key);
1043         } else {
1044                 ret = ltdb_store(module, msg, TDB_REPLACE);
1045         }
1046
1047         talloc_free(dn_key);
1048
1049         return ret;
1050 }
1051
1052 /*
1053   delete the index entries for a record
1054   return -1 on failure
1055 */
1056 int ltdb_index_del(struct ldb_module *module, const struct ldb_message *msg)
1057 {
1058         struct ltdb_private *ltdb = (struct ltdb_private *)module->private_data;
1059         int ret;
1060         const char *dn;
1061         unsigned int i, j;
1062
1063         /* find the list of indexed fields */
1064         if (ltdb->cache->indexlist->num_elements == 0) {
1065                 /* no indexed fields */
1066                 return LDB_SUCCESS;
1067         }
1068
1069         if (ldb_dn_is_special(msg->dn)) {
1070                 return LDB_SUCCESS;
1071         }
1072
1073         dn = ldb_dn_get_linearized(msg->dn);
1074         if (dn == NULL) {
1075                 return LDB_ERR_OPERATIONS_ERROR;
1076         }
1077
1078         for (i = 0; i < msg->num_elements; i++) {
1079                 ret = ldb_msg_find_idx(ltdb->cache->indexlist, msg->elements[i].name,
1080                                        NULL, LTDB_IDXATTR);
1081                 if (ret == -1) {
1082                         continue;
1083                 }
1084                 for (j = 0; j < msg->elements[i].num_values; j++) {
1085                         ret = ltdb_index_del_value(module, dn, &msg->elements[i], j);
1086                         if (ret != LDB_SUCCESS) {
1087                                 return ret;
1088                         }
1089                 }
1090         }
1091
1092         return LDB_SUCCESS;
1093 }
1094
1095 /*
1096   handle special index for one level searches
1097 */
1098 int ltdb_index_one(struct ldb_module *module, const struct ldb_message *msg, int add)
1099 {
1100         struct ltdb_private *ltdb = (struct ltdb_private *)module->private_data;
1101         struct ldb_message_element el;
1102         struct ldb_val val;
1103         struct ldb_dn *pdn;
1104         const char *dn;
1105         int ret;
1106
1107         /* We index for ONE Level only if requested */
1108         ret = ldb_msg_find_idx(ltdb->cache->indexlist, NULL, NULL, LTDB_IDXONE);
1109         if (ret != 0) {
1110                 return LDB_SUCCESS;
1111         }
1112
1113         pdn = ldb_dn_get_parent(module, msg->dn);
1114         if (pdn == NULL) {
1115                 return LDB_ERR_OPERATIONS_ERROR;
1116         }
1117
1118         dn = ldb_dn_get_linearized(msg->dn);
1119         if (dn == NULL) {
1120                 talloc_free(pdn);
1121                 return LDB_ERR_OPERATIONS_ERROR;
1122         }
1123
1124         val.data = (uint8_t *)((uintptr_t)ldb_dn_get_casefold(pdn));
1125         if (val.data == NULL) {
1126                 talloc_free(pdn);
1127                 return LDB_ERR_OPERATIONS_ERROR;
1128         }
1129
1130         val.length = strlen((char *)val.data);
1131         el.name = LTDB_IDXONE;
1132         el.values = &val;
1133         el.num_values = 1;
1134
1135         if (add) {
1136                 ret = ltdb_index_add1(module, dn, &el, 0);
1137         } else { /* delete */
1138                 ret = ltdb_index_del_value(module, dn, &el, 0);
1139         }
1140
1141         talloc_free(pdn);
1142
1143         return ret;
1144 }
1145
1146
1147 /*
1148   traversal function that deletes all @INDEX records
1149 */
1150 static int delete_index(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *state)
1151 {
1152         const char *dn = "DN=" LTDB_INDEX ":";
1153         if (strncmp((char *)key.dptr, dn, strlen(dn)) == 0) {
1154                 return tdb_delete(tdb, key);
1155         }
1156         return 0;
1157 }
1158
1159 /*
1160   traversal function that adds @INDEX records during a re index
1161 */
1162 static int re_index(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *state)
1163 {
1164         struct ldb_module *module = (struct ldb_module *)state;
1165         struct ldb_message *msg;
1166         const char *dn = NULL;
1167         int ret;
1168         TDB_DATA key2;
1169
1170         if (strncmp((char *)key.dptr, "DN=@", 4) == 0 ||
1171             strncmp((char *)key.dptr, "DN=", 3) != 0) {
1172                 return 0;
1173         }
1174
1175         msg = talloc(module, struct ldb_message);
1176         if (msg == NULL) {
1177                 return -1;
1178         }
1179
1180         ret = ltdb_unpack_data(module, &data, msg);
1181         if (ret != 0) {
1182                 talloc_free(msg);
1183                 return -1;
1184         }
1185
1186         /* check if the DN key has changed, perhaps due to the
1187            case insensitivity of an element changing */
1188         key2 = ltdb_key(module, msg->dn);
1189         if (key2.dptr == NULL) {
1190                 /* probably a corrupt record ... darn */
1191                 ldb_debug(module->ldb, LDB_DEBUG_ERROR, "Invalid DN in re_index: %s\n",
1192                                                         ldb_dn_get_linearized(msg->dn));
1193                 talloc_free(msg);
1194                 return 0;
1195         }
1196         if (strcmp((char *)key2.dptr, (char *)key.dptr) != 0) {
1197                 tdb_delete(tdb, key);
1198                 tdb_store(tdb, key2, data, 0);
1199         }
1200         talloc_free(key2.dptr);
1201
1202         if (msg->dn == NULL) {
1203                 dn = (char *)key.dptr + 3;
1204         } else {
1205                 dn = ldb_dn_get_linearized(msg->dn);
1206         }
1207
1208         ret = ltdb_index_one(module, msg, 1);
1209         if (ret == LDB_SUCCESS) {
1210                 ret = ltdb_index_add0(module, dn, msg->elements, msg->num_elements);
1211         } else {
1212                 ldb_debug(module->ldb, LDB_DEBUG_ERROR,
1213                         "Adding special ONE LEVEL index failed (%s)!\n",
1214                         ldb_dn_get_linearized(msg->dn));
1215         }
1216
1217         talloc_free(msg);
1218
1219         if (ret != LDB_SUCCESS) return -1;
1220
1221         return 0;
1222 }
1223
1224 /*
1225   force a complete reindex of the database
1226 */
1227 int ltdb_reindex(struct ldb_module *module)
1228 {
1229         struct ltdb_private *ltdb = (struct ltdb_private *)module->private_data;
1230         int ret;
1231
1232         if (ltdb_cache_reload(module) != 0) {
1233                 return LDB_ERR_OPERATIONS_ERROR;
1234         }
1235
1236         /* first traverse the database deleting any @INDEX records */
1237         ret = tdb_traverse(ltdb->tdb, delete_index, NULL);
1238         if (ret == -1) {
1239                 return LDB_ERR_OPERATIONS_ERROR;
1240         }
1241
1242         /* if we don't have indexes we have nothing todo */
1243         if (ltdb->cache->indexlist->num_elements == 0) {
1244                 return LDB_SUCCESS;
1245         }
1246
1247         /* now traverse adding any indexes for normal LDB records */
1248         ret = tdb_traverse(ltdb->tdb, re_index, module);
1249         if (ret == -1) {
1250                 return LDB_ERR_OPERATIONS_ERROR;
1251         }
1252
1253         return LDB_SUCCESS;
1254 }