r23367: check the "use mmap" option for ldb too
[sfrench/samba-autobuild/.git] / source / 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
7    
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 2 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, write to the Free Software
25    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
26 */
27
28 /*
29  *  Name: ldb_tdb
30  *
31  *  Component: ldb tdb backend
32  *
33  *  Description: core functions for tdb backend
34  *
35  *  Author: Andrew Tridgell
36  *  Author: Stefan Metzmacher
37  *
38  *  Modifications:
39  *
40  *  - description: make the module use asyncronous calls
41  *    date: Feb 2006
42  *    Author: Simo Sorce
43  */
44
45 #include "includes.h"
46 #include "ldb/include/includes.h"
47
48 #include "ldb/ldb_tdb/ldb_tdb.h"
49
50 int ltdb_check_special_dn(struct ldb_module *module, const struct ldb_message *msg);
51
52 /*
53   map a tdb error code to a ldb error code
54 */
55 static int ltdb_err_map(enum TDB_ERROR tdb_code)
56 {
57         switch (tdb_code) {
58         case TDB_SUCCESS:
59                 return LDB_SUCCESS;
60         case TDB_ERR_CORRUPT:
61         case TDB_ERR_OOM:
62         case TDB_ERR_EINVAL:
63                 return LDB_ERR_OPERATIONS_ERROR;
64         case TDB_ERR_IO:
65                 return LDB_ERR_PROTOCOL_ERROR;
66         case TDB_ERR_LOCK:
67         case TDB_ERR_NOLOCK:
68                 return LDB_ERR_BUSY;
69         case TDB_ERR_LOCK_TIMEOUT:
70                 return LDB_ERR_TIME_LIMIT_EXCEEDED;
71         case TDB_ERR_EXISTS:
72                 return LDB_ERR_ENTRY_ALREADY_EXISTS;
73         case TDB_ERR_NOEXIST:
74                 return LDB_ERR_NO_SUCH_OBJECT;
75         case TDB_ERR_RDONLY:
76                 return LDB_ERR_INSUFFICIENT_ACCESS_RIGHTS;
77         }
78         return LDB_ERR_OTHER;
79 }
80
81
82 struct ldb_handle *init_ltdb_handle(struct ltdb_private *ltdb, struct ldb_module *module,
83                                     struct ldb_request *req)
84 {
85         struct ltdb_context *ac;
86         struct ldb_handle *h;
87
88         h = talloc_zero(req, struct ldb_handle);
89         if (h == NULL) {
90                 ldb_set_errstring(module->ldb, "Out of Memory");
91                 return NULL;
92         }
93
94         h->module = module;
95
96         ac = talloc_zero(h, struct ltdb_context);
97         if (ac == NULL) {
98                 ldb_set_errstring(module->ldb, "Out of Memory");
99                 talloc_free(h);
100                 return NULL;
101         }
102
103         h->private_data = (void *)ac;
104
105         h->state = LDB_ASYNC_INIT;
106         h->status = LDB_SUCCESS;
107
108         ac->module = module;
109         ac->context = req->context;
110         ac->callback = req->callback;
111
112         return h;
113 }
114
115 /*
116   form a TDB_DATA for a record key
117   caller frees
118
119   note that the key for a record can depend on whether the 
120   dn refers to a case sensitive index record or not
121 */
122 struct TDB_DATA ltdb_key(struct ldb_module *module, const struct ldb_dn *dn)
123 {
124         struct ldb_context *ldb = module->ldb;
125         TDB_DATA key;
126         char *key_str = NULL;
127         char *dn_folded = NULL;
128
129         /*
130           most DNs are case insensitive. The exception is index DNs for
131           case sensitive attributes
132
133           there are 3 cases dealt with in this code:
134
135           1) if the dn doesn't start with @ then uppercase the attribute
136              names and the attributes values of case insensitive attributes
137           2) if the dn starts with @ then leave it alone - the indexing code handles
138              the rest
139         */
140
141         dn_folded = ldb_dn_linearize_casefold(ldb, ldb, dn);
142         if (!dn_folded) {
143                 goto failed;
144         }
145
146         key_str = talloc_asprintf(ldb, "DN=%s", dn_folded);
147
148         talloc_free(dn_folded);
149
150         if (!key_str) {
151                 goto failed;
152         }
153
154         key.dptr = (uint8_t *)key_str;
155         key.dsize = strlen(key_str) + 1;
156
157         return key;
158
159 failed:
160         errno = ENOMEM;
161         key.dptr = NULL;
162         key.dsize = 0;
163         return key;
164 }
165
166 /*
167   check special dn's have valid attributes
168   currently only @ATTRIBUTES is checked
169 */
170 int ltdb_check_special_dn(struct ldb_module *module, const struct ldb_message *msg)
171 {
172         int i, j;
173  
174         if (! ldb_dn_is_special(msg->dn) ||
175             ! ldb_dn_check_special(msg->dn, LTDB_ATTRIBUTES)) {
176                 return 0;
177         }
178
179         /* we have @ATTRIBUTES, let's check attributes are fine */
180         /* should we check that we deny multivalued attributes ? */
181         for (i = 0; i < msg->num_elements; i++) {
182                 for (j = 0; j < msg->elements[i].num_values; j++) {
183                         if (ltdb_check_at_attributes_values(&msg->elements[i].values[j]) != 0) {
184                                 ldb_set_errstring(module->ldb, "Invalid attribute value in an @ATTRIBUTES entry");
185                                 return LDB_ERR_INVALID_ATTRIBUTE_SYNTAX;
186                         }
187                 }
188         }
189
190         return 0;
191 }
192
193
194 /*
195   we've made a modification to a dn - possibly reindex and 
196   update sequence number
197 */
198 static int ltdb_modified(struct ldb_module *module, const struct ldb_dn *dn)
199 {
200         int ret = 0;
201
202         if (ldb_dn_is_special(dn) &&
203             (ldb_dn_check_special(dn, LTDB_INDEXLIST) ||
204              ldb_dn_check_special(dn, LTDB_ATTRIBUTES)) ) {
205                 ret = ltdb_reindex(module);
206         }
207
208         if (ret == 0 &&
209             !(ldb_dn_is_special(dn) &&
210               ldb_dn_check_special(dn, LTDB_BASEINFO)) ) {
211                 ret = ltdb_increase_sequence_number(module);
212         }
213
214         return ret;
215 }
216
217 /*
218   store a record into the db
219 */
220 int ltdb_store(struct ldb_module *module, const struct ldb_message *msg, int flgs)
221 {
222         struct ltdb_private *ltdb =
223                 talloc_get_type(module->private_data, struct ltdb_private);
224         TDB_DATA tdb_key, tdb_data;
225         int ret;
226
227         tdb_key = ltdb_key(module, msg->dn);
228         if (!tdb_key.dptr) {
229                 return LDB_ERR_OTHER;
230         }
231
232         ret = ltdb_pack_data(module, msg, &tdb_data);
233         if (ret == -1) {
234                 talloc_free(tdb_key.dptr);
235                 return LDB_ERR_OTHER;
236         }
237
238         ret = tdb_store(ltdb->tdb, tdb_key, tdb_data, flgs);
239         if (ret == -1) {
240                 ret = ltdb_err_map(tdb_error(ltdb->tdb));
241                 goto done;
242         }
243         
244         ret = ltdb_index_add(module, msg);
245         if (ret == -1) {
246                 tdb_delete(ltdb->tdb, tdb_key);
247         }
248
249 done:
250         talloc_free(tdb_key.dptr);
251         talloc_free(tdb_data.dptr);
252
253         return ret;
254 }
255
256
257 static int ltdb_add_internal(struct ldb_module *module, const struct ldb_message *msg)
258 {
259         int ret;
260         
261         ret = ltdb_check_special_dn(module, msg);
262         if (ret != LDB_SUCCESS) {
263                 return ret;
264         }
265         
266         if (ltdb_cache_load(module) != 0) {
267                 return LDB_ERR_OPERATIONS_ERROR;
268         }
269
270         ret = ltdb_store(module, msg, TDB_INSERT);
271
272         if (ret == LDB_ERR_ENTRY_ALREADY_EXISTS) {
273                 char *dn;
274
275                 dn = ldb_dn_linearize(module, msg->dn);
276                 if (!dn) {
277                         return ret;
278                 }
279                 ldb_asprintf_errstring(module->ldb, "Entry %s already exists", dn);
280                 talloc_free(dn);
281                 return ret;
282         }
283         
284         if (ret == LDB_SUCCESS) {
285                 ret = ltdb_modified(module, msg->dn);
286                 if (ret != LDB_SUCCESS) {
287                         return LDB_ERR_OPERATIONS_ERROR;
288                 }
289         }
290
291         return ret;
292 }
293
294 /*
295   add a record to the database
296 */
297 static int ltdb_add(struct ldb_module *module, struct ldb_request *req)
298 {
299         struct ltdb_private *ltdb = talloc_get_type(module->private_data, struct ltdb_private);
300         struct ltdb_context *ltdb_ac;
301         int tret, ret = LDB_SUCCESS;
302
303         if (req->controls != NULL) {
304                 ldb_debug(module->ldb, LDB_DEBUG_WARNING, "Controls should not reach the ldb_tdb backend!\n");
305                 if (check_critical_controls(req->controls)) {
306                         return LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
307                 }
308         }
309         
310         req->handle = init_ltdb_handle(ltdb, module, req);
311         if (req->handle == NULL) {
312                 return LDB_ERR_OPERATIONS_ERROR;
313         }
314         ltdb_ac = talloc_get_type(req->handle->private_data, struct ltdb_context);
315
316         tret = ltdb_add_internal(module, req->op.add.message);
317         if (tret != LDB_SUCCESS) {
318                 req->handle->status = tret;
319                 goto done;
320         }
321         
322         if (ltdb_ac->callback) {
323                 ret = ltdb_ac->callback(module->ldb, ltdb_ac->context, NULL);
324         }
325 done:
326         req->handle->state = LDB_ASYNC_DONE;
327         return ret;
328 }
329
330 /*
331   delete a record from the database, not updating indexes (used for deleting
332   index records)
333 */
334 int ltdb_delete_noindex(struct ldb_module *module, const struct ldb_dn *dn)
335 {
336         struct ltdb_private *ltdb =
337                 talloc_get_type(module->private_data, struct ltdb_private);
338         TDB_DATA tdb_key;
339         int ret;
340
341         tdb_key = ltdb_key(module, dn);
342         if (!tdb_key.dptr) {
343                 return LDB_ERR_OTHER;
344         }
345
346         ret = tdb_delete(ltdb->tdb, tdb_key);
347         talloc_free(tdb_key.dptr);
348
349         if (ret != 0) {
350                 ret = ltdb_err_map(tdb_error(ltdb->tdb));
351         }
352
353         return ret;
354 }
355
356 static int ltdb_delete_internal(struct ldb_module *module, const struct ldb_dn *dn)
357 {
358         struct ldb_message *msg;
359         int ret;
360
361         msg = talloc(module, struct ldb_message);
362         if (msg == NULL) {
363                 return LDB_ERR_OPERATIONS_ERROR;
364         }
365
366         /* in case any attribute of the message was indexed, we need
367            to fetch the old record */
368         ret = ltdb_search_dn1(module, dn, msg);
369         if (ret != 1) {
370                 /* not finding the old record is an error */
371                 talloc_free(msg);
372                 return LDB_ERR_NO_SUCH_OBJECT;
373         }
374
375         ret = ltdb_delete_noindex(module, dn);
376         if (ret != LDB_SUCCESS) {
377                 talloc_free(msg);
378                 return LDB_ERR_NO_SUCH_OBJECT;
379         }
380
381         /* remove any indexed attributes */
382         ret = ltdb_index_del(module, msg);
383         if (ret != LDB_SUCCESS) {
384                 talloc_free(msg);
385                 return LDB_ERR_OPERATIONS_ERROR;
386         }
387
388         ret = ltdb_modified(module, dn);
389         if (ret != LDB_SUCCESS) {
390                 talloc_free(msg);
391                 return LDB_ERR_OPERATIONS_ERROR;
392         }
393
394         talloc_free(msg);
395         return LDB_SUCCESS;
396 }
397
398 /*
399   delete a record from the database
400 */
401 static int ltdb_delete(struct ldb_module *module, struct ldb_request *req)
402 {
403         struct ltdb_private *ltdb = talloc_get_type(module->private_data, struct ltdb_private);
404         struct ltdb_context *ltdb_ac;
405         int tret, ret = LDB_SUCCESS;
406
407         if (req->controls != NULL) {
408                 ldb_debug(module->ldb, LDB_DEBUG_WARNING, "Controls should not reach the ldb_tdb backend!\n");
409                 if (check_critical_controls(req->controls)) {
410                         return LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
411                 }
412         }
413         
414         req->handle = NULL;
415
416         if (ltdb_cache_load(module) != 0) {
417                 return LDB_ERR_OPERATIONS_ERROR;
418         }
419
420         req->handle = init_ltdb_handle(ltdb, module, req);
421         if (req->handle == NULL) {
422                 return LDB_ERR_OPERATIONS_ERROR;
423         }
424         ltdb_ac = talloc_get_type(req->handle->private_data, struct ltdb_context);
425
426         tret = ltdb_delete_internal(module, req->op.del.dn);
427         if (tret != LDB_SUCCESS) {
428                 req->handle->status = tret; 
429                 goto done;
430         }
431
432         if (ltdb_ac->callback) {
433                 ret = ltdb_ac->callback(module->ldb, ltdb_ac->context, NULL);
434         }
435 done:
436         req->handle->state = LDB_ASYNC_DONE;
437         return ret;
438 }
439
440 /*
441   find an element by attribute name. At the moment this does a linear search, it should
442   be re-coded to use a binary search once all places that modify records guarantee
443   sorted order
444
445   return the index of the first matching element if found, otherwise -1
446 */
447 static int find_element(const struct ldb_message *msg, const char *name)
448 {
449         unsigned int i;
450         for (i=0;i<msg->num_elements;i++) {
451                 if (ldb_attr_cmp(msg->elements[i].name, name) == 0) {
452                         return i;
453                 }
454         }
455         return -1;
456 }
457
458
459 /*
460   add an element to an existing record. Assumes a elements array that we
461   can call re-alloc on, and assumed that we can re-use the data pointers from the 
462   passed in additional values. Use with care!
463
464   returns 0 on success, -1 on failure (and sets errno)
465 */
466 static int msg_add_element(struct ldb_context *ldb,
467                            struct ldb_message *msg, struct ldb_message_element *el)
468 {
469         struct ldb_message_element *e2;
470         unsigned int i;
471
472         e2 = talloc_realloc(msg, msg->elements, struct ldb_message_element, 
473                               msg->num_elements+1);
474         if (!e2) {
475                 errno = ENOMEM;
476                 return -1;
477         }
478
479         msg->elements = e2;
480
481         e2 = &msg->elements[msg->num_elements];
482
483         e2->name = el->name;
484         e2->flags = el->flags;
485         e2->values = NULL;
486         if (el->num_values != 0) {
487                 e2->values = talloc_array(msg->elements, struct ldb_val, el->num_values);
488                 if (!e2->values) {
489                         errno = ENOMEM;
490                         return -1;
491                 }
492         }
493         for (i=0;i<el->num_values;i++) {
494                 e2->values[i] = el->values[i];
495         }
496         e2->num_values = el->num_values;
497
498         msg->num_elements++;
499
500         return 0;
501 }
502
503 /*
504   delete all elements having a specified attribute name
505 */
506 static int msg_delete_attribute(struct ldb_module *module,
507                                 struct ldb_context *ldb,
508                                 struct ldb_message *msg, const char *name)
509 {
510         char *dn;
511         unsigned int i, j;
512
513         dn = ldb_dn_linearize(ldb, msg->dn);
514         if (dn == NULL) {
515                 return -1;
516         }
517
518         for (i=0;i<msg->num_elements;i++) {
519                 if (ldb_attr_cmp(msg->elements[i].name, name) == 0) {
520                         for (j=0;j<msg->elements[i].num_values;j++) {
521                                 ltdb_index_del_value(module, dn, &msg->elements[i], j);
522                         }
523                         talloc_free(msg->elements[i].values);
524                         if (msg->num_elements > (i+1)) {
525                                 memmove(&msg->elements[i], 
526                                         &msg->elements[i+1], 
527                                         sizeof(struct ldb_message_element)*
528                                         (msg->num_elements - (i+1)));
529                         }
530                         msg->num_elements--;
531                         i--;
532                         msg->elements = talloc_realloc(msg, msg->elements, 
533                                                          struct ldb_message_element, 
534                                                          msg->num_elements);
535                 }
536         }
537
538         talloc_free(dn);
539         return 0;
540 }
541
542 /*
543   delete all elements matching an attribute name/value 
544
545   return 0 on success, -1 on failure
546 */
547 static int msg_delete_element(struct ldb_module *module,
548                               struct ldb_message *msg, 
549                               const char *name,
550                               const struct ldb_val *val)
551 {
552         struct ldb_context *ldb = module->ldb;
553         unsigned int i;
554         int found;
555         struct ldb_message_element *el;
556         const struct ldb_attrib_handler *h;
557
558         found = find_element(msg, name);
559         if (found == -1) {
560                 return -1;
561         }
562
563         el = &msg->elements[found];
564
565         h = ldb_attrib_handler(ldb, el->name);
566
567         for (i=0;i<el->num_values;i++) {
568                 if (h->comparison_fn(ldb, ldb, &el->values[i], val) == 0) {
569                         if (i<el->num_values-1) {
570                                 memmove(&el->values[i], &el->values[i+1],
571                                         sizeof(el->values[i])*(el->num_values-(i+1)));
572                         }
573                         el->num_values--;
574                         if (el->num_values == 0) {
575                                 return msg_delete_attribute(module, ldb, msg, name);
576                         }
577                         return 0;
578                 }
579         }
580
581         return -1;
582 }
583
584
585 /*
586   modify a record - internal interface
587
588   yuck - this is O(n^2). Luckily n is usually small so we probably
589   get away with it, but if we ever have really large attribute lists 
590   then we'll need to look at this again
591 */
592 int ltdb_modify_internal(struct ldb_module *module, const struct ldb_message *msg)
593 {
594         struct ldb_context *ldb = module->ldb;
595         struct ltdb_private *ltdb =
596                 talloc_get_type(module->private_data, struct ltdb_private);
597         TDB_DATA tdb_key, tdb_data;
598         struct ldb_message *msg2;
599         unsigned i, j;
600         int ret;
601
602         tdb_key = ltdb_key(module, msg->dn);
603         if (!tdb_key.dptr) {
604                 return LDB_ERR_OTHER;
605         }
606
607         tdb_data = tdb_fetch(ltdb->tdb, tdb_key);
608         if (!tdb_data.dptr) {
609                 talloc_free(tdb_key.dptr);
610                 return ltdb_err_map(tdb_error(ltdb->tdb));
611         }
612
613         msg2 = talloc(tdb_key.dptr, struct ldb_message);
614         if (msg2 == NULL) {
615                 talloc_free(tdb_key.dptr);
616                 return LDB_ERR_OTHER;
617         }
618
619         ret = ltdb_unpack_data(module, &tdb_data, msg2);
620         if (ret == -1) {
621                 ret = LDB_ERR_OTHER;
622                 goto failed;
623         }
624
625         if (!msg2->dn) {
626                 msg2->dn = msg->dn;
627         }
628
629         for (i=0;i<msg->num_elements;i++) {
630                 struct ldb_message_element *el = &msg->elements[i];
631                 struct ldb_message_element *el2;
632                 struct ldb_val *vals;
633                 char *dn;
634
635                 switch (msg->elements[i].flags & LDB_FLAG_MOD_MASK) {
636
637                 case LDB_FLAG_MOD_ADD:
638                         /* add this element to the message. fail if it
639                            already exists */
640                         ret = find_element(msg2, el->name);
641
642                         if (ret == -1) {
643                                 if (msg_add_element(ldb, msg2, el) != 0) {
644                                         ret = LDB_ERR_OTHER;
645                                         goto failed;
646                                 }
647                                 continue;
648                         }
649
650                         el2 = &msg2->elements[ret];
651
652                         /* An attribute with this name already exists, add all
653                          * values if they don't already exist. */
654
655                         for (j=0;j<el->num_values;j++) {
656                                 if (ldb_msg_find_val(el2, &el->values[j])) {
657                                         ldb_set_errstring(module->ldb, "Type or value exists");
658                                         ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
659                                         goto failed;
660                                 }
661                         }
662
663                         vals = talloc_realloc(msg2->elements, el2->values, struct ldb_val,
664                                                 el2->num_values + el->num_values);
665
666                         if (vals == NULL) {
667                                 ret = LDB_ERR_OTHER;
668                                 goto failed;
669                         }
670
671                         for (j=0;j<el->num_values;j++) {
672                                 vals[el2->num_values + j] =
673                                         ldb_val_dup(vals, &el->values[j]);
674                         }
675
676                         el2->values = vals;
677                         el2->num_values += el->num_values;
678
679                         break;
680
681                 case LDB_FLAG_MOD_REPLACE:
682                         /* replace all elements of this attribute name with the elements
683                            listed. The attribute not existing is not an error */
684                         msg_delete_attribute(module, ldb, msg2, msg->elements[i].name);
685
686                         /* add the replacement element, if not empty */
687                         if (msg->elements[i].num_values != 0 &&
688                             msg_add_element(ldb, msg2, &msg->elements[i]) != 0) {
689                                 ret = LDB_ERR_OTHER;
690                                 goto failed;
691                         }
692                         break;
693
694                 case LDB_FLAG_MOD_DELETE:
695
696                         dn = ldb_dn_linearize(msg2, msg->dn);
697                         if (dn == NULL) {
698                                 ret = LDB_ERR_OTHER;
699                                 goto failed;
700                         }
701
702                         /* we could be being asked to delete all
703                            values or just some values */
704                         if (msg->elements[i].num_values == 0) {
705                                 if (msg_delete_attribute(module, ldb, msg2, 
706                                                          msg->elements[i].name) != 0) {
707                                         ldb_asprintf_errstring(module->ldb, "No such attribute: %s for delete on %s", msg->elements[i].name, dn);
708                                         ret = LDB_ERR_NO_SUCH_ATTRIBUTE;
709                                         goto failed;
710                                 }
711                                 break;
712                         }
713                         for (j=0;j<msg->elements[i].num_values;j++) {
714                                 if (msg_delete_element(module,
715                                                        msg2, 
716                                                        msg->elements[i].name,
717                                                        &msg->elements[i].values[j]) != 0) {
718                                         ldb_asprintf_errstring(module->ldb, "No matching attribute value when deleting attribute: %s on %s", msg->elements[i].name, dn);
719                                         ret = LDB_ERR_NO_SUCH_ATTRIBUTE;
720                                         goto failed;
721                                 }
722                                 if (ltdb_index_del_value(module, dn, &msg->elements[i], j) != 0) {
723                                         ret = LDB_ERR_OTHER;
724                                         goto failed;
725                                 }
726                         }
727                         break;
728                 default:
729                         ldb_asprintf_errstring(module->ldb, "Invalid ldb_modify flags on %s: 0x%x", 
730                                                              msg->elements[i].name, 
731                                                              msg->elements[i].flags & LDB_FLAG_MOD_MASK);
732                         ret = LDB_ERR_PROTOCOL_ERROR;
733                         goto failed;
734                 }
735         }
736
737         /* we've made all the mods - save the modified record back into the database */
738         ret = ltdb_store(module, msg2, TDB_MODIFY);
739         if (ret != LDB_SUCCESS) {
740                 goto failed;
741         }
742
743         if (ltdb_modified(module, msg->dn) != LDB_SUCCESS) {
744                 ret = LDB_ERR_OPERATIONS_ERROR;
745                 goto failed;
746         }
747
748         talloc_free(tdb_key.dptr);
749         free(tdb_data.dptr);
750         return ret;
751
752 failed:
753         talloc_free(tdb_key.dptr);
754         free(tdb_data.dptr);
755         return ret;
756 }
757
758 /*
759   modify a record
760 */
761 static int ltdb_modify(struct ldb_module *module, struct ldb_request *req)
762 {
763         struct ltdb_private *ltdb = talloc_get_type(module->private_data, struct ltdb_private);
764         struct ltdb_context *ltdb_ac;
765         int tret, ret = LDB_SUCCESS;
766
767         if (req->controls != NULL) {
768                 ldb_debug(module->ldb, LDB_DEBUG_WARNING, "Controls should not reach the ldb_tdb backend!\n");
769                 if (check_critical_controls(req->controls)) {
770                         return LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
771                 }
772         }
773         
774         req->handle = NULL;
775
776         req->handle = init_ltdb_handle(ltdb, module, req);
777         if (req->handle == NULL) {
778                 return LDB_ERR_OPERATIONS_ERROR;
779         }
780         ltdb_ac = talloc_get_type(req->handle->private_data, struct ltdb_context);
781
782         tret = ltdb_check_special_dn(module, req->op.mod.message);
783         if (tret != LDB_SUCCESS) {
784                 req->handle->status = tret;
785                 goto done;
786         }
787         
788         if (ltdb_cache_load(module) != 0) {
789                 ret = LDB_ERR_OPERATIONS_ERROR;
790                 goto done;
791         }
792
793         tret = ltdb_modify_internal(module, req->op.mod.message);
794         if (tret != LDB_SUCCESS) {
795                 req->handle->status = tret;
796                 goto done;
797         }
798
799         if (ltdb_ac->callback) {
800                 ret = ltdb_ac->callback(module->ldb, ltdb_ac->context, NULL);
801         }
802 done:
803         req->handle->state = LDB_ASYNC_DONE;
804         return ret;
805 }
806
807 /*
808   rename a record
809 */
810 static int ltdb_rename(struct ldb_module *module, struct ldb_request *req)
811 {
812         struct ltdb_private *ltdb = talloc_get_type(module->private_data, struct ltdb_private);
813         struct ltdb_context *ltdb_ac;
814         struct ldb_message *msg;
815         int tret, ret = LDB_SUCCESS;
816
817         if (req->controls != NULL) {
818                 ldb_debug(module->ldb, LDB_DEBUG_WARNING, "Controls should not reach the ldb_tdb backend!\n");
819                 if (check_critical_controls(req->controls)) {
820                         return LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
821                 }
822         }
823         
824         req->handle = NULL;
825
826         if (ltdb_cache_load(module) != 0) {
827                 return LDB_ERR_OPERATIONS_ERROR;
828         }
829
830         req->handle = init_ltdb_handle(ltdb, module, req);
831         if (req->handle == NULL) {
832                 return LDB_ERR_OPERATIONS_ERROR;
833         }
834         ltdb_ac = talloc_get_type(req->handle->private_data, struct ltdb_context);
835
836         msg = talloc(ltdb_ac, struct ldb_message);
837         if (msg == NULL) {
838                 ret = LDB_ERR_OPERATIONS_ERROR;
839                 goto done;
840         }
841
842         /* in case any attribute of the message was indexed, we need
843            to fetch the old record */
844         tret = ltdb_search_dn1(module, req->op.rename.olddn, msg);
845         if (tret != 1) {
846                 /* not finding the old record is an error */
847                 req->handle->status = LDB_ERR_NO_SUCH_OBJECT;
848                 goto done;
849         }
850
851         msg->dn = ldb_dn_copy(msg, req->op.rename.newdn);
852         if (!msg->dn) {
853                 ret = LDB_ERR_OPERATIONS_ERROR;
854                 goto done;
855         }
856
857         tret = ltdb_add_internal(module, msg);
858         if (tret != LDB_SUCCESS) {
859                 ret = LDB_ERR_OPERATIONS_ERROR;
860                 goto done;
861         }
862
863         tret = ltdb_delete_internal(module, req->op.rename.olddn);
864         if (tret != LDB_SUCCESS) {
865                 ltdb_delete_internal(module, req->op.rename.newdn);
866                 ret = LDB_ERR_OPERATIONS_ERROR;
867                 goto done;
868         }
869
870         if (ltdb_ac->callback) {
871                 ret = ltdb_ac->callback(module->ldb, ltdb_ac->context, NULL);
872         }
873 done:
874         req->handle->state = LDB_ASYNC_DONE;
875         return ret;
876 }
877
878 static int ltdb_start_trans(struct ldb_module *module)
879 {
880         struct ltdb_private *ltdb =
881                 talloc_get_type(module->private_data, struct ltdb_private);
882
883         if (tdb_transaction_start(ltdb->tdb) != 0) {
884                 return ltdb_err_map(tdb_error(ltdb->tdb));
885         }
886
887         return LDB_SUCCESS;
888 }
889
890 static int ltdb_end_trans(struct ldb_module *module)
891 {
892         struct ltdb_private *ltdb =
893                 talloc_get_type(module->private_data, struct ltdb_private);
894
895         if (tdb_transaction_commit(ltdb->tdb) != 0) {
896                 return ltdb_err_map(tdb_error(ltdb->tdb));
897         }
898
899         return LDB_SUCCESS;
900 }
901
902 static int ltdb_del_trans(struct ldb_module *module)
903 {
904         struct ltdb_private *ltdb =
905                 talloc_get_type(module->private_data, struct ltdb_private);
906
907         if (tdb_transaction_cancel(ltdb->tdb) != 0) {
908                 return ltdb_err_map(tdb_error(ltdb->tdb));
909         }
910
911         return LDB_SUCCESS;
912 }
913
914 static int ltdb_wait(struct ldb_handle *handle, enum ldb_wait_type type)
915 {
916         return handle->status;
917 }
918
919 static int ltdb_request(struct ldb_module *module, struct ldb_request *req)
920 {
921         /* check for oustanding critical controls and return an error if found */
922         if (req->controls != NULL) {
923                 ldb_debug(module->ldb, LDB_DEBUG_WARNING, "Controls should not reach the ldb_tdb backend!\n");
924                 if (check_critical_controls(req->controls)) {
925                         return LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
926                 }
927         }
928         
929         /* search, add, modify, delete, rename are handled by their own, no other op supported */
930         return LDB_ERR_OPERATIONS_ERROR;
931 }
932
933 /*
934   return sequenceNumber from @BASEINFO
935 */
936 static int ltdb_sequence_number(struct ldb_module *module, struct ldb_request *req)
937 {
938         TALLOC_CTX *tmp_ctx = talloc_new(req);
939         struct ldb_message *msg = NULL;
940         struct ldb_dn *dn = ldb_dn_explode(tmp_ctx, LTDB_BASEINFO);
941         int tret;
942
943         if (tmp_ctx == NULL) {
944                 talloc_free(tmp_ctx);
945                 return LDB_ERR_OPERATIONS_ERROR;
946         }
947
948         msg = talloc(tmp_ctx, struct ldb_message);
949         if (msg == NULL) {
950                 talloc_free(tmp_ctx);
951                 return LDB_ERR_OPERATIONS_ERROR;
952         }
953
954         req->op.seq_num.flags = 0;
955
956         tret = ltdb_search_dn1(module, dn, msg);
957         if (tret != 1) {
958                 talloc_free(tmp_ctx);
959                 req->op.seq_num.seq_num = 0;
960                 /* zero is as good as anything when we don't know */
961                 return LDB_SUCCESS;
962         }
963
964         switch (req->op.seq_num.type) {
965         case LDB_SEQ_HIGHEST_SEQ:
966                 req->op.seq_num.seq_num = ldb_msg_find_attr_as_uint64(msg, LTDB_SEQUENCE_NUMBER, 0);
967                 break;
968         case LDB_SEQ_NEXT:
969                 req->op.seq_num.seq_num = ldb_msg_find_attr_as_uint64(msg, LTDB_SEQUENCE_NUMBER, 0);
970                 req->op.seq_num.seq_num++;
971                 break;
972         case LDB_SEQ_HIGHEST_TIMESTAMP:
973         {
974                 const char *date = ldb_msg_find_attr_as_string(msg, LTDB_MOD_TIMESTAMP, NULL);
975                 if (date) {
976                         req->op.seq_num.seq_num = ldb_string_to_time(date);
977                 } else {
978                         req->op.seq_num.seq_num = 0;
979                         /* zero is as good as anything when we don't know */
980                 }
981                 break;
982         }
983         }
984         talloc_free(tmp_ctx);
985         return LDB_SUCCESS;
986 }
987
988 static const struct ldb_module_ops ltdb_ops = {
989         .name              = "tdb",
990         .search            = ltdb_search,
991         .add               = ltdb_add,
992         .modify            = ltdb_modify,
993         .del               = ltdb_delete,
994         .rename            = ltdb_rename,
995         .request           = ltdb_request,
996         .start_transaction = ltdb_start_trans,
997         .end_transaction   = ltdb_end_trans,
998         .del_transaction   = ltdb_del_trans,
999         .wait              = ltdb_wait,
1000         .sequence_number   = ltdb_sequence_number
1001 };
1002
1003 /*
1004   connect to the database
1005 */
1006 static int ltdb_connect(struct ldb_context *ldb, const char *url, 
1007                         unsigned int flags, const char *options[],
1008                         struct ldb_module **module)
1009 {
1010         const char *path;
1011         int tdb_flags, open_flags;
1012         struct ltdb_private *ltdb;
1013
1014         /* parse the url */
1015         if (strchr(url, ':')) {
1016                 if (strncmp(url, "tdb://", 6) != 0) {
1017                         ldb_debug(ldb, LDB_DEBUG_ERROR, "Invalid tdb URL '%s'", url);
1018                         return -1;
1019                 }
1020                 path = url+6;
1021         } else {
1022                 path = url;
1023         }
1024
1025         tdb_flags = TDB_DEFAULT | TDB_SEQNUM;
1026
1027         /* check for the 'nosync' option */
1028         if (flags & LDB_FLG_NOSYNC) {
1029                 tdb_flags |= TDB_NOSYNC;
1030         }
1031
1032         /* and nommap option */
1033         if (flags & LDB_FLG_NOMMAP) {
1034                 tdb_flags |= TDB_NOMMAP;
1035         }
1036
1037         if (flags & LDB_FLG_RDONLY) {
1038                 open_flags = O_RDONLY;
1039         } else {
1040                 open_flags = O_CREAT | O_RDWR;
1041         }
1042
1043         ltdb = talloc_zero(ldb, struct ltdb_private);
1044         if (!ltdb) {
1045                 ldb_oom(ldb);
1046                 return -1;
1047         }
1048
1049         /* note that we use quite a large default hash size */
1050         ltdb->tdb = ltdb_wrap_open(ltdb, path, 10000, 
1051                                    tdb_flags, open_flags, 
1052                                    ldb->create_perms, ldb);
1053         if (!ltdb->tdb) {
1054                 ldb_debug(ldb, LDB_DEBUG_ERROR, "Unable to open tdb '%s'\n", path);
1055                 talloc_free(ltdb);
1056                 return -1;
1057         }
1058
1059         ltdb->sequence_number = 0;
1060
1061         *module = talloc(ldb, struct ldb_module);
1062         if (!module) {
1063                 ldb_oom(ldb);
1064                 talloc_free(ltdb);
1065                 return -1;
1066         }
1067         talloc_set_name_const(*module, "ldb_tdb backend");
1068         (*module)->ldb = ldb;
1069         (*module)->prev = (*module)->next = NULL;
1070         (*module)->private_data = ltdb;
1071         (*module)->ops = &ltdb_ops;
1072
1073         if (ltdb_cache_load(*module) != 0) {
1074                 talloc_free(*module);
1075                 talloc_free(ltdb);
1076                 return -1;
1077         }
1078
1079         return 0;
1080 }
1081
1082 int ldb_tdb_init(void)
1083 {
1084         return ldb_register_backend("tdb", ltdb_connect);
1085 }