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