lib/ldb/ldb_key_value/ldb_kv_search.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 search functions
  28  *
  29  *  Description: functions to search ldb+tdb databases
  30  *
  31  *  Author: Andrew Tridgell
  32  */
  33
  34 #include "ldb_kv.h"
  35 #include "ldb_private.h"
  36
  37 /*
  38   add one element to a message
  39 */
  40 static int msg_add_element(struct ldb_message *ret,
  41                            const struct ldb_message_element *el,
  42                            int check_duplicates)
  43 {
  44         unsigned int i;
  45         struct ldb_message_element *e2, *elnew;
  46
  47         if (check_duplicates && ldb_msg_find_element(ret, el->name)) {
  48                 /* its already there */
  49                 return 0;
  50         }
  51
  52         e2 = talloc_realloc(ret, ret->elements, struct ldb_message_element, ret->num_elements+1);
  53         if (!e2) {
  54                 return -1;
  55         }
  56         ret->elements = e2;
  57
  58         elnew = &e2[ret->num_elements];
  59
  60         elnew->name = talloc_strdup(ret->elements, el->name);
  61         if (!elnew->name) {
  62                 return -1;
  63         }
  64
  65         if (el->num_values) {
  66                 elnew->values = talloc_array(ret->elements, struct ldb_val, el->num_values);
  67                 if (!elnew->values) {
  68                         return -1;
  69                 }
  70         } else {
  71                 elnew->values = NULL;
  72         }
  73
  74         for (i=0;i<el->num_values;i++) {
  75                 elnew->values[i] = ldb_val_dup(elnew->values, &el->values[i]);
  76                 if (elnew->values[i].length != el->values[i].length) {
  77                         return -1;
  78                 }
  79         }
  80
  81         elnew->num_values = el->num_values;
  82         elnew->flags = el->flags;
  83
  84         ret->num_elements++;
  85
  86         return 0;
  87 }
  88
  89 /*
  90   add the special distinguishedName element
  91 */
  92 static int msg_add_distinguished_name(struct ldb_message *msg)
  93 {
  94         struct ldb_message_element el;
  95         struct ldb_val val;
  96         int ret;
  97
  98         el.flags = 0;
  99         el.name = "distinguishedName";
 100         el.num_values = 1;
 101         el.values = &val;
 102         el.flags = 0;
 103         val.data = (uint8_t *)ldb_dn_alloc_linearized(msg, msg->dn);
 104         if (val.data == NULL) {
 105                 return -1;
 106         }
 107         val.length = strlen((char *)val.data);
 108
 109         ret = msg_add_element(msg, &el, 1);
 110         return ret;
 111 }
 112
 113 /*
 114   search the database for a single simple dn.
 115   return LDB_ERR_NO_SUCH_OBJECT on record-not-found
 116   and LDB_SUCCESS on success
 117 */
 118 int ldb_kv_search_base(struct ldb_module *module,
 119                        TALLOC_CTX *mem_ctx,
 120                        struct ldb_dn *dn,
 121                        struct ldb_dn **ret_dn)
 122 {
 123         int exists;
 124         int ret;
 125         struct ldb_message *msg = NULL;
 126
 127         if (ldb_dn_is_null(dn)) {
 128                 return LDB_ERR_NO_SUCH_OBJECT;
 129         }
 130
 131         /*
 132          * We can't use tdb_exists() directly on a key when the TDB
 133          * key is the GUID one, not the DN based one.  So we just do a
 134          * normal search and avoid most of the allocation with the
 135          * LDB_UNPACK_DATA_FLAG_NO_DN and
 136          * LDB_UNPACK_DATA_FLAG_NO_ATTRS flags
 137          */
 138         msg = ldb_msg_new(module);
 139         if (msg == NULL) {
 140                 return LDB_ERR_OPERATIONS_ERROR;
 141         }
 142
 143         ret = ldb_kv_search_dn1(module, dn, msg, LDB_UNPACK_DATA_FLAG_NO_ATTRS);
 144         if (ret == LDB_SUCCESS) {
 145                 const char *dn_linearized
 146                         = ldb_dn_get_linearized(dn);
 147                 const char *msg_dn_linearlized
 148                         = ldb_dn_get_linearized(msg->dn);
 149
 150                 if (strcmp(dn_linearized, msg_dn_linearlized) == 0) {
 151                         /*
 152                          * Re-use the full incoming DN for
 153                          * subtree checks
 154                          */
 155                         *ret_dn = dn;
 156                 } else {
 157                         /*
 158                          * Use the string DN from the unpack, so that
 159                          * we have a case-exact match of the base
 160                          */
 161                         *ret_dn = talloc_steal(mem_ctx, msg->dn);
 162                 }
 163                 exists = true;
 164         } else if (ret == LDB_ERR_NO_SUCH_OBJECT) {
 165                 exists = false;
 166         } else {
 167                 talloc_free(msg);
 168                 return ret;
 169         }
 170         talloc_free(msg);
 171         if (exists) {
 172                 return LDB_SUCCESS;
 173         }
 174         return LDB_ERR_NO_SUCH_OBJECT;
 175 }
 176
 177 struct ldb_kv_parse_data_unpack_ctx {
 178         struct ldb_message *msg;
 179         struct ldb_module *module;
 180         unsigned int unpack_flags;
 181 };
 182
 183 static int ldb_kv_parse_data_unpack(struct ldb_val key,
 184                                     struct ldb_val data,
 185                                     void *private_data)
 186 {
 187         struct ldb_kv_parse_data_unpack_ctx *ctx = private_data;
 188         unsigned int nb_elements_in_db;
 189         int ret;
 190         struct ldb_context *ldb = ldb_module_get_ctx(ctx->module);
 191         struct ldb_val data_parse = data;
 192
 193         if (ctx->unpack_flags & LDB_UNPACK_DATA_FLAG_NO_DATA_ALLOC) {
 194                 /*
 195                  * If we got LDB_UNPACK_DATA_FLAG_NO_DATA_ALLOC
 196                  * we need at least do a memdup on the whole
 197                  * data buffer as that may change later
 198                  * and the caller needs a stable result.
 199                  */
 200                 data_parse.data = talloc_memdup(ctx->msg,
 201                                                 data.data,
 202                                                 data.length);
 203                 if (data_parse.data == NULL) {
 204                         ldb_debug(ldb, LDB_DEBUG_ERROR,
 205                                   "Unable to allocate data(%d) for %*.*s\n",
 206                                   (int)data.length,
 207                                   (int)key.length, (int)key.length, key.data);
 208                         return LDB_ERR_OPERATIONS_ERROR;
 209                 }
 210         }
 211
 212         ret = ldb_unpack_data_only_attr_list_flags(ldb, &data_parse,
 213                                                    ctx->msg,
 214                                                    NULL, 0,
 215                                                    ctx->unpack_flags,
 216                                                    &nb_elements_in_db);
 217         if (ret == -1) {
 218                 if (data_parse.data != data.data) {
 219                         talloc_free(data_parse.data);
 220                 }
 221
 222                 ldb_debug(ldb, LDB_DEBUG_ERROR, "Invalid data for index %*.*s\n",
 223                           (int)key.length, (int)key.length, key.data);
 224                 return LDB_ERR_OPERATIONS_ERROR;
 225         }
 226         return ret;
 227 }
 228
 229 /*
 230   search the database for a single simple dn, returning all attributes
 231   in a single message
 232
 233   return LDB_ERR_NO_SUCH_OBJECT on record-not-found
 234   and LDB_SUCCESS on success
 235 */
 236 int ldb_kv_search_key(struct ldb_module *module,
 237                       struct ldb_kv_private *ldb_kv,
 238                       const struct ldb_val ldb_key,
 239                       struct ldb_message *msg,
 240                       unsigned int unpack_flags)
 241 {
 242         int ret;
 243         struct ldb_kv_parse_data_unpack_ctx ctx = {
 244                 .msg = msg,
 245                 .module = module,
 246                 .unpack_flags = unpack_flags
 247         };
 248
 249         memset(msg, 0, sizeof(*msg));
 250
 251         msg->num_elements = 0;
 252         msg->elements = NULL;
 253
 254         ret = ldb_kv->kv_ops->fetch_and_parse(
 255             ldb_kv, ldb_key, ldb_kv_parse_data_unpack, &ctx);
 256
 257         if (ret == -1) {
 258                 ret = ldb_kv->kv_ops->error(ldb_kv);
 259                 if (ret == LDB_SUCCESS) {
 260                         /*
 261                          * Just to be sure we don't turn errors
 262                          * into success
 263                          */
 264                         return LDB_ERR_OPERATIONS_ERROR;
 265                 }
 266                 return ret;
 267         } else if (ret != LDB_SUCCESS) {
 268                 return ret;
 269         }
 270
 271         return LDB_SUCCESS;
 272 }
 273
 274 /*
 275   search the database for a single simple dn, returning all attributes
 276   in a single message
 277
 278   return LDB_ERR_NO_SUCH_OBJECT on record-not-found
 279   and LDB_SUCCESS on success
 280 */
 281 int ldb_kv_search_dn1(struct ldb_module *module,
 282                       struct ldb_dn *dn,
 283                       struct ldb_message *msg,
 284                       unsigned int unpack_flags)
 285 {
 286         void *data = ldb_module_get_private(module);
 287         struct ldb_kv_private *ldb_kv =
 288             talloc_get_type(data, struct ldb_kv_private);
 289         int ret;
 290         uint8_t guid_key[LDB_KV_GUID_KEY_SIZE];
 291         struct ldb_val key = {
 292                 .data = guid_key,
 293                 .length = sizeof(guid_key)
 294         };
 295         TALLOC_CTX *tdb_key_ctx = NULL;
 296
 297         bool valid_dn = ldb_dn_validate(dn);
 298         if (valid_dn == false) {
 299                 ldb_asprintf_errstring(ldb_module_get_ctx(module),
 300                                        "Invalid Base DN: %s",
 301                                        ldb_dn_get_linearized(dn));
 302                 return LDB_ERR_INVALID_DN_SYNTAX;
 303         }
 304
 305         if (ldb_kv->cache->GUID_index_attribute == NULL ||
 306             ldb_dn_is_special(dn)) {
 307
 308                 tdb_key_ctx = talloc_new(msg);
 309                 if (!tdb_key_ctx) {
 310                         return ldb_module_oom(module);
 311                 }
 312
 313                 /* form the key */
 314                 key = ldb_kv_key_dn(module, tdb_key_ctx, dn);
 315                 if (!key.data) {
 316                         TALLOC_FREE(tdb_key_ctx);
 317                         return LDB_ERR_OPERATIONS_ERROR;
 318                 }
 319         } else {
 320                 /*
 321                  * Look in the index to find the key for this DN.
 322                  *
 323                  * the tdb_key memory is allocated above, msg is just
 324                  * used for internal memory.
 325                  *
 326                  */
 327                 ret = ldb_kv_key_dn_from_idx(module, ldb_kv, msg, dn, &key);
 328                 if (ret != LDB_SUCCESS) {
 329                         return ret;
 330                 }
 331         }
 332
 333         ret = ldb_kv_search_key(module, ldb_kv, key, msg, unpack_flags);
 334
 335         TALLOC_FREE(tdb_key_ctx);
 336
 337         if (ret != LDB_SUCCESS) {
 338                 return ret;
 339         }
 340
 341         if ((unpack_flags & LDB_UNPACK_DATA_FLAG_NO_DN) == 0) {
 342                 if (!msg->dn) {
 343                         msg->dn = ldb_dn_copy(msg, dn);
 344                 }
 345                 if (!msg->dn) {
 346                         return LDB_ERR_OPERATIONS_ERROR;
 347                 }
 348         }
 349
 350         return LDB_SUCCESS;
 351 }
 352
 353 /*
 354   filter the specified list of attributes from a message
 355   removing not requested attrs from the new message constructed.
 356
 357   The reason this makes a new message is that the old one may not be
 358   individually allocated, which is what our callers expect.
 359
 360  */
 361 int ldb_kv_filter_attrs(TALLOC_CTX *mem_ctx,
 362                         const struct ldb_message *msg,
 363                         const char *const *attrs,
 364                         struct ldb_message **filtered_msg)
 365 {
 366         unsigned int i;
 367         bool keep_all = false;
 368         bool add_dn = false;
 369         uint32_t num_elements;
 370         uint32_t elements_size;
 371         struct ldb_message *msg2;
 372
 373         msg2 = ldb_msg_new(mem_ctx);
 374         if (msg2 == NULL) {
 375                 goto failed;
 376         }
 377
 378         msg2->dn = ldb_dn_copy(msg2, msg->dn);
 379         if (msg2->dn == NULL) {
 380                 goto failed;
 381         }
 382
 383         if (attrs) {
 384                 /* check for special attrs */
 385                 for (i = 0; attrs[i]; i++) {
 386                         int cmp = strcmp(attrs[i], "*");
 387                         if (cmp == 0) {
 388                                 keep_all = true;
 389                                 break;
 390                         }
 391                         cmp = ldb_attr_cmp(attrs[i], "distinguishedName");
 392                         if (cmp == 0) {
 393                                 add_dn = true;
 394                         }
 395                 }
 396         } else {
 397                 keep_all = true;
 398         }
 399
 400         if (keep_all) {
 401                 add_dn = true;
 402                 elements_size = msg->num_elements + 1;
 403
 404         /* Shortcuts for the simple cases */
 405         } else if (add_dn && i == 1) {
 406                 if (msg_add_distinguished_name(msg2) != 0) {
 407                         goto failed;
 408                 }
 409                 *filtered_msg = msg2;
 410                 return 0;
 411         } else if (i == 0) {
 412                 *filtered_msg = msg2;
 413                 return 0;
 414
 415         /* Otherwise we are copying at most as many element as we have attributes */
 416         } else {
 417                 elements_size = i;
 418         }
 419
 420         msg2->elements = talloc_array(msg2, struct ldb_message_element,
 421                                       elements_size);
 422         if (msg2->elements == NULL) goto failed;
 423
 424         num_elements = 0;
 425
 426         for (i = 0; i < msg->num_elements; i++) {
 427                 struct ldb_message_element *el = &msg->elements[i];
 428                 struct ldb_message_element *el2 = &msg2->elements[num_elements];
 429                 unsigned int j;
 430
 431                 if (keep_all == false) {
 432                         bool found = false;
 433                         for (j = 0; attrs[j]; j++) {
 434                                 int cmp = ldb_attr_cmp(el->name, attrs[j]);
 435                                 if (cmp == 0) {
 436                                         found = true;
 437                                         break;
 438                                 }
 439                         }
 440                         if (found == false) {
 441                                 continue;
 442                         }
 443                 }
 444                 *el2 = *el;
 445                 el2->name = talloc_strdup(msg2->elements, el->name);
 446                 if (el2->name == NULL) {
 447                         goto failed;
 448                 }
 449                 el2->values = talloc_array(msg2->elements, struct ldb_val, el->num_values);
 450                 if (el2->values == NULL) {
 451                         goto failed;
 452                 }
 453                 for (j=0;j<el->num_values;j++) {
 454                         el2->values[j] = ldb_val_dup(el2->values, &el->values[j]);
 455                         if (el2->values[j].data == NULL && el->values[j].length != 0) {
 456                                 goto failed;
 457                         }
 458                 }
 459                 num_elements++;
 460
 461                 /* Pidginhole principle: we can't have more elements
 462                  * than the number of attributes if they are unique in
 463                  * the DB */
 464                 if (num_elements > elements_size) {
 465                         goto failed;
 466                 }
 467         }
 468
 469         msg2->num_elements = num_elements;
 470
 471         if (add_dn) {
 472                 if (msg_add_distinguished_name(msg2) != 0) {
 473                         goto failed;
 474                 }
 475         }
 476
 477         if (msg2->num_elements > 0) {
 478                 msg2->elements = talloc_realloc(msg2, msg2->elements,
 479                                                 struct ldb_message_element,
 480                                                 msg2->num_elements);
 481                 if (msg2->elements == NULL) {
 482                         goto failed;
 483                 }
 484         } else {
 485                 talloc_free(msg2->elements);
 486                 msg2->elements = NULL;
 487         }
 488
 489         *filtered_msg = msg2;
 490
 491         return 0;
 492 failed:
 493         TALLOC_FREE(msg2);
 494         return -1;
 495 }
 496
 497 /*
 498   search function for a non-indexed search
 499  */
 500 static int search_func(struct ldb_kv_private *ldb_kv,
 501                        struct ldb_val key,
 502                        struct ldb_val val,
 503                        void *state)
 504 {
 505         struct ldb_context *ldb;
 506         struct ldb_kv_context *ac;
 507         struct ldb_message *msg, *filtered_msg;
 508         int ret;
 509         bool matched;
 510         unsigned int nb_elements_in_db;
 511
 512         ac = talloc_get_type(state, struct ldb_kv_context);
 513         ldb = ldb_module_get_ctx(ac->module);
 514
 515         if (ldb_kv_key_is_record(key) == false) {
 516                 return 0;
 517         }
 518
 519         msg = ldb_msg_new(ac);
 520         if (!msg) {
 521                 ac->error = LDB_ERR_OPERATIONS_ERROR;
 522                 return -1;
 523         }
 524
 525         /* unpack the record */
 526         ret = ldb_unpack_data_only_attr_list_flags(ldb, &val,
 527                                                    msg,
 528                                                    NULL, 0,
 529                                                    LDB_UNPACK_DATA_FLAG_NO_DATA_ALLOC|
 530                                                    LDB_UNPACK_DATA_FLAG_NO_VALUES_ALLOC,
 531                                                    &nb_elements_in_db);
 532         if (ret == -1) {
 533                 talloc_free(msg);
 534                 ac->error = LDB_ERR_OPERATIONS_ERROR;
 535                 return -1;
 536         }
 537
 538         if (!msg->dn) {
 539                 msg->dn = ldb_dn_new(msg, ldb,
 540                                      (char *)key.data + 3);
 541                 if (msg->dn == NULL) {
 542                         talloc_free(msg);
 543                         ac->error = LDB_ERR_OPERATIONS_ERROR;
 544                         return -1;
 545                 }
 546         }
 547
 548         /* see if it matches the given expression */
 549         ret = ldb_match_msg_error(ldb, msg,
 550                                   ac->tree, ac->base, ac->scope, &matched);
 551         if (ret != LDB_SUCCESS) {
 552                 talloc_free(msg);
 553                 ac->error = LDB_ERR_OPERATIONS_ERROR;
 554                 return -1;
 555         }
 556         if (!matched) {
 557                 talloc_free(msg);
 558                 return 0;
 559         }
 560
 561         /* filter the attributes that the user wants */
 562         ret = ldb_kv_filter_attrs(ac, msg, ac->attrs, &filtered_msg);
 563         talloc_free(msg);
 564
 565         if (ret == -1) {
 566                 ac->error = LDB_ERR_OPERATIONS_ERROR;
 567                 return -1;
 568         }
 569
 570         ret = ldb_module_send_entry(ac->req, filtered_msg, NULL);
 571         if (ret != LDB_SUCCESS) {
 572                 ac->request_terminated = true;
 573                 /* the callback failed, abort the operation */
 574                 ac->error = LDB_ERR_OPERATIONS_ERROR;
 575                 return -1;
 576         }
 577
 578         return 0;
 579 }
 580
 581
 582 /*
 583   search the database with a LDAP-like expression.
 584   this is the "full search" non-indexed variant
 585 */
 586 static int ldb_kv_search_full(struct ldb_kv_context *ctx)
 587 {
 588         void *data = ldb_module_get_private(ctx->module);
 589         struct ldb_kv_private *ldb_kv =
 590             talloc_get_type(data, struct ldb_kv_private);
 591         int ret;
 592
 593         ctx->error = LDB_SUCCESS;
 594         ret = ldb_kv->kv_ops->iterate(ldb_kv, search_func, ctx);
 595
 596         if (ret < 0) {
 597                 return LDB_ERR_OPERATIONS_ERROR;
 598         }
 599
 600         return ctx->error;
 601 }
 602
 603 static int ldb_kv_search_and_return_base(struct ldb_kv_private *ldb_kv,
 604                                          struct ldb_kv_context *ctx)
 605 {
 606         struct ldb_message *msg, *filtered_msg;
 607         struct ldb_context *ldb = ldb_module_get_ctx(ctx->module);
 608         const char *dn_linearized;
 609         const char *msg_dn_linearlized;
 610         int ret;
 611         bool matched;
 612
 613         msg = ldb_msg_new(ctx);
 614         if (!msg) {
 615                 return LDB_ERR_OPERATIONS_ERROR;
 616         }
 617         ret = ldb_kv_search_dn1(ctx->module,
 618                                 ctx->base,
 619                                 msg,
 620                                 LDB_UNPACK_DATA_FLAG_NO_DATA_ALLOC |
 621                                     LDB_UNPACK_DATA_FLAG_NO_VALUES_ALLOC);
 622
 623         if (ret == LDB_ERR_NO_SUCH_OBJECT) {
 624                 if (ldb_kv->check_base == false) {
 625                         /*
 626                          * In this case, we are done, as no base
 627                          * checking is allowed in this DB
 628                          */
 629                         talloc_free(msg);
 630                         return LDB_SUCCESS;
 631                 }
 632                 ldb_asprintf_errstring(ldb,
 633                                        "No such Base DN: %s",
 634                                        ldb_dn_get_linearized(ctx->base));
 635         }
 636         if (ret != LDB_SUCCESS) {
 637                 talloc_free(msg);
 638                 return ret;
 639         }
 640
 641
 642         /*
 643          * We use this, not ldb_match_msg_error() as we know
 644          * we matched on the scope BASE, as we just fetched
 645          * the base DN
 646          */
 647
 648         ret = ldb_match_message(ldb, msg,
 649                                 ctx->tree,
 650                                 ctx->scope,
 651                                 &matched);
 652         if (ret != LDB_SUCCESS) {
 653                 talloc_free(msg);
 654                 return ret;
 655         }
 656         if (!matched) {
 657                 talloc_free(msg);
 658                 return LDB_SUCCESS;
 659         }
 660
 661         dn_linearized = ldb_dn_get_linearized(ctx->base);
 662         msg_dn_linearlized = ldb_dn_get_linearized(msg->dn);
 663
 664         if (strcmp(dn_linearized, msg_dn_linearlized) == 0) {
 665                 /*
 666                  * If the DN is exactly the same string, then
 667                  * re-use the full incoming DN for the
 668                  * returned result, as it has already been
 669                  * casefolded
 670                  */
 671                 msg->dn = ctx->base;
 672         }
 673
 674         /*
 675          * filter the attributes that the user wants.
 676          *
 677          * This copies msg->dn including the casefolding, so the above
 678          * assignment is safe
 679          */
 680         ret = ldb_kv_filter_attrs(ctx, msg, ctx->attrs, &filtered_msg);
 681
 682         /*
 683          * Remove any extended components possibly copied in from
 684          * msg->dn, we just want the casefold components
 685          */
 686         ldb_dn_remove_extended_components(filtered_msg->dn);
 687         talloc_free(msg);
 688
 689         if (ret == -1) {
 690                 return LDB_ERR_OPERATIONS_ERROR;
 691         }
 692
 693         ret = ldb_module_send_entry(ctx->req, filtered_msg, NULL);
 694         if (ret != LDB_SUCCESS) {
 695                 /* Regardless of success or failure, the msg
 696                  * is the callbacks responsiblity, and should
 697                  * not be talloc_free()'ed */
 698                 ctx->request_terminated = true;
 699                 return ret;
 700         }
 701
 702         return LDB_SUCCESS;
 703 }
 704
 705 /*
 706   search the database with a LDAP-like expression.
 707   choses a search method
 708 */
 709 int ldb_kv_search(struct ldb_kv_context *ctx)
 710 {
 711         struct ldb_context *ldb;
 712         struct ldb_module *module = ctx->module;
 713         struct ldb_request *req = ctx->req;
 714         void *data = ldb_module_get_private(module);
 715         struct ldb_kv_private *ldb_kv =
 716             talloc_get_type(data, struct ldb_kv_private);
 717         int ret;
 718
 719         ldb = ldb_module_get_ctx(module);
 720
 721         ldb_request_set_state(req, LDB_ASYNC_PENDING);
 722
 723         if (ldb_kv->kv_ops->lock_read(module) != 0) {
 724                 return LDB_ERR_OPERATIONS_ERROR;
 725         }
 726
 727         if (ldb_kv_cache_load(module) != 0) {
 728                 ldb_kv->kv_ops->unlock_read(module);
 729                 return LDB_ERR_OPERATIONS_ERROR;
 730         }
 731
 732         if (req->op.search.tree == NULL) {
 733                 ldb_kv->kv_ops->unlock_read(module);
 734                 return LDB_ERR_OPERATIONS_ERROR;
 735         }
 736
 737         ctx->tree = req->op.search.tree;
 738         ctx->scope = req->op.search.scope;
 739         ctx->base = req->op.search.base;
 740         ctx->attrs = req->op.search.attrs;
 741
 742         if ((req->op.search.base == NULL) || (ldb_dn_is_null(req->op.search.base) == true)) {
 743
 744                 /* Check what we should do with a NULL dn */
 745                 switch (req->op.search.scope) {
 746                 case LDB_SCOPE_BASE:
 747                         ldb_asprintf_errstring(ldb,
 748                                                "NULL Base DN invalid for a base search");
 749                         ret = LDB_ERR_INVALID_DN_SYNTAX;
 750                         break;
 751                 case LDB_SCOPE_ONELEVEL:
 752                         ldb_asprintf_errstring(ldb,
 753                                                "NULL Base DN invalid for a one-level search");
 754                         ret = LDB_ERR_INVALID_DN_SYNTAX;
 755                         break;
 756                 case LDB_SCOPE_SUBTREE:
 757                 default:
 758                         /* We accept subtree searches from a NULL base DN, ie over the whole DB */
 759                         ret = LDB_SUCCESS;
 760                 }
 761         } else if (req->op.search.scope == LDB_SCOPE_BASE) {
 762
 763                 /*
 764                  * If we are LDB_SCOPE_BASE, do just one search and
 765                  * return early.  This is critical to ensure we do not
 766                  * go into the index code for special DNs, as that
 767                  * will try to look up an index record for a special
 768                  * record (which doesn't exist).
 769                  */
 770                 ret = ldb_kv_search_and_return_base(ldb_kv, ctx);
 771
 772                 ldb_kv->kv_ops->unlock_read(module);
 773
 774                 return ret;
 775
 776         } else if (ldb_kv->check_base) {
 777                 /*
 778                  * This database has been marked as
 779                  * 'checkBaseOnSearch', so do a spot check of the base
 780                  * dn.  Also optimise the subsequent filter by filling
 781                  * in the ctx->base to be exactly case correct
 782                  */
 783                 ret = ldb_kv_search_base(
 784                     module, ctx, req->op.search.base, &ctx->base);
 785
 786                 if (ret == LDB_ERR_NO_SUCH_OBJECT) {
 787                         ldb_asprintf_errstring(ldb,
 788                                                "No such Base DN: %s",
 789                                                ldb_dn_get_linearized(req->op.search.base));
 790                 }
 791
 792         } else if (ldb_dn_validate(req->op.search.base) == false) {
 793
 794                 /* We don't want invalid base DNs here */
 795                 ldb_asprintf_errstring(ldb,
 796                                        "Invalid Base DN: %s",
 797                                        ldb_dn_get_linearized(req->op.search.base));
 798                 ret = LDB_ERR_INVALID_DN_SYNTAX;
 799
 800         } else {
 801                 /* If we are not checking the base DN life is easy */
 802                 ret = LDB_SUCCESS;
 803         }
 804
 805         if (ret == LDB_SUCCESS) {
 806                 uint32_t match_count = 0;
 807
 808                 ret = ldb_kv_search_indexed(ctx, &match_count);
 809                 if (ret == LDB_ERR_NO_SUCH_OBJECT) {
 810                         /* Not in the index, therefore OK! */
 811                         ret = LDB_SUCCESS;
 812
 813                 }
 814                 /* Check if we got just a normal error.
 815                  * In that case proceed to a full search unless we got a
 816                  * callback error */
 817                 if (!ctx->request_terminated && ret != LDB_SUCCESS) {
 818                         /* Not indexed, so we need to do a full scan */
 819                         if (ldb_kv->warn_unindexed ||
 820                             ldb_kv->disable_full_db_scan) {
 821                                 /* useful for debugging when slow performance
 822                                  * is caused by unindexed searches */
 823                                 char *expression = ldb_filter_from_tree(ctx, ctx->tree);
 824                                 ldb_debug(ldb, LDB_DEBUG_ERROR, "ldb FULL SEARCH: %s SCOPE: %s DN: %s",
 825                                                         expression,
 826                                                         req->op.search.scope==LDB_SCOPE_BASE?"base":
 827                                                         req->op.search.scope==LDB_SCOPE_ONELEVEL?"one":
 828                                                         req->op.search.scope==LDB_SCOPE_SUBTREE?"sub":"UNKNOWN",
 829                                                         ldb_dn_get_linearized(req->op.search.base));
 830
 831                                 talloc_free(expression);
 832                         }
 833
 834                         if (match_count != 0) {
 835                                 /* the indexing code gave an error
 836                                  * after having returned at least one
 837                                  * entry. This means the indexes are
 838                                  * corrupt or a database record is
 839                                  * corrupt. We cannot continue with a
 840                                  * full search or we may return
 841                                  * duplicate entries
 842                                  */
 843                                 ldb_kv->kv_ops->unlock_read(module);
 844                                 return LDB_ERR_OPERATIONS_ERROR;
 845                         }
 846
 847                         if (ldb_kv->disable_full_db_scan) {
 848                                 ldb_set_errstring(ldb,
 849                                                   "ldb FULL SEARCH disabled");
 850                                 ldb_kv->kv_ops->unlock_read(module);
 851                                 return LDB_ERR_INAPPROPRIATE_MATCHING;
 852                         }
 853
 854                         ret = ldb_kv_search_full(ctx);
 855                         if (ret != LDB_SUCCESS) {
 856                                 ldb_set_errstring(ldb, "Indexed and full searches both failed!\n");
 857                         }
 858                 }
 859         }
 860
 861         ldb_kv->kv_ops->unlock_read(module);
 862
 863         return ret;
 864 }