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