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