ldb: Change some prototypes to using ldb_val instead of TDB_DATA
[sfrench/samba-autobuild/.git] / lib / ldb / ldb_tdb / ldb_tdb.c
1 /*
2    ldb database library
3
4    Copyright (C) Andrew Tridgell 2004
5    Copyright (C) Stefan Metzmacher 2004
6    Copyright (C) Simo Sorce 2006-2008
7    Copyright (C) Matthias Dieter Wallnöfer 2009-2010
8
9      ** NOTE! The following LGPL license applies to the ldb
10      ** library. This does NOT imply that all of Samba is released
11      ** under the LGPL
12
13    This library is free software; you can redistribute it and/or
14    modify it under the terms of the GNU Lesser General Public
15    License as published by the Free Software Foundation; either
16    version 3 of the License, or (at your option) any later version.
17
18    This library is distributed in the hope that it will be useful,
19    but WITHOUT ANY WARRANTY; without even the implied warranty of
20    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
21    Lesser General Public License for more details.
22
23    You should have received a copy of the GNU Lesser General Public
24    License along with this library; if not, see <http://www.gnu.org/licenses/>.
25 */
26
27 /*
28  *  Name: ldb_tdb
29  *
30  *  Component: ldb tdb backend
31  *
32  *  Description: core functions for tdb backend
33  *
34  *  Author: Andrew Tridgell
35  *  Author: Stefan Metzmacher
36  *
37  *  Modifications:
38  *
39  *  - description: make the module use asynchronous calls
40  *    date: Feb 2006
41  *    Author: Simo Sorce
42  *
43  *  - description: make it possible to use event contexts
44  *    date: Jan 2008
45  *    Author: Simo Sorce
46  *
47  *  - description: fix up memory leaks and small bugs
48  *    date: Oct 2009
49  *    Author: Matthias Dieter Wallnöfer
50  */
51
52 #include "ldb_tdb.h"
53 #include "ldb_private.h"
54 #include <tdb.h>
55
56 /*
57   prevent memory errors on callbacks
58 */
59 struct ltdb_req_spy {
60         struct ltdb_context *ctx;
61 };
62
63 /*
64   map a tdb error code to a ldb error code
65 */
66 int ltdb_err_map(enum TDB_ERROR tdb_code)
67 {
68         switch (tdb_code) {
69         case TDB_SUCCESS:
70                 return LDB_SUCCESS;
71         case TDB_ERR_CORRUPT:
72         case TDB_ERR_OOM:
73         case TDB_ERR_EINVAL:
74                 return LDB_ERR_OPERATIONS_ERROR;
75         case TDB_ERR_IO:
76                 return LDB_ERR_PROTOCOL_ERROR;
77         case TDB_ERR_LOCK:
78         case TDB_ERR_NOLOCK:
79                 return LDB_ERR_BUSY;
80         case TDB_ERR_LOCK_TIMEOUT:
81                 return LDB_ERR_TIME_LIMIT_EXCEEDED;
82         case TDB_ERR_EXISTS:
83                 return LDB_ERR_ENTRY_ALREADY_EXISTS;
84         case TDB_ERR_NOEXIST:
85                 return LDB_ERR_NO_SUCH_OBJECT;
86         case TDB_ERR_RDONLY:
87                 return LDB_ERR_INSUFFICIENT_ACCESS_RIGHTS;
88         default:
89                 break;
90         }
91         return LDB_ERR_OTHER;
92 }
93
94 /*
95   lock the database for read - use by ltdb_search and ltdb_sequence_number
96 */
97 static int ltdb_lock_read(struct ldb_module *module)
98 {
99         void *data = ldb_module_get_private(module);
100         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
101         int tdb_ret = 0;
102         int ret;
103
104         if (ltdb->in_transaction == 0 &&
105             ltdb->read_lock_count == 0) {
106                 tdb_ret = tdb_lockall_read(ltdb->tdb);
107         }
108         if (tdb_ret == 0) {
109                 ltdb->read_lock_count++;
110                 return LDB_SUCCESS;
111         }
112         ret = ltdb_err_map(tdb_error(ltdb->tdb));
113         if (ret == LDB_SUCCESS) {
114                 ret = LDB_ERR_OPERATIONS_ERROR;
115         }
116         ldb_debug_set(ldb_module_get_ctx(module),
117                       LDB_DEBUG_FATAL,
118                       "Failure during ltdb_lock_read(): %s -> %s",
119                       tdb_errorstr(ltdb->tdb),
120                       ldb_strerror(ret));
121         return ret;
122 }
123
124 /*
125   unlock the database after a ltdb_lock_read()
126 */
127 static int ltdb_unlock_read(struct ldb_module *module)
128 {
129         void *data = ldb_module_get_private(module);
130         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
131         if (ltdb->in_transaction == 0 && ltdb->read_lock_count == 1) {
132                 tdb_unlockall_read(ltdb->tdb);
133                 ltdb->read_lock_count--;
134                 return 0;
135         }
136         ltdb->read_lock_count--;
137         return 0;
138 }
139
140
141 /* 
142  * Determine if this key could hold a record.  We allow the new GUID
143  * index, the old DN index and a possible future ID=
144  */
145 bool ltdb_key_is_record(TDB_DATA key)
146 {
147         if (key.dsize < 4) {
148                 return false;
149         }
150
151         if (memcmp(key.dptr, "DN=", 3) == 0) {
152                 return true;
153         }
154         
155         if (memcmp(key.dptr, "ID=", 3) == 0) {
156                 return true;
157         }
158
159         if (key.dsize < sizeof(LTDB_GUID_KEY_PREFIX)) {
160                 return false;
161         }
162
163         if (memcmp(key.dptr, LTDB_GUID_KEY_PREFIX,
164                    sizeof(LTDB_GUID_KEY_PREFIX) - 1) == 0) {
165                 return true;
166         }
167         
168         return false;
169 }
170
171 /*
172   form a TDB_DATA for a record key
173   caller frees
174
175   note that the key for a record can depend on whether the
176   dn refers to a case sensitive index record or not
177 */
178 TDB_DATA ltdb_key_dn(struct ldb_module *module, TALLOC_CTX *mem_ctx,
179                      struct ldb_dn *dn)
180 {
181         TDB_DATA key;
182         char *key_str = NULL;
183         const char *dn_folded = NULL;
184
185         /*
186           most DNs are case insensitive. The exception is index DNs for
187           case sensitive attributes
188
189           there are 3 cases dealt with in this code:
190
191           1) if the dn doesn't start with @ then uppercase the attribute
192              names and the attributes values of case insensitive attributes
193           2) if the dn starts with @ then leave it alone -
194              the indexing code handles the rest
195         */
196
197         dn_folded = ldb_dn_get_casefold(dn);
198         if (!dn_folded) {
199                 goto failed;
200         }
201
202         key_str = talloc_strdup(mem_ctx, "DN=");
203         if (!key_str) {
204                 goto failed;
205         }
206
207         key_str = talloc_strdup_append_buffer(key_str, dn_folded);
208         if (!key_str) {
209                 goto failed;
210         }
211
212         key.dptr = (uint8_t *)key_str;
213         key.dsize = strlen(key_str) + 1;
214
215         return key;
216
217 failed:
218         errno = ENOMEM;
219         key.dptr = NULL;
220         key.dsize = 0;
221         return key;
222 }
223
224 /* The caller is to provide a correctly sized key */
225 int ltdb_guid_to_key(struct ldb_module *module,
226                      struct ltdb_private *ltdb,
227                      const struct ldb_val *GUID_val,
228                      TDB_DATA *key)
229 {
230         const char *GUID_prefix = LTDB_GUID_KEY_PREFIX;
231         const int GUID_prefix_len = sizeof(LTDB_GUID_KEY_PREFIX) - 1;
232
233         if (key->dsize != (GUID_val->length+GUID_prefix_len)) {
234                 return LDB_ERR_OPERATIONS_ERROR;
235         }
236
237         memcpy(key->dptr, GUID_prefix, GUID_prefix_len);
238         memcpy(&key->dptr[GUID_prefix_len],
239                GUID_val->data, GUID_val->length);
240         return LDB_SUCCESS;
241 }
242
243 /*
244  * The caller is to provide a correctly sized key, used only in
245  * the GUID index mode
246  */
247 int ltdb_idx_to_key(struct ldb_module *module,
248                     struct ltdb_private *ltdb,
249                     TALLOC_CTX *mem_ctx,
250                     const struct ldb_val *idx_val,
251                     TDB_DATA *key)
252 {
253         struct ldb_context *ldb = ldb_module_get_ctx(module);
254         struct ldb_dn *dn;
255
256         if (ltdb->cache->GUID_index_attribute != NULL) {
257                 return ltdb_guid_to_key(module, ltdb,
258                                         idx_val, key);
259         }
260
261         dn = ldb_dn_from_ldb_val(mem_ctx, ldb, idx_val);
262         if (dn == NULL) {
263                 /*
264                  * LDB_ERR_INVALID_DN_SYNTAX would just be confusing
265                  * to the caller, as this in an invalid index value
266                  */
267                 return LDB_ERR_OPERATIONS_ERROR;
268         }
269         /* form the key */
270         *key = ltdb_key_dn(module, mem_ctx, dn);
271         TALLOC_FREE(dn);
272         if (!key->dptr) {
273                 return ldb_module_oom(module);
274         }
275         return LDB_SUCCESS;
276 }
277
278 /*
279   form a TDB_DATA for a record key
280   caller frees mem_ctx, which may or may not have the key
281   as a child.
282
283   note that the key for a record can depend on whether a
284   GUID index is in use, or the DN is used as the key
285 */
286 TDB_DATA ltdb_key_msg(struct ldb_module *module, TALLOC_CTX *mem_ctx,
287                       const struct ldb_message *msg)
288 {
289         void *data = ldb_module_get_private(module);
290         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
291         TDB_DATA key;
292         const struct ldb_val *guid_val;
293         int ret;
294
295         if (ltdb->cache->GUID_index_attribute == NULL) {
296                 return ltdb_key_dn(module, mem_ctx, msg->dn);
297         }
298
299         if (ldb_dn_is_special(msg->dn)) {
300                 return ltdb_key_dn(module, mem_ctx, msg->dn);
301         }
302
303         guid_val = ldb_msg_find_ldb_val(msg,
304                                        ltdb->cache->GUID_index_attribute);
305         if (guid_val == NULL) {
306                 ldb_asprintf_errstring(ldb_module_get_ctx(module),
307                                        "Did not find GUID attribute %s "
308                                        "in %s, required for TDB record "
309                                        "key in " LTDB_IDXGUID " mode.",
310                                        ltdb->cache->GUID_index_attribute,
311                                        ldb_dn_get_linearized(msg->dn));
312                 errno = EINVAL;
313                 key.dptr = NULL;
314                 key.dsize = 0;
315                 return key;
316         }
317
318         /* In this case, allocate with talloc */
319         key.dptr = talloc_size(mem_ctx, LTDB_GUID_KEY_SIZE);
320         if (key.dptr == NULL) {
321                 errno = ENOMEM;
322                 key.dptr = NULL;
323                 key.dsize = 0;
324                 return key;
325         }
326         key.dsize = talloc_get_size(key.dptr);
327
328         ret = ltdb_guid_to_key(module, ltdb, guid_val, &key);
329
330         if (ret != LDB_SUCCESS) {
331                 errno = EINVAL;
332                 key.dptr = NULL;
333                 key.dsize = 0;
334                 return key;
335         }
336         return key;
337 }
338
339 /*
340   check special dn's have valid attributes
341   currently only @ATTRIBUTES is checked
342 */
343 static int ltdb_check_special_dn(struct ldb_module *module,
344                                  const struct ldb_message *msg)
345 {
346         struct ldb_context *ldb = ldb_module_get_ctx(module);
347         unsigned int i, j;
348
349         if (! ldb_dn_is_special(msg->dn) ||
350             ! ldb_dn_check_special(msg->dn, LTDB_ATTRIBUTES)) {
351                 return LDB_SUCCESS;
352         }
353
354         /* we have @ATTRIBUTES, let's check attributes are fine */
355         /* should we check that we deny multivalued attributes ? */
356         for (i = 0; i < msg->num_elements; i++) {
357                 if (ldb_attr_cmp(msg->elements[i].name, "distinguishedName") == 0) continue;
358
359                 for (j = 0; j < msg->elements[i].num_values; j++) {
360                         if (ltdb_check_at_attributes_values(&msg->elements[i].values[j]) != 0) {
361                                 ldb_set_errstring(ldb, "Invalid attribute value in an @ATTRIBUTES entry");
362                                 return LDB_ERR_INVALID_ATTRIBUTE_SYNTAX;
363                         }
364                 }
365         }
366
367         return LDB_SUCCESS;
368 }
369
370
371 /*
372   we've made a modification to a dn - possibly reindex and
373   update sequence number
374 */
375 static int ltdb_modified(struct ldb_module *module, struct ldb_dn *dn)
376 {
377         int ret = LDB_SUCCESS;
378         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(module), struct ltdb_private);
379
380         /* only allow modifies inside a transaction, otherwise the
381          * ldb is unsafe */
382         if (ltdb->in_transaction == 0) {
383                 ldb_set_errstring(ldb_module_get_ctx(module), "ltdb modify without transaction");
384                 return LDB_ERR_OPERATIONS_ERROR;
385         }
386
387         if (ldb_dn_is_special(dn) &&
388             (ldb_dn_check_special(dn, LTDB_INDEXLIST) ||
389              ldb_dn_check_special(dn, LTDB_ATTRIBUTES)) )
390         {
391                 if (ltdb->warn_reindex) {
392                         ldb_debug(ldb_module_get_ctx(module),
393                                 LDB_DEBUG_ERROR, "Reindexing %s due to modification on %s",
394                                 ltdb->kv_ops->name(ltdb), ldb_dn_get_linearized(dn));
395                 }
396                 ret = ltdb_reindex(module);
397         }
398
399         /* If the modify was to a normal record, or any special except @BASEINFO, update the seq number */
400         if (ret == LDB_SUCCESS &&
401             !(ldb_dn_is_special(dn) &&
402               ldb_dn_check_special(dn, LTDB_BASEINFO)) ) {
403                 ret = ltdb_increase_sequence_number(module);
404         }
405
406         /* If the modify was to @OPTIONS, reload the cache */
407         if (ret == LDB_SUCCESS &&
408             ldb_dn_is_special(dn) &&
409             (ldb_dn_check_special(dn, LTDB_OPTIONS)) ) {
410                 ret = ltdb_cache_reload(module);
411         }
412
413         if (ret != LDB_SUCCESS) {
414                 ltdb->reindex_failed = true;
415         }
416
417         return ret;
418 }
419
420 static int ltdb_tdb_store(struct ltdb_private *ltdb, struct ldb_val ldb_key,
421                           struct ldb_val ldb_data, int flags)
422 {
423         TDB_DATA key = {
424                 .dptr = ldb_key.data,
425                 .dsize = ldb_key.length
426         };
427         TDB_DATA data = {
428                 .dptr = ldb_data.data,
429                 .dsize = ldb_data.length
430         };
431         return tdb_store(ltdb->tdb, key, data, flags);
432 }
433
434 static int ltdb_error(struct ltdb_private *ltdb)
435 {
436         return ltdb_err_map(tdb_error(ltdb->tdb));
437 }
438
439 static const char *ltdb_errorstr(struct ltdb_private *ltdb)
440 {
441         return tdb_errorstr(ltdb->tdb);
442 }
443
444 /*
445   store a record into the db
446 */
447 int ltdb_store(struct ldb_module *module, const struct ldb_message *msg, int flgs)
448 {
449         void *data = ldb_module_get_private(module);
450         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
451         TDB_DATA tdb_key;
452         struct ldb_val ldb_key;
453         struct ldb_val ldb_data;
454         int ret = LDB_SUCCESS;
455         TALLOC_CTX *tdb_key_ctx = talloc_new(module);
456
457         if (tdb_key_ctx == NULL) {
458                 return ldb_module_oom(module);
459         }
460
461         if (ltdb->read_only) {
462                 return LDB_ERR_UNWILLING_TO_PERFORM;
463         }
464
465         tdb_key = ltdb_key_msg(module, tdb_key_ctx, msg);
466         if (tdb_key.dptr == NULL) {
467                 TALLOC_FREE(tdb_key_ctx);
468                 return LDB_ERR_OTHER;
469         }
470
471         ret = ldb_pack_data(ldb_module_get_ctx(module),
472                             msg, &ldb_data);
473         if (ret == -1) {
474                 TALLOC_FREE(tdb_key_ctx);
475                 return LDB_ERR_OTHER;
476         }
477
478         ldb_key.data = tdb_key.dptr;
479         ldb_key.length = tdb_key.dsize;
480
481         ret = ltdb->kv_ops->store(ltdb, ldb_key, ldb_data, flgs);
482         if (ret != 0) {
483                 bool is_special = ldb_dn_is_special(msg->dn);
484                 ret = ltdb->kv_ops->error(ltdb);
485
486                 /*
487                  * LDB_ERR_ENTRY_ALREADY_EXISTS means the DN, not
488                  * the GUID, so re-map
489                  */
490                 if (ret == LDB_ERR_ENTRY_ALREADY_EXISTS
491                     && !is_special
492                     && ltdb->cache->GUID_index_attribute != NULL) {
493                         ret = LDB_ERR_CONSTRAINT_VIOLATION;
494                 }
495                 goto done;
496         }
497
498 done:
499         TALLOC_FREE(tdb_key_ctx);
500         talloc_free(ldb_data.data);
501
502         return ret;
503 }
504
505
506 /*
507   check if a attribute is a single valued, for a given element
508  */
509 static bool ldb_tdb_single_valued(const struct ldb_schema_attribute *a,
510                                   struct ldb_message_element *el)
511 {
512         if (!a) return false;
513         if (el != NULL) {
514                 if (el->flags & LDB_FLAG_INTERNAL_FORCE_SINGLE_VALUE_CHECK) {
515                         /* override from a ldb module, for example
516                            used for the description field, which is
517                            marked multi-valued in the schema but which
518                            should not actually accept multiple
519                            values */
520                         return true;
521                 }
522                 if (el->flags & LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK) {
523                         /* override from a ldb module, for example used for
524                            deleted linked attribute entries */
525                         return false;
526                 }
527         }
528         if (a->flags & LDB_ATTR_FLAG_SINGLE_VALUE) {
529                 return true;
530         }
531         return false;
532 }
533
534 static int ltdb_add_internal(struct ldb_module *module,
535                              struct ltdb_private *ltdb,
536                              const struct ldb_message *msg,
537                              bool check_single_value)
538 {
539         struct ldb_context *ldb = ldb_module_get_ctx(module);
540         int ret = LDB_SUCCESS;
541         unsigned int i;
542
543         for (i=0;i<msg->num_elements;i++) {
544                 struct ldb_message_element *el = &msg->elements[i];
545                 const struct ldb_schema_attribute *a = ldb_schema_attribute_by_name(ldb, el->name);
546
547                 if (el->num_values == 0) {
548                         ldb_asprintf_errstring(ldb, "attribute '%s' on '%s' specified, but with 0 values (illegal)",
549                                                el->name, ldb_dn_get_linearized(msg->dn));
550                         return LDB_ERR_CONSTRAINT_VIOLATION;
551                 }
552                 if (check_single_value &&
553                                 el->num_values > 1 &&
554                                 ldb_tdb_single_valued(a, el)) {
555                         ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
556                                                el->name, ldb_dn_get_linearized(msg->dn));
557                         return LDB_ERR_CONSTRAINT_VIOLATION;
558                 }
559
560                 /* Do not check "@ATTRIBUTES" for duplicated values */
561                 if (ldb_dn_is_special(msg->dn) &&
562                     ldb_dn_check_special(msg->dn, LTDB_ATTRIBUTES)) {
563                         continue;
564                 }
565
566                 if (check_single_value &&
567                     !(el->flags &
568                       LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK)) {
569                         struct ldb_val *duplicate = NULL;
570
571                         ret = ldb_msg_find_duplicate_val(ldb, discard_const(msg),
572                                                          el, &duplicate, 0);
573                         if (ret != LDB_SUCCESS) {
574                                 return ret;
575                         }
576                         if (duplicate != NULL) {
577                                 ldb_asprintf_errstring(
578                                         ldb,
579                                         "attribute '%s': value '%.*s' on '%s' "
580                                         "provided more than once in ADD object",
581                                         el->name,
582                                         (int)duplicate->length,
583                                         duplicate->data,
584                                         ldb_dn_get_linearized(msg->dn));
585                                 return LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
586                         }
587                 }
588         }
589
590         ret = ltdb_store(module, msg, TDB_INSERT);
591         if (ret != LDB_SUCCESS) {
592                 /*
593                  * Try really hard to get the right error code for
594                  * a re-add situation, as this can matter!
595                  */
596                 if (ret == LDB_ERR_CONSTRAINT_VIOLATION) {
597                         int ret2;
598                         struct ldb_dn *dn2 = NULL;
599                         TALLOC_CTX *mem_ctx = talloc_new(module);
600                         if (mem_ctx == NULL) {
601                                 return ldb_module_operr(module);
602                         }
603                         ret2 = ltdb_search_base(module, module,
604                                                 msg->dn, &dn2);
605                         TALLOC_FREE(mem_ctx);
606                         if (ret2 == LDB_SUCCESS) {
607                                 ret = LDB_ERR_ENTRY_ALREADY_EXISTS;
608                         }
609                 }
610                 if (ret == LDB_ERR_ENTRY_ALREADY_EXISTS) {
611                         ldb_asprintf_errstring(ldb,
612                                                "Entry %s already exists",
613                                                ldb_dn_get_linearized(msg->dn));
614                 }
615                 return ret;
616         }
617
618         ret = ltdb_index_add_new(module, ltdb, msg);
619         if (ret != LDB_SUCCESS) {
620                 /*
621                  * If we failed to index, delete the message again.
622                  *
623                  * This is particularly important for the GUID index
624                  * case, which will only fail for a duplicate DN
625                  * in the index add.
626                  *
627                  * Note that the caller may not cancel the transation
628                  * and this means the above add might really show up!
629                  */
630                 ltdb_delete_noindex(module, msg);
631                 return ret;
632         }
633
634         ret = ltdb_modified(module, msg->dn);
635
636         return ret;
637 }
638
639 /*
640   add a record to the database
641 */
642 static int ltdb_add(struct ltdb_context *ctx)
643 {
644         struct ldb_module *module = ctx->module;
645         struct ldb_request *req = ctx->req;
646         void *data = ldb_module_get_private(module);
647         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
648         int ret = LDB_SUCCESS;
649
650         ret = ltdb_check_special_dn(module, req->op.add.message);
651         if (ret != LDB_SUCCESS) {
652                 return ret;
653         }
654
655         ldb_request_set_state(req, LDB_ASYNC_PENDING);
656
657         if (ltdb_cache_load(module) != 0) {
658                 return LDB_ERR_OPERATIONS_ERROR;
659         }
660
661         ret = ltdb_add_internal(module, ltdb,
662                                 req->op.add.message, true);
663
664         return ret;
665 }
666
667 static int ltdb_tdb_delete(struct ltdb_private *ltdb, struct ldb_val ldb_key)
668 {
669         TDB_DATA tdb_key = {
670                 .dptr = ldb_key.data,
671                 .dsize = ldb_key.length
672         };
673         return tdb_delete(ltdb->tdb, tdb_key);
674 }
675
676 /*
677   delete a record from the database, not updating indexes (used for deleting
678   index records)
679 */
680 int ltdb_delete_noindex(struct ldb_module *module,
681                         const struct ldb_message *msg)
682 {
683         void *data = ldb_module_get_private(module);
684         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
685         struct ldb_val ldb_key;
686         TDB_DATA tdb_key;
687         int ret;
688         TALLOC_CTX *tdb_key_ctx = talloc_new(module);
689
690         if (tdb_key_ctx == NULL) {
691                 return ldb_module_oom(module);
692         }
693
694         if (ltdb->read_only) {
695                 return LDB_ERR_UNWILLING_TO_PERFORM;
696         }
697
698         tdb_key = ltdb_key_msg(module, tdb_key_ctx, msg);
699         if (!tdb_key.dptr) {
700                 TALLOC_FREE(tdb_key_ctx);
701                 return LDB_ERR_OTHER;
702         }
703
704         ldb_key.data = tdb_key.dptr;
705         ldb_key.length = tdb_key.dsize;
706
707         ret = ltdb->kv_ops->delete(ltdb, ldb_key);
708         TALLOC_FREE(tdb_key_ctx);
709
710         if (ret != 0) {
711                 ret = ltdb->kv_ops->error(ltdb);
712         }
713
714         return ret;
715 }
716
717 static int ltdb_delete_internal(struct ldb_module *module, struct ldb_dn *dn)
718 {
719         struct ldb_message *msg;
720         int ret = LDB_SUCCESS;
721
722         msg = ldb_msg_new(module);
723         if (msg == NULL) {
724                 return LDB_ERR_OPERATIONS_ERROR;
725         }
726
727         /* in case any attribute of the message was indexed, we need
728            to fetch the old record */
729         ret = ltdb_search_dn1(module, dn, msg, LDB_UNPACK_DATA_FLAG_NO_DATA_ALLOC);
730         if (ret != LDB_SUCCESS) {
731                 /* not finding the old record is an error */
732                 goto done;
733         }
734
735         ret = ltdb_delete_noindex(module, msg);
736         if (ret != LDB_SUCCESS) {
737                 goto done;
738         }
739
740         /* remove any indexed attributes */
741         ret = ltdb_index_delete(module, msg);
742         if (ret != LDB_SUCCESS) {
743                 goto done;
744         }
745
746         ret = ltdb_modified(module, dn);
747         if (ret != LDB_SUCCESS) {
748                 goto done;
749         }
750
751 done:
752         talloc_free(msg);
753         return ret;
754 }
755
756 /*
757   delete a record from the database
758 */
759 static int ltdb_delete(struct ltdb_context *ctx)
760 {
761         struct ldb_module *module = ctx->module;
762         struct ldb_request *req = ctx->req;
763         int ret = LDB_SUCCESS;
764
765         ldb_request_set_state(req, LDB_ASYNC_PENDING);
766
767         if (ltdb_cache_load(module) != 0) {
768                 return LDB_ERR_OPERATIONS_ERROR;
769         }
770
771         ret = ltdb_delete_internal(module, req->op.del.dn);
772
773         return ret;
774 }
775
776 /*
777   find an element by attribute name. At the moment this does a linear search,
778   it should be re-coded to use a binary search once all places that modify
779   records guarantee sorted order
780
781   return the index of the first matching element if found, otherwise -1
782 */
783 static int find_element(const struct ldb_message *msg, const char *name)
784 {
785         unsigned int i;
786         for (i=0;i<msg->num_elements;i++) {
787                 if (ldb_attr_cmp(msg->elements[i].name, name) == 0) {
788                         return i;
789                 }
790         }
791         return -1;
792 }
793
794
795 /*
796   add an element to an existing record. Assumes a elements array that we
797   can call re-alloc on, and assumed that we can re-use the data pointers from
798   the passed in additional values. Use with care!
799
800   returns 0 on success, -1 on failure (and sets errno)
801 */
802 static int ltdb_msg_add_element(struct ldb_message *msg,
803                                 struct ldb_message_element *el)
804 {
805         struct ldb_message_element *e2;
806         unsigned int i;
807
808         if (el->num_values == 0) {
809                 /* nothing to do here - we don't add empty elements */
810                 return 0;
811         }
812
813         e2 = talloc_realloc(msg, msg->elements, struct ldb_message_element,
814                               msg->num_elements+1);
815         if (!e2) {
816                 errno = ENOMEM;
817                 return -1;
818         }
819
820         msg->elements = e2;
821
822         e2 = &msg->elements[msg->num_elements];
823
824         e2->name = el->name;
825         e2->flags = el->flags;
826         e2->values = talloc_array(msg->elements,
827                                   struct ldb_val, el->num_values);
828         if (!e2->values) {
829                 errno = ENOMEM;
830                 return -1;
831         }
832         for (i=0;i<el->num_values;i++) {
833                 e2->values[i] = el->values[i];
834         }
835         e2->num_values = el->num_values;
836
837         ++msg->num_elements;
838
839         return 0;
840 }
841
842 /*
843   delete all elements having a specified attribute name
844 */
845 static int msg_delete_attribute(struct ldb_module *module,
846                                 struct ltdb_private *ltdb,
847                                 struct ldb_message *msg, const char *name)
848 {
849         unsigned int i;
850         int ret;
851         struct ldb_message_element *el;
852         bool is_special = ldb_dn_is_special(msg->dn);
853
854         if (!is_special
855             && ltdb->cache->GUID_index_attribute != NULL
856             && ldb_attr_cmp(name, ltdb->cache->GUID_index_attribute) == 0) {
857                 struct ldb_context *ldb = ldb_module_get_ctx(module);
858                 ldb_asprintf_errstring(ldb, "Must not modify GUID "
859                                        "attribute %s (used as DB index)",
860                                        ltdb->cache->GUID_index_attribute);
861                 return LDB_ERR_CONSTRAINT_VIOLATION;
862         }
863
864         el = ldb_msg_find_element(msg, name);
865         if (el == NULL) {
866                 return LDB_ERR_NO_SUCH_ATTRIBUTE;
867         }
868         i = el - msg->elements;
869
870         ret = ltdb_index_del_element(module, ltdb, msg, el);
871         if (ret != LDB_SUCCESS) {
872                 return ret;
873         }
874
875         talloc_free(el->values);
876         if (msg->num_elements > (i+1)) {
877                 memmove(el, el+1, sizeof(*el) * (msg->num_elements - (i+1)));
878         }
879         msg->num_elements--;
880         msg->elements = talloc_realloc(msg, msg->elements,
881                                        struct ldb_message_element,
882                                        msg->num_elements);
883         return LDB_SUCCESS;
884 }
885
886 /*
887   delete all elements matching an attribute name/value
888
889   return LDB Error on failure
890 */
891 static int msg_delete_element(struct ldb_module *module,
892                               struct ltdb_private *ltdb,
893                               struct ldb_message *msg,
894                               const char *name,
895                               const struct ldb_val *val)
896 {
897         struct ldb_context *ldb = ldb_module_get_ctx(module);
898         unsigned int i;
899         int found, ret;
900         struct ldb_message_element *el;
901         const struct ldb_schema_attribute *a;
902
903         found = find_element(msg, name);
904         if (found == -1) {
905                 return LDB_ERR_NO_SUCH_ATTRIBUTE;
906         }
907
908         i = (unsigned int) found;
909         el = &(msg->elements[i]);
910
911         a = ldb_schema_attribute_by_name(ldb, el->name);
912
913         for (i=0;i<el->num_values;i++) {
914                 bool matched;
915                 if (a->syntax->operator_fn) {
916                         ret = a->syntax->operator_fn(ldb, LDB_OP_EQUALITY, a,
917                                                      &el->values[i], val, &matched);
918                         if (ret != LDB_SUCCESS) return ret;
919                 } else {
920                         matched = (a->syntax->comparison_fn(ldb, ldb,
921                                                             &el->values[i], val) == 0);
922                 }
923                 if (matched) {
924                         if (el->num_values == 1) {
925                                 return msg_delete_attribute(module,
926                                                             ltdb, msg, name);
927                         }
928
929                         ret = ltdb_index_del_value(module, ltdb, msg, el, i);
930                         if (ret != LDB_SUCCESS) {
931                                 return ret;
932                         }
933
934                         if (i<el->num_values-1) {
935                                 memmove(&el->values[i], &el->values[i+1],
936                                         sizeof(el->values[i])*
937                                                 (el->num_values-(i+1)));
938                         }
939                         el->num_values--;
940
941                         /* per definition we find in a canonicalised message an
942                            attribute value only once. So we are finished here */
943                         return LDB_SUCCESS;
944                 }
945         }
946
947         /* Not found */
948         return LDB_ERR_NO_SUCH_ATTRIBUTE;
949 }
950
951 /*
952   modify a record - internal interface
953
954   yuck - this is O(n^2). Luckily n is usually small so we probably
955   get away with it, but if we ever have really large attribute lists
956   then we'll need to look at this again
957
958   'req' is optional, and is used to specify controls if supplied
959 */
960 int ltdb_modify_internal(struct ldb_module *module,
961                          const struct ldb_message *msg,
962                          struct ldb_request *req)
963 {
964         struct ldb_context *ldb = ldb_module_get_ctx(module);
965         void *data = ldb_module_get_private(module);
966         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
967         struct ldb_message *msg2;
968         unsigned int i, j;
969         int ret = LDB_SUCCESS, idx;
970         struct ldb_control *control_permissive = NULL;
971         TALLOC_CTX *mem_ctx = talloc_new(req);
972
973         if (mem_ctx == NULL) {
974                 return ldb_module_oom(module);
975         }
976         
977         if (req) {
978                 control_permissive = ldb_request_get_control(req,
979                                         LDB_CONTROL_PERMISSIVE_MODIFY_OID);
980         }
981
982         msg2 = ldb_msg_new(mem_ctx);
983         if (msg2 == NULL) {
984                 ret = LDB_ERR_OTHER;
985                 goto done;
986         }
987
988         ret = ltdb_search_dn1(module, msg->dn,
989                               msg2,
990                               LDB_UNPACK_DATA_FLAG_NO_DATA_ALLOC);
991         if (ret != LDB_SUCCESS) {
992                 goto done;
993         }
994
995         for (i=0; i<msg->num_elements; i++) {
996                 struct ldb_message_element *el = &msg->elements[i], *el2;
997                 struct ldb_val *vals;
998                 const struct ldb_schema_attribute *a = ldb_schema_attribute_by_name(ldb, el->name);
999                 const char *dn;
1000                 uint32_t options = 0;
1001                 if (control_permissive != NULL) {
1002                         options |= LDB_MSG_FIND_COMMON_REMOVE_DUPLICATES;
1003                 }
1004
1005                 switch (msg->elements[i].flags & LDB_FLAG_MOD_MASK) {
1006                 case LDB_FLAG_MOD_ADD:
1007
1008                         if (el->num_values == 0) {
1009                                 ldb_asprintf_errstring(ldb,
1010                                                        "attribute '%s': attribute on '%s' specified, but with 0 values (illegal)",
1011                                                        el->name, ldb_dn_get_linearized(msg2->dn));
1012                                 ret = LDB_ERR_CONSTRAINT_VIOLATION;
1013                                 goto done;
1014                         }
1015
1016                         /* make a copy of the array so that a permissive
1017                          * control can remove duplicates without changing the
1018                          * original values, but do not copy data as we do not
1019                          * need to keep it around once the operation is
1020                          * finished */
1021                         if (control_permissive) {
1022                                 el = talloc(msg2, struct ldb_message_element);
1023                                 if (!el) {
1024                                         ret = LDB_ERR_OTHER;
1025                                         goto done;
1026                                 }
1027                                 *el = msg->elements[i];
1028                                 el->values = talloc_array(el, struct ldb_val, el->num_values);
1029                                 if (el->values == NULL) {
1030                                         ret = LDB_ERR_OTHER;
1031                                         goto done;
1032                                 }
1033                                 for (j = 0; j < el->num_values; j++) {
1034                                         el->values[j] = msg->elements[i].values[j];
1035                                 }
1036                         }
1037
1038                         if (el->num_values > 1 && ldb_tdb_single_valued(a, el)) {
1039                                 ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
1040                                                        el->name, ldb_dn_get_linearized(msg2->dn));
1041                                 ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
1042                                 goto done;
1043                         }
1044
1045                         /* Checks if element already exists */
1046                         idx = find_element(msg2, el->name);
1047                         if (idx == -1) {
1048                                 if (ltdb_msg_add_element(msg2, el) != 0) {
1049                                         ret = LDB_ERR_OTHER;
1050                                         goto done;
1051                                 }
1052                                 ret = ltdb_index_add_element(module, ltdb,
1053                                                              msg2,
1054                                                              el);
1055                                 if (ret != LDB_SUCCESS) {
1056                                         goto done;
1057                                 }
1058                         } else {
1059                                 j = (unsigned int) idx;
1060                                 el2 = &(msg2->elements[j]);
1061
1062                                 /* We cannot add another value on a existing one
1063                                    if the attribute is single-valued */
1064                                 if (ldb_tdb_single_valued(a, el)) {
1065                                         ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
1066                                                                el->name, ldb_dn_get_linearized(msg2->dn));
1067                                         ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
1068                                         goto done;
1069                                 }
1070
1071                                 /* Check that values don't exist yet on multi-
1072                                    valued attributes or aren't provided twice */
1073                                 if (!(el->flags &
1074                                       LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK)) {
1075                                         struct ldb_val *duplicate = NULL;
1076                                         ret = ldb_msg_find_common_values(ldb,
1077                                                                          msg2,
1078                                                                          el,
1079                                                                          el2,
1080                                                                          options);
1081
1082                                         if (ret ==
1083                                             LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS) {
1084                                                 ldb_asprintf_errstring(ldb,
1085                                                         "attribute '%s': value "
1086                                                         "#%u on '%s' already "
1087                                                         "exists", el->name, j,
1088                                                         ldb_dn_get_linearized(msg2->dn));
1089                                                 goto done;
1090                                         } else if (ret != LDB_SUCCESS) {
1091                                                 goto done;
1092                                         }
1093
1094                                         ret = ldb_msg_find_duplicate_val(
1095                                                 ldb, msg2, el, &duplicate, 0);
1096                                         if (ret != LDB_SUCCESS) {
1097                                                 goto done;
1098                                         }
1099                                         if (duplicate != NULL) {
1100                                                 ldb_asprintf_errstring(
1101                                                         ldb,
1102                                                         "attribute '%s': value "
1103                                                         "'%.*s' on '%s' "
1104                                                         "provided more than "
1105                                                         "once in ADD",
1106                                                         el->name,
1107                                                         (int)duplicate->length,
1108                                                         duplicate->data,
1109                                                         ldb_dn_get_linearized(msg->dn));
1110                                                 ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
1111                                                 goto done;
1112                                         }
1113                                 }
1114
1115                                 /* Now combine existing and new values to a new
1116                                    attribute record */
1117                                 vals = talloc_realloc(msg2->elements,
1118                                                       el2->values, struct ldb_val,
1119                                                       el2->num_values + el->num_values);
1120                                 if (vals == NULL) {
1121                                         ldb_oom(ldb);
1122                                         ret = LDB_ERR_OTHER;
1123                                         goto done;
1124                                 }
1125
1126                                 for (j=0; j<el->num_values; j++) {
1127                                         vals[el2->num_values + j] =
1128                                                 ldb_val_dup(vals, &el->values[j]);
1129                                 }
1130
1131                                 el2->values = vals;
1132                                 el2->num_values += el->num_values;
1133
1134                                 ret = ltdb_index_add_element(module, ltdb,
1135                                                              msg2, el);
1136                                 if (ret != LDB_SUCCESS) {
1137                                         goto done;
1138                                 }
1139                         }
1140
1141                         break;
1142
1143                 case LDB_FLAG_MOD_REPLACE:
1144
1145                         if (el->num_values > 1 && ldb_tdb_single_valued(a, el)) {
1146                                 ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
1147                                                        el->name, ldb_dn_get_linearized(msg2->dn));
1148                                 ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
1149                                 goto done;
1150                         }
1151
1152                         /*
1153                          * We don't need to check this if we have been
1154                          * pre-screened by the repl_meta_data module
1155                          * in Samba, or someone else who can claim to
1156                          * know what they are doing. 
1157                          */
1158                         if (!(el->flags & LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK)) {
1159                                 struct ldb_val *duplicate = NULL;
1160
1161                                 ret = ldb_msg_find_duplicate_val(ldb, msg2, el,
1162                                                                  &duplicate, 0);
1163                                 if (ret != LDB_SUCCESS) {
1164                                         goto done;
1165                                 }
1166                                 if (duplicate != NULL) {
1167                                         ldb_asprintf_errstring(
1168                                                 ldb,
1169                                                 "attribute '%s': value '%.*s' "
1170                                                 "on '%s' provided more than "
1171                                                 "once in REPLACE",
1172                                                 el->name,
1173                                                 (int)duplicate->length,
1174                                                 duplicate->data,
1175                                                 ldb_dn_get_linearized(msg2->dn));
1176                                         ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
1177                                         goto done;
1178                                 }
1179                         }
1180
1181                         /* Checks if element already exists */
1182                         idx = find_element(msg2, el->name);
1183                         if (idx != -1) {
1184                                 j = (unsigned int) idx;
1185                                 el2 = &(msg2->elements[j]);
1186
1187                                 /* we consider two elements to be
1188                                  * equal only if the order
1189                                  * matches. This allows dbcheck to
1190                                  * fix the ordering on attributes
1191                                  * where order matters, such as
1192                                  * objectClass
1193                                  */
1194                                 if (ldb_msg_element_equal_ordered(el, el2)) {
1195                                         continue;
1196                                 }
1197
1198                                 /* Delete the attribute if it exists in the DB */
1199                                 if (msg_delete_attribute(module, ltdb,
1200                                                          msg2,
1201                                                          el->name) != 0) {
1202                                         ret = LDB_ERR_OTHER;
1203                                         goto done;
1204                                 }
1205                         }
1206
1207                         /* Recreate it with the new values */
1208                         if (ltdb_msg_add_element(msg2, el) != 0) {
1209                                 ret = LDB_ERR_OTHER;
1210                                 goto done;
1211                         }
1212
1213                         ret = ltdb_index_add_element(module, ltdb,
1214                                                      msg2, el);
1215                         if (ret != LDB_SUCCESS) {
1216                                 goto done;
1217                         }
1218
1219                         break;
1220
1221                 case LDB_FLAG_MOD_DELETE:
1222                         dn = ldb_dn_get_linearized(msg2->dn);
1223                         if (dn == NULL) {
1224                                 ret = LDB_ERR_OTHER;
1225                                 goto done;
1226                         }
1227
1228                         if (msg->elements[i].num_values == 0) {
1229                                 /* Delete the whole attribute */
1230                                 ret = msg_delete_attribute(module,
1231                                                            ltdb,
1232                                                            msg2,
1233                                                            msg->elements[i].name);
1234                                 if (ret == LDB_ERR_NO_SUCH_ATTRIBUTE &&
1235                                     control_permissive) {
1236                                         ret = LDB_SUCCESS;
1237                                 } else {
1238                                         ldb_asprintf_errstring(ldb,
1239                                                                "attribute '%s': no such attribute for delete on '%s'",
1240                                                                msg->elements[i].name, dn);
1241                                 }
1242                                 if (ret != LDB_SUCCESS) {
1243                                         goto done;
1244                                 }
1245                         } else {
1246                                 /* Delete specified values from an attribute */
1247                                 for (j=0; j < msg->elements[i].num_values; j++) {
1248                                         ret = msg_delete_element(module,
1249                                                                  ltdb,
1250                                                                  msg2,
1251                                                                  msg->elements[i].name,
1252                                                                  &msg->elements[i].values[j]);
1253                                         if (ret == LDB_ERR_NO_SUCH_ATTRIBUTE &&
1254                                             control_permissive) {
1255                                                 ret = LDB_SUCCESS;
1256                                         } else if (ret == LDB_ERR_NO_SUCH_ATTRIBUTE) {
1257                                                 ldb_asprintf_errstring(ldb,
1258                                                                        "attribute '%s': no matching attribute value while deleting attribute on '%s'",
1259                                                                        msg->elements[i].name, dn);
1260                                         }
1261                                         if (ret != LDB_SUCCESS) {
1262                                                 goto done;
1263                                         }
1264                                 }
1265                         }
1266                         break;
1267                 default:
1268                         ldb_asprintf_errstring(ldb,
1269                                                "attribute '%s': invalid modify flags on '%s': 0x%x",
1270                                                msg->elements[i].name, ldb_dn_get_linearized(msg->dn),
1271                                                msg->elements[i].flags & LDB_FLAG_MOD_MASK);
1272                         ret = LDB_ERR_PROTOCOL_ERROR;
1273                         goto done;
1274                 }
1275         }
1276
1277         ret = ltdb_store(module, msg2, TDB_MODIFY);
1278         if (ret != LDB_SUCCESS) {
1279                 goto done;
1280         }
1281
1282         ret = ltdb_modified(module, msg2->dn);
1283         if (ret != LDB_SUCCESS) {
1284                 goto done;
1285         }
1286
1287 done:
1288         TALLOC_FREE(mem_ctx);
1289         return ret;
1290 }
1291
1292 /*
1293   modify a record
1294 */
1295 static int ltdb_modify(struct ltdb_context *ctx)
1296 {
1297         struct ldb_module *module = ctx->module;
1298         struct ldb_request *req = ctx->req;
1299         int ret = LDB_SUCCESS;
1300
1301         ret = ltdb_check_special_dn(module, req->op.mod.message);
1302         if (ret != LDB_SUCCESS) {
1303                 return ret;
1304         }
1305
1306         ldb_request_set_state(req, LDB_ASYNC_PENDING);
1307
1308         if (ltdb_cache_load(module) != 0) {
1309                 return LDB_ERR_OPERATIONS_ERROR;
1310         }
1311
1312         ret = ltdb_modify_internal(module, req->op.mod.message, req);
1313
1314         return ret;
1315 }
1316
1317 /*
1318   rename a record
1319 */
1320 static int ltdb_rename(struct ltdb_context *ctx)
1321 {
1322         struct ldb_module *module = ctx->module;
1323         void *data = ldb_module_get_private(module);
1324         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
1325         struct ldb_request *req = ctx->req;
1326         struct ldb_message *msg;
1327         int ret = LDB_SUCCESS;
1328         TDB_DATA tdb_key, tdb_key_old;
1329         struct ldb_dn *db_dn;
1330
1331         ldb_request_set_state(req, LDB_ASYNC_PENDING);
1332
1333         if (ltdb_cache_load(ctx->module) != 0) {
1334                 return LDB_ERR_OPERATIONS_ERROR;
1335         }
1336
1337         msg = ldb_msg_new(ctx);
1338         if (msg == NULL) {
1339                 return LDB_ERR_OPERATIONS_ERROR;
1340         }
1341
1342         /* we need to fetch the old record to re-add under the new name */
1343         ret = ltdb_search_dn1(module, req->op.rename.olddn, msg,
1344                               LDB_UNPACK_DATA_FLAG_NO_DATA_ALLOC);
1345         if (ret != LDB_SUCCESS) {
1346                 /* not finding the old record is an error */
1347                 return ret;
1348         }
1349
1350         /* We need to, before changing the DB, check if the new DN
1351          * exists, so we can return this error to the caller with an
1352          * unmodified DB
1353          *
1354          * Even in GUID index mode we use ltdb_key_dn() as we are
1355          * trying to figure out if this is just a case rename
1356          */
1357         tdb_key = ltdb_key_dn(module, msg, req->op.rename.newdn);
1358         if (!tdb_key.dptr) {
1359                 talloc_free(msg);
1360                 return LDB_ERR_OPERATIONS_ERROR;
1361         }
1362
1363         tdb_key_old = ltdb_key_dn(module, msg, req->op.rename.olddn);
1364         if (!tdb_key_old.dptr) {
1365                 talloc_free(msg);
1366                 talloc_free(tdb_key.dptr);
1367                 return LDB_ERR_OPERATIONS_ERROR;
1368         }
1369
1370         /*
1371          * Only declare a conflict if the new DN already exists,
1372          * and it isn't a case change on the old DN
1373          */
1374         if (tdb_key_old.dsize != tdb_key.dsize
1375             || memcmp(tdb_key.dptr, tdb_key_old.dptr, tdb_key.dsize) != 0) {
1376                 ret = ltdb_search_base(module, msg,
1377                                        req->op.rename.newdn,
1378                                        &db_dn);
1379                 if (ret == LDB_SUCCESS) {
1380                         ret = LDB_ERR_ENTRY_ALREADY_EXISTS;
1381                 } else if (ret == LDB_ERR_NO_SUCH_OBJECT) {
1382                         ret = LDB_SUCCESS;
1383                 }
1384         }
1385
1386         /* finding the new record already in the DB is an error */
1387
1388         if (ret == LDB_ERR_ENTRY_ALREADY_EXISTS) {
1389                 ldb_asprintf_errstring(ldb_module_get_ctx(module),
1390                                        "Entry %s already exists",
1391                                        ldb_dn_get_linearized(req->op.rename.newdn));
1392         }
1393         if (ret != LDB_SUCCESS) {
1394                 talloc_free(tdb_key_old.dptr);
1395                 talloc_free(tdb_key.dptr);
1396                 talloc_free(msg);
1397                 return ret;
1398         }
1399
1400         talloc_free(tdb_key_old.dptr);
1401         talloc_free(tdb_key.dptr);
1402
1403         /* Always delete first then add, to avoid conflicts with
1404          * unique indexes. We rely on the transaction to make this
1405          * atomic
1406          */
1407         ret = ltdb_delete_internal(module, msg->dn);
1408         if (ret != LDB_SUCCESS) {
1409                 talloc_free(msg);
1410                 return ret;
1411         }
1412
1413         msg->dn = ldb_dn_copy(msg, req->op.rename.newdn);
1414         if (msg->dn == NULL) {
1415                 talloc_free(msg);
1416                 return LDB_ERR_OPERATIONS_ERROR;
1417         }
1418
1419         /* We don't check single value as we can have more than 1 with
1420          * deleted attributes. We could go through all elements but that's
1421          * maybe not the most efficient way
1422          */
1423         ret = ltdb_add_internal(module, ltdb, msg, false);
1424
1425         talloc_free(msg);
1426
1427         return ret;
1428 }
1429
1430 static int ltdb_tdb_transaction_start(struct ltdb_private *ltdb)
1431 {
1432         return tdb_transaction_start(ltdb->tdb);
1433 }
1434
1435 static int ltdb_tdb_transaction_cancel(struct ltdb_private *ltdb)
1436 {
1437         return tdb_transaction_cancel(ltdb->tdb);
1438 }
1439
1440 static int ltdb_tdb_transaction_prepare_commit(struct ltdb_private *ltdb)
1441 {
1442         return tdb_transaction_prepare_commit(ltdb->tdb);
1443 }
1444
1445 static int ltdb_tdb_transaction_commit(struct ltdb_private *ltdb)
1446 {
1447         return tdb_transaction_commit(ltdb->tdb);
1448 }
1449
1450 static int ltdb_start_trans(struct ldb_module *module)
1451 {
1452         void *data = ldb_module_get_private(module);
1453         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
1454
1455         /* Do not take out the transaction lock on a read-only DB */
1456         if (ltdb->read_only) {
1457                 return LDB_ERR_UNWILLING_TO_PERFORM;
1458         }
1459
1460         if (ltdb->kv_ops->begin_write(ltdb) != 0) {
1461                 return ltdb->kv_ops->error(ltdb);
1462         }
1463
1464         ltdb->in_transaction++;
1465
1466         ltdb_index_transaction_start(module);
1467
1468         ltdb->reindex_failed = false;
1469
1470         return LDB_SUCCESS;
1471 }
1472
1473 /*
1474  * Forward declaration to allow prepare_commit to in fact abort the
1475  * transaction
1476  */
1477 static int ltdb_del_trans(struct ldb_module *module);
1478
1479 static int ltdb_prepare_commit(struct ldb_module *module)
1480 {
1481         int ret;
1482         void *data = ldb_module_get_private(module);
1483         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
1484
1485         if (ltdb->in_transaction != 1) {
1486                 return LDB_SUCCESS;
1487         }
1488
1489         /*
1490          * Check if the last re-index failed.
1491          *
1492          * This can happen if for example a duplicate value was marked
1493          * unique.  We must not write a partial re-index into the DB.
1494          */
1495         if (ltdb->reindex_failed) {
1496                 /*
1497                  * We must instead abort the transaction so we get the
1498                  * old values and old index back
1499                  */
1500                 ltdb_del_trans(module);
1501                 ldb_set_errstring(ldb_module_get_ctx(module),
1502                                   "Failure during re-index, so "
1503                                   "transaction must be aborted.");
1504                 return LDB_ERR_OPERATIONS_ERROR;
1505         }
1506
1507         ret = ltdb_index_transaction_commit(module);
1508         if (ret != LDB_SUCCESS) {
1509                 ltdb->kv_ops->abort_write(ltdb);
1510                 ltdb->in_transaction--;
1511                 return ret;
1512         }
1513
1514         if (ltdb->kv_ops->prepare_write(ltdb) != 0) {
1515                 ret = ltdb->kv_ops->error(ltdb);
1516                 ltdb->in_transaction--;
1517                 ldb_debug_set(ldb_module_get_ctx(module),
1518                               LDB_DEBUG_FATAL,
1519                               "Failure during "
1520                               "prepare_write): %s -> %s",
1521                               ltdb->kv_ops->errorstr(ltdb),
1522                               ldb_strerror(ret));
1523                 return ret;
1524         }
1525
1526         ltdb->prepared_commit = true;
1527
1528         return LDB_SUCCESS;
1529 }
1530
1531 static int ltdb_end_trans(struct ldb_module *module)
1532 {
1533         int ret;
1534         void *data = ldb_module_get_private(module);
1535         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
1536
1537         if (!ltdb->prepared_commit) {
1538                 ret = ltdb_prepare_commit(module);
1539                 if (ret != LDB_SUCCESS) {
1540                         return ret;
1541                 }
1542         }
1543
1544         ltdb->in_transaction--;
1545         ltdb->prepared_commit = false;
1546
1547         if (ltdb->kv_ops->finish_write(ltdb) != 0) {
1548                 ret = ltdb->kv_ops->error(ltdb);
1549                 ldb_asprintf_errstring(ldb_module_get_ctx(module),
1550                                        "Failure during tdb_transaction_commit(): %s -> %s",
1551                                        ltdb->kv_ops->errorstr(ltdb),
1552                                        ldb_strerror(ret));
1553                 return ret;
1554         }
1555
1556         return LDB_SUCCESS;
1557 }
1558
1559 static int ltdb_del_trans(struct ldb_module *module)
1560 {
1561         void *data = ldb_module_get_private(module);
1562         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
1563
1564         ltdb->in_transaction--;
1565
1566         if (ltdb_index_transaction_cancel(module) != 0) {
1567                 ltdb->kv_ops->abort_write(ltdb);
1568                 return ltdb->kv_ops->error(ltdb);
1569         }
1570
1571         ltdb->kv_ops->abort_write(ltdb);
1572         return LDB_SUCCESS;
1573 }
1574
1575 /*
1576   return sequenceNumber from @BASEINFO
1577 */
1578 static int ltdb_sequence_number(struct ltdb_context *ctx,
1579                                 struct ldb_extended **ext)
1580 {
1581         struct ldb_context *ldb;
1582         struct ldb_module *module = ctx->module;
1583         struct ldb_request *req = ctx->req;
1584         void *data = ldb_module_get_private(module);
1585         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
1586         TALLOC_CTX *tmp_ctx = NULL;
1587         struct ldb_seqnum_request *seq;
1588         struct ldb_seqnum_result *res;
1589         struct ldb_message *msg = NULL;
1590         struct ldb_dn *dn;
1591         const char *date;
1592         int ret = LDB_SUCCESS;
1593
1594         ldb = ldb_module_get_ctx(module);
1595
1596         seq = talloc_get_type(req->op.extended.data,
1597                                 struct ldb_seqnum_request);
1598         if (seq == NULL) {
1599                 return LDB_ERR_OPERATIONS_ERROR;
1600         }
1601
1602         ldb_request_set_state(req, LDB_ASYNC_PENDING);
1603
1604         if (ltdb->kv_ops->lock_read(module) != 0) {
1605                 return LDB_ERR_OPERATIONS_ERROR;
1606         }
1607
1608         res = talloc_zero(req, struct ldb_seqnum_result);
1609         if (res == NULL) {
1610                 ret = LDB_ERR_OPERATIONS_ERROR;
1611                 goto done;
1612         }
1613
1614         tmp_ctx = talloc_new(req);
1615         if (tmp_ctx == NULL) {
1616                 ret = LDB_ERR_OPERATIONS_ERROR;
1617                 goto done;
1618         }
1619
1620         dn = ldb_dn_new(tmp_ctx, ldb, LTDB_BASEINFO);
1621         if (dn == NULL) {
1622                 ret = LDB_ERR_OPERATIONS_ERROR;
1623                 goto done;
1624         }
1625
1626         msg = ldb_msg_new(tmp_ctx);
1627         if (msg == NULL) {
1628                 ret = LDB_ERR_OPERATIONS_ERROR;
1629                 goto done;
1630         }
1631
1632         ret = ltdb_search_dn1(module, dn, msg, 0);
1633         if (ret != LDB_SUCCESS) {
1634                 goto done;
1635         }
1636
1637         switch (seq->type) {
1638         case LDB_SEQ_HIGHEST_SEQ:
1639                 res->seq_num = ldb_msg_find_attr_as_uint64(msg, LTDB_SEQUENCE_NUMBER, 0);
1640                 break;
1641         case LDB_SEQ_NEXT:
1642                 res->seq_num = ldb_msg_find_attr_as_uint64(msg, LTDB_SEQUENCE_NUMBER, 0);
1643                 res->seq_num++;
1644                 break;
1645         case LDB_SEQ_HIGHEST_TIMESTAMP:
1646                 date = ldb_msg_find_attr_as_string(msg, LTDB_MOD_TIMESTAMP, NULL);
1647                 if (date) {
1648                         res->seq_num = ldb_string_to_time(date);
1649                 } else {
1650                         res->seq_num = 0;
1651                         /* zero is as good as anything when we don't know */
1652                 }
1653                 break;
1654         }
1655
1656         *ext = talloc_zero(req, struct ldb_extended);
1657         if (*ext == NULL) {
1658                 ret = LDB_ERR_OPERATIONS_ERROR;
1659                 goto done;
1660         }
1661         (*ext)->oid = LDB_EXTENDED_SEQUENCE_NUMBER;
1662         (*ext)->data = talloc_steal(*ext, res);
1663
1664 done:
1665         talloc_free(tmp_ctx);
1666
1667         ltdb->kv_ops->unlock_read(module);
1668         return ret;
1669 }
1670
1671 static void ltdb_request_done(struct ltdb_context *ctx, int error)
1672 {
1673         struct ldb_context *ldb;
1674         struct ldb_request *req;
1675         struct ldb_reply *ares;
1676
1677         ldb = ldb_module_get_ctx(ctx->module);
1678         req = ctx->req;
1679
1680         /* if we already returned an error just return */
1681         if (ldb_request_get_status(req) != LDB_SUCCESS) {
1682                 return;
1683         }
1684
1685         ares = talloc_zero(req, struct ldb_reply);
1686         if (!ares) {
1687                 ldb_oom(ldb);
1688                 req->callback(req, NULL);
1689                 return;
1690         }
1691         ares->type = LDB_REPLY_DONE;
1692         ares->error = error;
1693
1694         req->callback(req, ares);
1695 }
1696
1697 static void ltdb_timeout(struct tevent_context *ev,
1698                           struct tevent_timer *te,
1699                           struct timeval t,
1700                           void *private_data)
1701 {
1702         struct ltdb_context *ctx;
1703         ctx = talloc_get_type(private_data, struct ltdb_context);
1704
1705         if (!ctx->request_terminated) {
1706                 /* request is done now */
1707                 ltdb_request_done(ctx, LDB_ERR_TIME_LIMIT_EXCEEDED);
1708         }
1709
1710         if (ctx->spy) {
1711                 /* neutralize the spy */
1712                 ctx->spy->ctx = NULL;
1713                 ctx->spy = NULL;
1714         }
1715         talloc_free(ctx);
1716 }
1717
1718 static void ltdb_request_extended_done(struct ltdb_context *ctx,
1719                                         struct ldb_extended *ext,
1720                                         int error)
1721 {
1722         struct ldb_context *ldb;
1723         struct ldb_request *req;
1724         struct ldb_reply *ares;
1725
1726         ldb = ldb_module_get_ctx(ctx->module);
1727         req = ctx->req;
1728
1729         /* if we already returned an error just return */
1730         if (ldb_request_get_status(req) != LDB_SUCCESS) {
1731                 return;
1732         }
1733
1734         ares = talloc_zero(req, struct ldb_reply);
1735         if (!ares) {
1736                 ldb_oom(ldb);
1737                 req->callback(req, NULL);
1738                 return;
1739         }
1740         ares->type = LDB_REPLY_DONE;
1741         ares->response = ext;
1742         ares->error = error;
1743
1744         req->callback(req, ares);
1745 }
1746
1747 static void ltdb_handle_extended(struct ltdb_context *ctx)
1748 {
1749         struct ldb_extended *ext = NULL;
1750         int ret;
1751
1752         if (strcmp(ctx->req->op.extended.oid,
1753                    LDB_EXTENDED_SEQUENCE_NUMBER) == 0) {
1754                 /* get sequence number */
1755                 ret = ltdb_sequence_number(ctx, &ext);
1756         } else {
1757                 /* not recognized */
1758                 ret = LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
1759         }
1760
1761         ltdb_request_extended_done(ctx, ext, ret);
1762 }
1763
1764 struct kv_ctx {
1765         ldb_kv_traverse_fn kv_traverse_fn;
1766         void *ctx;
1767         struct ltdb_private *ltdb;
1768 };
1769
1770 static int ldb_tdb_traverse_fn_wrapper(struct tdb_context *tdb, TDB_DATA tdb_key, TDB_DATA tdb_data, void *ctx)
1771 {
1772         struct kv_ctx *kv_ctx = ctx;
1773         struct ldb_val key = {
1774                 .length = tdb_key.dsize,
1775                 .data = tdb_key.dptr,
1776         };
1777         struct ldb_val data = {
1778                 .length = tdb_data.dsize,
1779                 .data = tdb_data.dptr,
1780         };
1781         return kv_ctx->kv_traverse_fn(kv_ctx->ltdb, key, data, kv_ctx->ctx);
1782 }
1783
1784 static int ltdb_tdb_traverse_fn(struct ltdb_private *ltdb, ldb_kv_traverse_fn fn, void *ctx)
1785 {
1786         struct kv_ctx kv_ctx = {
1787                 .kv_traverse_fn = fn,
1788                 .ctx = ctx,
1789                 .ltdb = ltdb
1790         };
1791         if (ltdb->in_transaction != 0) {
1792                 return tdb_traverse(ltdb->tdb, ldb_tdb_traverse_fn_wrapper, &kv_ctx);
1793         } else {
1794                 return tdb_traverse_read(ltdb->tdb, ldb_tdb_traverse_fn_wrapper, &kv_ctx);
1795         }
1796 }
1797
1798 static int ltdb_tdb_update_in_iterate(struct ltdb_private *ltdb,
1799                                       struct ldb_val ldb_key,
1800                                       struct ldb_val ldb_key2,
1801                                       struct ldb_val ldb_data, void *state)
1802 {
1803         int tdb_ret;
1804         struct ldb_context *ldb;
1805         struct ltdb_reindex_context *ctx = (struct ltdb_reindex_context *)state;
1806         struct ldb_module *module = ctx->module;
1807         TDB_DATA key = {
1808                 .dptr = ldb_key.data,
1809                 .dsize = ldb_key.length
1810         };
1811         TDB_DATA key2 = {
1812                 .dptr = ldb_key2.data,
1813                 .dsize = ldb_key2.length
1814         };
1815         TDB_DATA data = {
1816                 .dptr = ldb_data.data,
1817                 .dsize = ldb_data.length
1818         };
1819
1820         ldb = ldb_module_get_ctx(module);
1821
1822         tdb_ret = tdb_delete(ltdb->tdb, key);
1823         if (tdb_ret != 0) {
1824                 ldb_debug(ldb, LDB_DEBUG_ERROR,
1825                           "Failed to delete %*.*s "
1826                           "for rekey as %*.*s: %s",
1827                           (int)key.dsize, (int)key.dsize,
1828                           (const char *)key.dptr,
1829                           (int)key2.dsize, (int)key2.dsize,
1830                           (const char *)key.dptr,
1831                           tdb_errorstr(ltdb->tdb));
1832                 ctx->error = ltdb_err_map(tdb_error(ltdb->tdb));
1833                 return -1;
1834         }
1835         tdb_ret = tdb_store(ltdb->tdb, key2, data, 0);
1836         if (tdb_ret != 0) {
1837                 ldb_debug(ldb, LDB_DEBUG_ERROR,
1838                           "Failed to rekey %*.*s as %*.*s: %s",
1839                           (int)key.dsize, (int)key.dsize,
1840                           (const char *)key.dptr,
1841                           (int)key2.dsize, (int)key2.dsize,
1842                           (const char *)key.dptr,
1843                           tdb_errorstr(ltdb->tdb));
1844                 ctx->error = ltdb_err_map(tdb_error(ltdb->tdb));
1845                 return -1;
1846         }
1847         return tdb_ret;
1848 }
1849
1850 static int ltdb_tdb_parse_record(struct ltdb_private *ltdb, TDB_DATA key,
1851                                  int (*parser)(TDB_DATA key, TDB_DATA data,
1852                                                void *private_data),
1853                                  void *ctx)
1854 {
1855         return tdb_parse_record(ltdb->tdb, key, parser, ctx);
1856 }
1857
1858 static const char * ltdb_tdb_name(struct ltdb_private *ltdb)
1859 {
1860         return tdb_name(ltdb->tdb);
1861 }
1862
1863 static bool ltdb_tdb_changed(struct ltdb_private *ltdb)
1864 {
1865         int seq = tdb_get_seqnum(ltdb->tdb);
1866         bool has_changed = (seq != ltdb->tdb_seqnum);
1867
1868         ltdb->tdb_seqnum = seq;
1869
1870         return has_changed;
1871 }
1872
1873 static const struct kv_db_ops key_value_ops = {
1874         .store = ltdb_tdb_store,
1875         .delete = ltdb_tdb_delete,
1876         .iterate = ltdb_tdb_traverse_fn,
1877         .update_in_iterate = ltdb_tdb_update_in_iterate,
1878         .fetch_and_parse = ltdb_tdb_parse_record,
1879         .lock_read = ltdb_lock_read,
1880         .unlock_read = ltdb_unlock_read,
1881         .begin_write = ltdb_tdb_transaction_start,
1882         .prepare_write = ltdb_tdb_transaction_prepare_commit,
1883         .finish_write = ltdb_tdb_transaction_commit,
1884         .abort_write = ltdb_tdb_transaction_cancel,
1885         .error = ltdb_error,
1886         .errorstr = ltdb_errorstr,
1887         .name = ltdb_tdb_name,
1888         .has_changed = ltdb_tdb_changed,
1889 };
1890
1891 static void ltdb_callback(struct tevent_context *ev,
1892                           struct tevent_timer *te,
1893                           struct timeval t,
1894                           void *private_data)
1895 {
1896         struct ltdb_context *ctx;
1897         int ret;
1898
1899         ctx = talloc_get_type(private_data, struct ltdb_context);
1900
1901         if (ctx->request_terminated) {
1902                 goto done;
1903         }
1904
1905         switch (ctx->req->operation) {
1906         case LDB_SEARCH:
1907                 ret = ltdb_search(ctx);
1908                 break;
1909         case LDB_ADD:
1910                 ret = ltdb_add(ctx);
1911                 break;
1912         case LDB_MODIFY:
1913                 ret = ltdb_modify(ctx);
1914                 break;
1915         case LDB_DELETE:
1916                 ret = ltdb_delete(ctx);
1917                 break;
1918         case LDB_RENAME:
1919                 ret = ltdb_rename(ctx);
1920                 break;
1921         case LDB_EXTENDED:
1922                 ltdb_handle_extended(ctx);
1923                 goto done;
1924         default:
1925                 /* no other op supported */
1926                 ret = LDB_ERR_PROTOCOL_ERROR;
1927         }
1928
1929         if (!ctx->request_terminated) {
1930                 /* request is done now */
1931                 ltdb_request_done(ctx, ret);
1932         }
1933
1934 done:
1935         if (ctx->spy) {
1936                 /* neutralize the spy */
1937                 ctx->spy->ctx = NULL;
1938                 ctx->spy = NULL;
1939         }
1940         talloc_free(ctx);
1941 }
1942
1943 static int ltdb_request_destructor(void *ptr)
1944 {
1945         struct ltdb_req_spy *spy = talloc_get_type(ptr, struct ltdb_req_spy);
1946
1947         if (spy->ctx != NULL) {
1948                 spy->ctx->spy = NULL;
1949                 spy->ctx->request_terminated = true;
1950                 spy->ctx = NULL;
1951         }
1952
1953         return 0;
1954 }
1955
1956 static int ltdb_handle_request(struct ldb_module *module,
1957                                struct ldb_request *req)
1958 {
1959         struct ldb_control *control_permissive;
1960         struct ldb_context *ldb;
1961         struct tevent_context *ev;
1962         struct ltdb_context *ac;
1963         struct tevent_timer *te;
1964         struct timeval tv;
1965         unsigned int i;
1966
1967         ldb = ldb_module_get_ctx(module);
1968
1969         control_permissive = ldb_request_get_control(req,
1970                                         LDB_CONTROL_PERMISSIVE_MODIFY_OID);
1971
1972         for (i = 0; req->controls && req->controls[i]; i++) {
1973                 if (req->controls[i]->critical &&
1974                     req->controls[i] != control_permissive) {
1975                         ldb_asprintf_errstring(ldb, "Unsupported critical extension %s",
1976                                                req->controls[i]->oid);
1977                         return LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
1978                 }
1979         }
1980
1981         if (req->starttime == 0 || req->timeout == 0) {
1982                 ldb_set_errstring(ldb, "Invalid timeout settings");
1983                 return LDB_ERR_TIME_LIMIT_EXCEEDED;
1984         }
1985
1986         ev = ldb_handle_get_event_context(req->handle);
1987
1988         ac = talloc_zero(ldb, struct ltdb_context);
1989         if (ac == NULL) {
1990                 ldb_oom(ldb);
1991                 return LDB_ERR_OPERATIONS_ERROR;
1992         }
1993
1994         ac->module = module;
1995         ac->req = req;
1996
1997         tv.tv_sec = 0;
1998         tv.tv_usec = 0;
1999         te = tevent_add_timer(ev, ac, tv, ltdb_callback, ac);
2000         if (NULL == te) {
2001                 talloc_free(ac);
2002                 return LDB_ERR_OPERATIONS_ERROR;
2003         }
2004
2005         if (req->timeout > 0) {
2006                 tv.tv_sec = req->starttime + req->timeout;
2007                 tv.tv_usec = 0;
2008                 ac->timeout_event = tevent_add_timer(ev, ac, tv,
2009                                                      ltdb_timeout, ac);
2010                 if (NULL == ac->timeout_event) {
2011                         talloc_free(ac);
2012                         return LDB_ERR_OPERATIONS_ERROR;
2013                 }
2014         }
2015
2016         /* set a spy so that we do not try to use the request context
2017          * if it is freed before ltdb_callback fires */
2018         ac->spy = talloc(req, struct ltdb_req_spy);
2019         if (NULL == ac->spy) {
2020                 talloc_free(ac);
2021                 return LDB_ERR_OPERATIONS_ERROR;
2022         }
2023         ac->spy->ctx = ac;
2024
2025         talloc_set_destructor((TALLOC_CTX *)ac->spy, ltdb_request_destructor);
2026
2027         return LDB_SUCCESS;
2028 }
2029
2030 static int ltdb_init_rootdse(struct ldb_module *module)
2031 {
2032         /* ignore errors on this - we expect it for non-sam databases */
2033         ldb_mod_register_control(module, LDB_CONTROL_PERMISSIVE_MODIFY_OID);
2034
2035         /* there can be no module beyond the backend, just return */
2036         return LDB_SUCCESS;
2037 }
2038
2039
2040 static int generic_lock_read(struct ldb_module *module)
2041 {
2042         void *data = ldb_module_get_private(module);
2043         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
2044         return ltdb->kv_ops->lock_read(module);
2045 }
2046
2047 static int generic_unlock_read(struct ldb_module *module)
2048 {
2049         void *data = ldb_module_get_private(module);
2050         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
2051         return ltdb->kv_ops->unlock_read(module);
2052 }
2053
2054 static const struct ldb_module_ops ltdb_ops = {
2055         .name              = "tdb",
2056         .init_context      = ltdb_init_rootdse,
2057         .search            = ltdb_handle_request,
2058         .add               = ltdb_handle_request,
2059         .modify            = ltdb_handle_request,
2060         .del               = ltdb_handle_request,
2061         .rename            = ltdb_handle_request,
2062         .extended          = ltdb_handle_request,
2063         .start_transaction = ltdb_start_trans,
2064         .end_transaction   = ltdb_end_trans,
2065         .prepare_commit    = ltdb_prepare_commit,
2066         .del_transaction   = ltdb_del_trans,
2067         .read_lock         = generic_lock_read,
2068         .read_unlock       = generic_unlock_read,
2069 };
2070
2071 int init_store(struct ltdb_private *ltdb,
2072                       const char *name,
2073                       struct ldb_context *ldb,
2074                       const char *options[],
2075                       struct ldb_module **_module)
2076 {
2077         struct ldb_module *module;
2078
2079         if (getenv("LDB_WARN_UNINDEXED")) {
2080                 ltdb->warn_unindexed = true;
2081         }
2082
2083         if (getenv("LDB_WARN_REINDEX")) {
2084                 ltdb->warn_reindex = true;
2085         }
2086
2087         ltdb->sequence_number = 0;
2088
2089         module = ldb_module_new(ldb, ldb, name, &ltdb_ops);
2090         if (!module) {
2091                 ldb_oom(ldb);
2092                 talloc_free(ltdb);
2093                 return LDB_ERR_OPERATIONS_ERROR;
2094         }
2095         ldb_module_set_private(module, ltdb);
2096         talloc_steal(module, ltdb);
2097
2098         if (ltdb_cache_load(module) != 0) {
2099                 ldb_asprintf_errstring(ldb, "Unable to load ltdb cache "
2100                                        "records for backend '%s'", name);
2101                 talloc_free(module);
2102                 return LDB_ERR_OPERATIONS_ERROR;
2103         }
2104
2105         *_module = module;
2106         /*
2107          * Set the maximum key length
2108          */
2109         {
2110                 const char *len_str =
2111                         ldb_options_find(ldb, options,
2112                                          "max_key_len_for_self_test");
2113                 if (len_str != NULL) {
2114                         unsigned len = strtoul(len_str, NULL, 0);
2115                         ltdb->max_key_length = len;
2116                 }
2117         }
2118
2119         return LDB_SUCCESS;
2120 }
2121
2122 /*
2123   connect to the database
2124 */
2125 int ltdb_connect(struct ldb_context *ldb, const char *url,
2126                  unsigned int flags, const char *options[],
2127                  struct ldb_module **_module)
2128 {
2129         const char *path;
2130         int tdb_flags, open_flags;
2131         struct ltdb_private *ltdb;
2132
2133         /*
2134          * We hold locks, so we must use a private event context
2135          * on each returned handle
2136          */
2137         ldb_set_require_private_event_context(ldb);
2138
2139         /* parse the url */
2140         if (strchr(url, ':')) {
2141                 if (strncmp(url, "tdb://", 6) != 0) {
2142                         ldb_debug(ldb, LDB_DEBUG_ERROR,
2143                                   "Invalid tdb URL '%s'", url);
2144                         return LDB_ERR_OPERATIONS_ERROR;
2145                 }
2146                 path = url+6;
2147         } else {
2148                 path = url;
2149         }
2150
2151         tdb_flags = TDB_DEFAULT | TDB_SEQNUM;
2152
2153         /* check for the 'nosync' option */
2154         if (flags & LDB_FLG_NOSYNC) {
2155                 tdb_flags |= TDB_NOSYNC;
2156         }
2157
2158         /* and nommap option */
2159         if (flags & LDB_FLG_NOMMAP) {
2160                 tdb_flags |= TDB_NOMMAP;
2161         }
2162
2163         ltdb = talloc_zero(ldb, struct ltdb_private);
2164         if (!ltdb) {
2165                 ldb_oom(ldb);
2166                 return LDB_ERR_OPERATIONS_ERROR;
2167         }
2168
2169         if (flags & LDB_FLG_RDONLY) {
2170                 /*
2171                  * This is weird, but because we can only have one tdb
2172                  * in this process, and the other one could be
2173                  * read-write, we can't use the tdb readonly.  Plus a
2174                  * read only tdb prohibits the all-record lock.
2175                  */
2176                 open_flags = O_RDWR;
2177
2178                 ltdb->read_only = true;
2179
2180         } else if (flags & LDB_FLG_DONT_CREATE_DB) {
2181                 /*
2182                  * This is used by ldbsearch to prevent creation of the database
2183                  * if the name is wrong
2184                  */
2185                 open_flags = O_RDWR;
2186         } else {
2187                 /*
2188                  * This is the normal case
2189                  */
2190                 open_flags = O_CREAT | O_RDWR;
2191         }
2192
2193         ltdb->kv_ops = &key_value_ops;
2194
2195         /* note that we use quite a large default hash size */
2196         ltdb->tdb = ltdb_wrap_open(ltdb, path, 10000,
2197                                    tdb_flags, open_flags,
2198                                    ldb_get_create_perms(ldb), ldb);
2199         if (!ltdb->tdb) {
2200                 ldb_asprintf_errstring(ldb,
2201                                        "Unable to open tdb '%s': %s", path, strerror(errno));
2202                 ldb_debug(ldb, LDB_DEBUG_ERROR,
2203                           "Unable to open tdb '%s': %s", path, strerror(errno));
2204                 talloc_free(ltdb);
2205                 if (errno == EACCES || errno == EPERM) {
2206                         return LDB_ERR_INSUFFICIENT_ACCESS_RIGHTS;
2207                 }
2208                 return LDB_ERR_OPERATIONS_ERROR;
2209         }
2210
2211         return init_store(ltdb, "ldb_tdb backend", ldb, options, _module);
2212 }