b7eec40e9a5663b62be0c38489adb7d231f1a071
[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 static 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         if (ltdb->in_transaction == 0) {
92                 return tdb_lockall_read(ltdb->tdb);
93         }
94         return 0;
95 }
96
97 /*
98   unlock the database after a ltdb_lock_read()
99 */
100 int ltdb_unlock_read(struct ldb_module *module)
101 {
102         void *data = ldb_module_get_private(module);
103         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
104         if (ltdb->in_transaction == 0) {
105                 return tdb_unlockall_read(ltdb->tdb);
106         }
107         return 0;
108 }
109
110
111 /*
112   form a TDB_DATA for a record key
113   caller frees
114
115   note that the key for a record can depend on whether the
116   dn refers to a case sensitive index record or not
117 */
118 struct TDB_DATA ltdb_key(struct ldb_module *module, struct ldb_dn *dn)
119 {
120         struct ldb_context *ldb = ldb_module_get_ctx(module);
121         TDB_DATA key;
122         char *key_str = NULL;
123         const char *dn_folded = NULL;
124
125         /*
126           most DNs are case insensitive. The exception is index DNs for
127           case sensitive attributes
128
129           there are 3 cases dealt with in this code:
130
131           1) if the dn doesn't start with @ then uppercase the attribute
132              names and the attributes values of case insensitive attributes
133           2) if the dn starts with @ then leave it alone -
134              the indexing code handles the rest
135         */
136
137         dn_folded = ldb_dn_get_casefold(dn);
138         if (!dn_folded) {
139                 goto failed;
140         }
141
142         key_str = talloc_strdup(ldb, "DN=");
143         if (!key_str) {
144                 goto failed;
145         }
146
147         key_str = talloc_strdup_append_buffer(key_str, dn_folded);
148         if (!key_str) {
149                 goto failed;
150         }
151
152         key.dptr = (uint8_t *)key_str;
153         key.dsize = strlen(key_str) + 1;
154
155         return key;
156
157 failed:
158         errno = ENOMEM;
159         key.dptr = NULL;
160         key.dsize = 0;
161         return key;
162 }
163
164 /*
165   check special dn's have valid attributes
166   currently only @ATTRIBUTES is checked
167 */
168 static int ltdb_check_special_dn(struct ldb_module *module,
169                           const struct ldb_message *msg)
170 {
171         struct ldb_context *ldb = ldb_module_get_ctx(module);
172         int i, j;
173
174         if (! ldb_dn_is_special(msg->dn) ||
175             ! ldb_dn_check_special(msg->dn, LTDB_ATTRIBUTES)) {
176                 return LDB_SUCCESS;
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(ldb, "Invalid attribute value in an @ATTRIBUTES entry");
185                                 return LDB_ERR_INVALID_ATTRIBUTE_SYNTAX;
186                         }
187                 }
188         }
189
190         return LDB_SUCCESS;
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, struct ldb_dn *dn)
199 {
200         int ret = LDB_SUCCESS;
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 the modify was to a normal record, or any special except @BASEINFO, update the seq number */
209         if (ret == LDB_SUCCESS &&
210             !(ldb_dn_is_special(dn) &&
211               ldb_dn_check_special(dn, LTDB_BASEINFO)) ) {
212                 ret = ltdb_increase_sequence_number(module);
213         }
214
215         /* If the modify was to @OPTIONS, reload the cache */
216         if (ldb_dn_is_special(dn) &&
217             (ldb_dn_check_special(dn, LTDB_OPTIONS)) ) {
218                 ret = ltdb_cache_reload(module);
219         }
220
221         return ret;
222 }
223
224 /*
225   store a record into the db
226 */
227 int ltdb_store(struct ldb_module *module, const struct ldb_message *msg, int flgs)
228 {
229         void *data = ldb_module_get_private(module);
230         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
231         TDB_DATA tdb_key, tdb_data;
232         int ret = LDB_SUCCESS;
233
234         tdb_key = ltdb_key(module, msg->dn);
235         if (!tdb_key.dptr) {
236                 return LDB_ERR_OTHER;
237         }
238
239         ret = ltdb_pack_data(module, msg, &tdb_data);
240         if (ret == -1) {
241                 talloc_free(tdb_key.dptr);
242                 return LDB_ERR_OTHER;
243         }
244
245         ret = tdb_store(ltdb->tdb, tdb_key, tdb_data, flgs);
246         if (ret == -1) {
247                 ret = ltdb_err_map(tdb_error(ltdb->tdb));
248                 goto done;
249         }
250
251         ret = ltdb_index_add(module, msg);
252         if (ret != LDB_SUCCESS) {
253                 tdb_delete(ltdb->tdb, tdb_key);
254         }
255
256 done:
257         talloc_free(tdb_key.dptr);
258         talloc_free(tdb_data.dptr);
259
260         return ret;
261 }
262
263
264 static int ltdb_add_internal(struct ldb_module *module,
265                              const struct ldb_message *msg)
266 {
267         struct ldb_context *ldb = ldb_module_get_ctx(module);
268         int ret = LDB_SUCCESS, i;
269
270         ret = ltdb_check_special_dn(module, msg);
271         if (ret != LDB_SUCCESS) {
272                 goto done;
273         }
274
275         if (ltdb_cache_load(module) != 0) {
276                 ret = LDB_ERR_OPERATIONS_ERROR;
277                 goto done;
278         }
279
280         for (i=0;i<msg->num_elements;i++) {
281                 struct ldb_message_element *el = &msg->elements[i];
282                 const struct ldb_schema_attribute *a = ldb_schema_attribute_by_name(ldb, el->name);
283
284                 if (el->num_values == 0) {
285                         ldb_asprintf_errstring(ldb, "attribute %s on %s specified, but with 0 values (illegal)", 
286                                                el->name, ldb_dn_get_linearized(msg->dn));
287                         ret = LDB_ERR_CONSTRAINT_VIOLATION;
288                         goto done;
289                 }
290                 if (a && a->flags & LDB_ATTR_FLAG_SINGLE_VALUE) {
291                         if (el->num_values > 1) {
292                                 ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
293                                                        el->name, ldb_dn_get_linearized(msg->dn));
294                                 ret = LDB_ERR_CONSTRAINT_VIOLATION;
295                                 goto done;
296                         }
297                 }
298         }
299
300         ret = ltdb_store(module, msg, TDB_INSERT);
301         if (ret != LDB_SUCCESS) {
302                 if (ret == LDB_ERR_ENTRY_ALREADY_EXISTS) {
303                         ldb_asprintf_errstring(ldb,
304                                                "Entry %s already exists",
305                                                ldb_dn_get_linearized(msg->dn));
306                 }
307                 goto done;
308         }
309
310         ret = ltdb_index_one(module, msg, 1);
311         if (ret != LDB_SUCCESS) {
312                 goto done;
313         }
314
315         ret = ltdb_modified(module, msg->dn);
316         if (ret != LDB_SUCCESS) {
317                 goto done;
318         }
319
320 done:
321         return ret;
322 }
323
324 /*
325   add a record to the database
326 */
327 static int ltdb_add(struct ltdb_context *ctx)
328 {
329         struct ldb_module *module = ctx->module;
330         struct ldb_request *req = ctx->req;
331         int ret = LDB_SUCCESS;
332
333         ldb_request_set_state(req, LDB_ASYNC_PENDING);
334
335         if (ltdb_cache_load(module) != 0) {
336                 return LDB_ERR_OPERATIONS_ERROR;
337         }
338
339         ret = ltdb_add_internal(module, req->op.add.message);
340
341         return ret;
342 }
343
344 /*
345   delete a record from the database, not updating indexes (used for deleting
346   index records)
347 */
348 int ltdb_delete_noindex(struct ldb_module *module, struct ldb_dn *dn)
349 {
350         void *data = ldb_module_get_private(module);
351         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
352         TDB_DATA tdb_key;
353         int ret;
354
355         tdb_key = ltdb_key(module, dn);
356         if (!tdb_key.dptr) {
357                 return LDB_ERR_OTHER;
358         }
359
360         ret = tdb_delete(ltdb->tdb, tdb_key);
361         talloc_free(tdb_key.dptr);
362
363         if (ret != 0) {
364                 ret = ltdb_err_map(tdb_error(ltdb->tdb));
365         }
366
367         return ret;
368 }
369
370 static int ltdb_delete_internal(struct ldb_module *module, struct ldb_dn *dn)
371 {
372         struct ldb_message *msg;
373         int ret = LDB_SUCCESS;
374
375         msg = talloc(module, struct ldb_message);
376         if (msg == NULL) {
377                 return LDB_ERR_OPERATIONS_ERROR;
378         }
379
380         /* in case any attribute of the message was indexed, we need
381            to fetch the old record */
382         ret = ltdb_search_dn1(module, dn, msg);
383         if (ret != LDB_SUCCESS) {
384                 /* not finding the old record is an error */
385                 goto done;
386         }
387
388         ret = ltdb_delete_noindex(module, dn);
389         if (ret != LDB_SUCCESS) {
390                 goto done;
391         }
392
393         /* remove one level attribute */
394         ret = ltdb_index_one(module, msg, 0);
395         if (ret != LDB_SUCCESS) {
396                 goto done;
397         }
398
399         /* remove any indexed attributes */
400         ret = ltdb_index_del(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 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         const char *dn;
510         unsigned int i, j;
511
512         dn = ldb_dn_get_linearized(msg->dn);
513         if (dn == NULL) {
514                 return -1;
515         }
516
517         for (i=0;i<msg->num_elements;i++) {
518                 if (ldb_attr_cmp(msg->elements[i].name, name) == 0) {
519                         for (j=0;j<msg->elements[i].num_values;j++) {
520                                 ltdb_index_del_value(module, dn,
521                                                      &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                         /* per definition we find in a canonicalised message an
537                            attribute only once. So we are finished here. */
538                         return 0;
539                 }
540         }
541
542         /* Not found */
543         return -1;
544 }
545
546 /*
547   delete all elements matching an attribute name/value
548
549   return 0 on success, -1 on failure
550 */
551 static int msg_delete_element(struct ldb_module *module,
552                               struct ldb_message *msg,
553                               const char *name,
554                               const struct ldb_val *val)
555 {
556         struct ldb_context *ldb = ldb_module_get_ctx(module);
557         unsigned int i;
558         int found;
559         struct ldb_message_element *el;
560         const struct ldb_schema_attribute *a;
561
562         found = find_element(msg, name);
563         if (found == -1) {
564                 return -1;
565         }
566
567         el = &msg->elements[found];
568
569         a = ldb_schema_attribute_by_name(ldb, el->name);
570
571         for (i=0;i<el->num_values;i++) {
572                 if (a->syntax->comparison_fn(ldb, ldb,
573                                                 &el->values[i], val) == 0) {
574                         if (i<el->num_values-1) {
575                                 memmove(&el->values[i], &el->values[i+1],
576                                         sizeof(el->values[i])*
577                                                 (el->num_values-(i+1)));
578                         }
579                         el->num_values--;
580                         if (el->num_values == 0) {
581                                 return msg_delete_attribute(module, ldb,
582                                                             msg, name);
583                         }
584
585                         /* per definition we find in a canonicalised message an
586                            attribute value only once. So we are finished here */
587                         return 0;
588                 }
589         }
590
591         /* Not found */
592         return -1;
593 }
594
595
596 /*
597   modify a record - internal interface
598
599   yuck - this is O(n^2). Luckily n is usually small so we probably
600   get away with it, but if we ever have really large attribute lists
601   then we'll need to look at this again
602 */
603 int ltdb_modify_internal(struct ldb_module *module,
604                          const struct ldb_message *msg)
605 {
606         struct ldb_context *ldb = ldb_module_get_ctx(module);
607         void *data = ldb_module_get_private(module);
608         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
609         TDB_DATA tdb_key, tdb_data;
610         struct ldb_message *msg2;
611         unsigned i, j;
612         int ret = LDB_SUCCESS, idx;
613
614         tdb_key = ltdb_key(module, msg->dn);
615         if (!tdb_key.dptr) {
616                 return LDB_ERR_OTHER;
617         }
618
619         tdb_data = tdb_fetch(ltdb->tdb, tdb_key);
620         if (!tdb_data.dptr) {
621                 talloc_free(tdb_key.dptr);
622                 return ltdb_err_map(tdb_error(ltdb->tdb));
623         }
624
625         msg2 = talloc(tdb_key.dptr, struct ldb_message);
626         if (msg2 == NULL) {
627                 free(tdb_data.dptr);
628                 ret = LDB_ERR_OTHER;
629                 goto done;
630         }
631
632         ret = ltdb_unpack_data(module, &tdb_data, msg2);
633         free(tdb_data.dptr);
634         if (ret == -1) {
635                 ret = LDB_ERR_OTHER;
636                 goto done;
637         }
638
639         if (!msg2->dn) {
640                 msg2->dn = msg->dn;
641         }
642
643         for (i=0; i<msg->num_elements; i++) {
644                 struct ldb_message_element *el = &msg->elements[i], *el2;
645                 struct ldb_val *vals;
646                 const struct ldb_schema_attribute *a = ldb_schema_attribute_by_name(ldb, el->name);
647                 const char *dn;
648
649                 if (ldb_attr_cmp(el->name, "distinguishedName") == 0) {
650                         ldb_asprintf_errstring(ldb, "it is not permitted to perform a modify on 'distinguishedName' (use rename instead): %s",
651                                                ldb_dn_get_linearized(msg->dn));
652                         ret = LDB_ERR_CONSTRAINT_VIOLATION;
653                         goto done;
654                 }
655
656                 switch (msg->elements[i].flags & LDB_FLAG_MOD_MASK) {
657                 case LDB_FLAG_MOD_ADD:
658                         if (el->num_values == 0) {
659                                 ldb_asprintf_errstring(ldb, "attribute %s on %s specified, but with 0 values (illigal)",
660                                                        el->name, ldb_dn_get_linearized(msg->dn));
661                                 ret = LDB_ERR_CONSTRAINT_VIOLATION;
662                                 goto done;
663                         }
664
665                         if (a && a->flags & LDB_ATTR_FLAG_SINGLE_VALUE) {
666                                 if (el->num_values > 1) {
667                                         ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
668                                                                el->name, ldb_dn_get_linearized(msg->dn));
669                                         ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
670                                         goto done;
671                                 }
672                         }
673
674                         /* Checks if element already exists */
675                         idx = find_element(msg2, el->name);
676                         if (idx == -1) {
677                                 if (msg_add_element(ldb, msg2, el) != 0) {
678                                         ret = LDB_ERR_OTHER;
679                                         goto done;
680                                 }
681                         } else {
682                                 /* We cannot add another value on a existing one
683                                    if the attribute is single-valued */
684                                 if (a && a->flags & LDB_ATTR_FLAG_SINGLE_VALUE) {
685                                         ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
686                                                                el->name, ldb_dn_get_linearized(msg->dn));
687                                         ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
688                                         goto done;
689                                 }
690
691                                 el2 = &(msg2->elements[idx]);
692
693                                 /* Check that values don't exist yet on multi-
694                                    valued attributes or aren't provided twice */
695                                 for (j=0; j<el->num_values; j++) {
696                                         if (ldb_msg_find_val(el2, &el->values[j]) != NULL) {
697                                                 ldb_asprintf_errstring(ldb, "%s: value #%d already exists", el->name, j);
698                                                 ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
699                                                 goto done;
700                                         }
701                                         if (ldb_msg_find_val(el, &el->values[j]) != &el->values[j]) {
702                                                 ldb_asprintf_errstring(ldb, "%s: value #%d provided more than once", el->name, j);
703                                                 ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
704                                                 goto done;
705                                         }
706                                 }
707
708                                 /* Now combine existing and new values to a new
709                                    attribute record */
710                                 vals = talloc_realloc(msg2->elements,
711                                         el2->values, struct ldb_val,
712                                         el2->num_values + el->num_values);
713                                 if (vals == NULL) {
714                                         ldb_oom(ldb);
715                                         ret = LDB_ERR_OTHER;
716                                         goto done;
717                                 }
718
719                                 for (j=0; j<el->num_values; j++) {
720                                         vals[el2->num_values + j] =
721                                                 ldb_val_dup(vals, &el->values[j]);
722                                 }
723
724                                 el2->values = vals;
725                                 el2->num_values += el->num_values;
726                         }
727
728                         break;
729
730                 case LDB_FLAG_MOD_REPLACE:
731                         if (a && a->flags & LDB_ATTR_FLAG_SINGLE_VALUE) {
732                                 if (el->num_values > 1) {
733                                         ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
734                                                                el->name, ldb_dn_get_linearized(msg->dn));
735                                         ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
736                                         goto done;
737                                 }
738                         }
739
740                         for (j=0; j<el->num_values; j++) {
741                                 if (ldb_msg_find_val(el, &el->values[j]) != &el->values[j]) {
742                                         ldb_asprintf_errstring(ldb, "%s: value #%d provided more than once", el->name, j);
743                                         ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
744                                         goto done;
745                                 }
746                         }
747
748                         /* Delete the attribute if it exists in the DB */
749                         msg_delete_attribute(module, ldb, msg2, el->name);
750
751                         /* Recreate it with the new values */
752                         if (msg_add_element(ldb, msg2, el) != 0) {
753                                 ret = LDB_ERR_OTHER;
754                                 goto done;
755                         }
756
757                         break;
758
759                 case LDB_FLAG_MOD_DELETE:
760                         dn = ldb_dn_get_linearized(msg->dn);
761                         if (dn == NULL) {
762                                 ret = LDB_ERR_OTHER;
763                                 goto done;
764                         }
765
766                         if (msg->elements[i].num_values == 0) {
767                                 /* Delete the whole attribute */
768                                 if (msg_delete_attribute(module, ldb, msg2,
769                                                          msg->elements[i].name) != 0) {
770                                         ldb_asprintf_errstring(ldb, "No such attribute: %s for delete on %s",
771                                                                msg->elements[i].name, dn);
772                                         ret = LDB_ERR_NO_SUCH_ATTRIBUTE;
773                                         goto done;
774                                 }
775                         } else {
776                                 /* Delete specified values from an attribute */
777                                 for (j=0; j < msg->elements[i].num_values; j++) {
778                                         if (msg_delete_element(module,
779                                                                msg2,
780                                                                msg->elements[i].name,
781                                                                &msg->elements[i].values[j]) != 0) {
782                                                 ldb_asprintf_errstring(ldb, "No matching attribute value when deleting attribute: %s on %s",
783                                                                        msg->elements[i].name, dn);
784                                                 ret = LDB_ERR_NO_SUCH_ATTRIBUTE;
785                                                 goto done;
786                                         }
787
788                                         ret = ltdb_index_del_value(module, dn,
789                                                 &msg->elements[i], j);
790                                         if (ret != LDB_SUCCESS) {
791                                                 goto done;
792                                         }
793                                 }
794                         }
795
796                         break;
797                 default:
798                         ldb_asprintf_errstring(ldb,
799                                 "Invalid ldb_modify flags on %s: 0x%x",
800                                 msg->elements[i].name,
801                                 msg->elements[i].flags & LDB_FLAG_MOD_MASK);
802                         ret = LDB_ERR_PROTOCOL_ERROR;
803                         goto done;
804                 }
805         }
806
807         ret = ltdb_store(module, msg2, TDB_MODIFY);
808         if (ret != LDB_SUCCESS) {
809                 goto done;
810         }
811
812         ret = ltdb_modified(module, msg->dn);
813         if (ret != LDB_SUCCESS) {
814                 goto done;
815         }
816
817 done:
818         talloc_free(tdb_key.dptr);
819         return ret;
820 }
821
822 /*
823   modify a record
824 */
825 static int ltdb_modify(struct ltdb_context *ctx)
826 {
827         struct ldb_module *module = ctx->module;
828         struct ldb_request *req = ctx->req;
829         int ret = LDB_SUCCESS;
830
831         ret = ltdb_check_special_dn(module, req->op.mod.message);
832         if (ret != LDB_SUCCESS) {
833                 return ret;
834         }
835
836         ldb_request_set_state(req, LDB_ASYNC_PENDING);
837
838         if (ltdb_cache_load(module) != 0) {
839                 return LDB_ERR_OPERATIONS_ERROR;
840         }
841
842         ret = ltdb_modify_internal(module, req->op.mod.message);
843
844         return ret;
845 }
846
847 /*
848   rename a record
849 */
850 static int ltdb_rename(struct ltdb_context *ctx)
851 {
852         struct ldb_module *module = ctx->module;
853         struct ldb_request *req = ctx->req;
854         struct ldb_message *msg;
855         int ret = LDB_SUCCESS;
856
857         ldb_request_set_state(req, LDB_ASYNC_PENDING);
858
859         if (ltdb_cache_load(ctx->module) != 0) {
860                 return LDB_ERR_OPERATIONS_ERROR;
861         }
862
863         msg = talloc(ctx, struct ldb_message);
864         if (msg == NULL) {
865                 ret = LDB_ERR_OPERATIONS_ERROR;
866                 goto done;
867         }
868
869         /* in case any attribute of the message was indexed, we need
870            to fetch the old record */
871         ret = ltdb_search_dn1(module, req->op.rename.olddn, msg);
872         if (ret != LDB_SUCCESS) {
873                 /* not finding the old record is an error */
874                 goto done;
875         }
876
877         msg->dn = ldb_dn_copy(msg, req->op.rename.newdn);
878         if (msg->dn == NULL) {
879                 ret = LDB_ERR_OPERATIONS_ERROR;
880                 goto done;
881         }
882
883         /* Always delete first then add, to avoid conflicts with
884          * unique indexes. We rely on the transaction to make this
885          * atomic
886          */
887         ret = ltdb_delete_internal(module, req->op.rename.olddn);
888         if (ret != LDB_SUCCESS) {
889                 goto done;
890         }
891
892         ret = ltdb_add_internal(module, msg);
893
894 done:
895         return ret;
896 }
897
898 static int ltdb_start_trans(struct ldb_module *module)
899 {
900         void *data = ldb_module_get_private(module);
901         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
902
903         if (tdb_transaction_start(ltdb->tdb) != 0) {
904                 return ltdb_err_map(tdb_error(ltdb->tdb));
905         }
906
907         ltdb->in_transaction++;
908
909         ltdb_index_transaction_start(module);
910
911         return LDB_SUCCESS;
912 }
913
914 static int ltdb_prepare_commit(struct ldb_module *module)
915 {
916         void *data = ldb_module_get_private(module);
917         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
918
919         if (ltdb->in_transaction != 1) {
920                 return LDB_SUCCESS;
921         }
922
923         if (ltdb_index_transaction_commit(module) != 0) {
924                 tdb_transaction_cancel(ltdb->tdb);
925                 ltdb->in_transaction--;
926                 return ltdb_err_map(tdb_error(ltdb->tdb));
927         }
928
929         if (tdb_transaction_prepare_commit(ltdb->tdb) != 0) {
930                 ltdb->in_transaction--;
931                 return ltdb_err_map(tdb_error(ltdb->tdb));
932         }
933
934         ltdb->prepared_commit = true;
935
936         return LDB_SUCCESS;
937 }
938
939 static int ltdb_end_trans(struct ldb_module *module)
940 {
941         void *data = ldb_module_get_private(module);
942         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
943
944         if (!ltdb->prepared_commit) {
945                 int ret = ltdb_prepare_commit(module);
946                 if (ret != LDB_SUCCESS) {
947                         return ret;
948                 }
949         }
950
951         ltdb->in_transaction--;
952         ltdb->prepared_commit = false;
953
954         if (tdb_transaction_commit(ltdb->tdb) != 0) {
955                 return ltdb_err_map(tdb_error(ltdb->tdb));
956         }
957
958         return LDB_SUCCESS;
959 }
960
961 static int ltdb_del_trans(struct ldb_module *module)
962 {
963         void *data = ldb_module_get_private(module);
964         struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
965
966         ltdb->in_transaction--;
967
968         if (ltdb_index_transaction_cancel(module) != 0) {
969                 tdb_transaction_cancel(ltdb->tdb);
970                 return ltdb_err_map(tdb_error(ltdb->tdb));
971         }
972
973         if (tdb_transaction_cancel(ltdb->tdb) != 0) {
974                 return ltdb_err_map(tdb_error(ltdb->tdb));
975         }
976
977         return LDB_SUCCESS;
978 }
979
980 /*
981   return sequenceNumber from @BASEINFO
982 */
983 static int ltdb_sequence_number(struct ltdb_context *ctx,
984                                 struct ldb_extended **ext)
985 {
986         struct ldb_context *ldb;
987         struct ldb_module *module = ctx->module;
988         struct ldb_request *req = ctx->req;
989         TALLOC_CTX *tmp_ctx;
990         struct ldb_seqnum_request *seq;
991         struct ldb_seqnum_result *res;
992         struct ldb_message *msg = NULL;
993         struct ldb_dn *dn;
994         const char *date;
995         int ret = LDB_SUCCESS;
996
997         ldb = ldb_module_get_ctx(module);
998
999         seq = talloc_get_type(req->op.extended.data,
1000                                 struct ldb_seqnum_request);
1001         if (seq == NULL) {
1002                 return LDB_ERR_OPERATIONS_ERROR;
1003         }
1004
1005         ldb_request_set_state(req, LDB_ASYNC_PENDING);
1006
1007         if (ltdb_lock_read(module) != 0) {
1008                 return LDB_ERR_OPERATIONS_ERROR;
1009         }
1010
1011         res = talloc_zero(req, struct ldb_seqnum_result);
1012         if (res == NULL) {
1013                 ret = LDB_ERR_OPERATIONS_ERROR;
1014                 goto done;
1015         }
1016         tmp_ctx = talloc_new(req);
1017         if (tmp_ctx == NULL) {
1018                 ret = LDB_ERR_OPERATIONS_ERROR;
1019                 goto done;
1020         }
1021
1022         dn = ldb_dn_new(tmp_ctx, ldb, LTDB_BASEINFO);
1023
1024         msg = talloc(tmp_ctx, struct ldb_message);
1025         if (msg == NULL) {
1026                 ret = LDB_ERR_OPERATIONS_ERROR;
1027                 goto done;
1028         }
1029
1030         ret = ltdb_search_dn1(module, dn, msg);
1031         if (ret != LDB_SUCCESS) {
1032                 goto done;
1033         }
1034
1035         switch (seq->type) {
1036         case LDB_SEQ_HIGHEST_SEQ:
1037                 res->seq_num = ldb_msg_find_attr_as_uint64(msg, LTDB_SEQUENCE_NUMBER, 0);
1038                 break;
1039         case LDB_SEQ_NEXT:
1040                 res->seq_num = ldb_msg_find_attr_as_uint64(msg, LTDB_SEQUENCE_NUMBER, 0);
1041                 res->seq_num++;
1042                 break;
1043         case LDB_SEQ_HIGHEST_TIMESTAMP:
1044                 date = ldb_msg_find_attr_as_string(msg, LTDB_MOD_TIMESTAMP, NULL);
1045                 if (date) {
1046                         res->seq_num = ldb_string_to_time(date);
1047                 } else {
1048                         res->seq_num = 0;
1049                         /* zero is as good as anything when we don't know */
1050                 }
1051                 break;
1052         }
1053
1054         *ext = talloc_zero(req, struct ldb_extended);
1055         if (*ext == NULL) {
1056                 ret = LDB_ERR_OPERATIONS_ERROR;
1057                 goto done;
1058         }
1059         (*ext)->oid = LDB_EXTENDED_SEQUENCE_NUMBER;
1060         (*ext)->data = talloc_steal(*ext, res);
1061
1062 done:
1063         talloc_free(tmp_ctx);
1064         ltdb_unlock_read(module);
1065         return ret;
1066 }
1067
1068 static void ltdb_request_done(struct ltdb_context *ctx, int error)
1069 {
1070         struct ldb_context *ldb;
1071         struct ldb_request *req;
1072         struct ldb_reply *ares;
1073
1074         ldb = ldb_module_get_ctx(ctx->module);
1075         req = ctx->req;
1076
1077         /* if we already returned an error just return */
1078         if (ldb_request_get_status(req) != LDB_SUCCESS) {
1079                 return;
1080         }
1081
1082         ares = talloc_zero(req, struct ldb_reply);
1083         if (!ares) {
1084                 ldb_oom(ldb);
1085                 req->callback(req, NULL);
1086                 return;
1087         }
1088         ares->type = LDB_REPLY_DONE;
1089         ares->error = error;
1090
1091         req->callback(req, ares);
1092 }
1093
1094 static void ltdb_timeout(struct tevent_context *ev,
1095                           struct tevent_timer *te,
1096                           struct timeval t,
1097                           void *private_data)
1098 {
1099         struct ltdb_context *ctx;
1100         ctx = talloc_get_type(private_data, struct ltdb_context);
1101
1102         if (!ctx->request_terminated) {
1103                 /* request is done now */
1104                 ltdb_request_done(ctx, LDB_ERR_TIME_LIMIT_EXCEEDED);
1105         }
1106
1107         if (!ctx->request_terminated) {
1108                 /* neutralize the spy */
1109                 ctx->spy->ctx = NULL;
1110         }
1111         talloc_free(ctx);
1112 }
1113
1114 static void ltdb_request_extended_done(struct ltdb_context *ctx,
1115                                         struct ldb_extended *ext,
1116                                         int error)
1117 {
1118         struct ldb_context *ldb;
1119         struct ldb_request *req;
1120         struct ldb_reply *ares;
1121
1122         ldb = ldb_module_get_ctx(ctx->module);
1123         req = ctx->req;
1124
1125         /* if we already returned an error just return */
1126         if (ldb_request_get_status(req) != LDB_SUCCESS) {
1127                 return;
1128         }
1129
1130         ares = talloc_zero(req, struct ldb_reply);
1131         if (!ares) {
1132                 ldb_oom(ldb);
1133                 req->callback(req, NULL);
1134                 return;
1135         }
1136         ares->type = LDB_REPLY_DONE;
1137         ares->response = ext;
1138         ares->error = error;
1139
1140         req->callback(req, ares);
1141 }
1142
1143 static void ltdb_handle_extended(struct ltdb_context *ctx)
1144 {
1145         struct ldb_extended *ext = NULL;
1146         int ret;
1147
1148         if (strcmp(ctx->req->op.extended.oid,
1149                    LDB_EXTENDED_SEQUENCE_NUMBER) == 0) {
1150                 /* get sequence number */
1151                 ret = ltdb_sequence_number(ctx, &ext);
1152         } else {
1153                 /* not recognized */
1154                 ret = LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
1155         }
1156
1157         ltdb_request_extended_done(ctx, ext, ret);
1158 }
1159
1160 static void ltdb_callback(struct tevent_context *ev,
1161                           struct tevent_timer *te,
1162                           struct timeval t,
1163                           void *private_data)
1164 {
1165         struct ltdb_context *ctx;
1166         int ret;
1167
1168         ctx = talloc_get_type(private_data, struct ltdb_context);
1169
1170         if (ctx->request_terminated) {
1171                 goto done;
1172         }
1173
1174         switch (ctx->req->operation) {
1175         case LDB_SEARCH:
1176                 ret = ltdb_search(ctx);
1177                 break;
1178         case LDB_ADD:
1179                 ret = ltdb_add(ctx);
1180                 break;
1181         case LDB_MODIFY:
1182                 ret = ltdb_modify(ctx);
1183                 break;
1184         case LDB_DELETE:
1185                 ret = ltdb_delete(ctx);
1186                 break;
1187         case LDB_RENAME:
1188                 ret = ltdb_rename(ctx);
1189                 break;
1190         case LDB_EXTENDED:
1191                 ltdb_handle_extended(ctx);
1192                 goto done;
1193         default:
1194                 /* no other op supported */
1195                 ret = LDB_ERR_UNWILLING_TO_PERFORM;
1196         }
1197
1198         if (!ctx->request_terminated) {
1199                 /* request is done now */
1200                 ltdb_request_done(ctx, ret);
1201         }
1202
1203 done:
1204         if (!ctx->request_terminated) {
1205                 /* neutralize the spy */
1206                 ctx->spy->ctx = NULL;
1207         }
1208         talloc_free(ctx);
1209 }
1210
1211 static int ltdb_request_destructor(void *ptr)
1212 {
1213         struct ltdb_req_spy *spy = talloc_get_type(ptr, struct ltdb_req_spy);
1214
1215         if (spy->ctx != NULL) {
1216                 spy->ctx->request_terminated = true;
1217         }
1218
1219         return 0;
1220 }
1221
1222 static int ltdb_handle_request(struct ldb_module *module,
1223                                struct ldb_request *req)
1224 {
1225         struct ldb_context *ldb;
1226         struct tevent_context *ev;
1227         struct ltdb_context *ac;
1228         struct tevent_timer *te;
1229         struct timeval tv;
1230
1231         if (check_critical_controls(req->controls)) {
1232                 return LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
1233         }
1234
1235         ldb = ldb_module_get_ctx(module);
1236
1237         if (req->starttime == 0 || req->timeout == 0) {
1238                 ldb_set_errstring(ldb, "Invalid timeout settings");
1239                 return LDB_ERR_TIME_LIMIT_EXCEEDED;
1240         }
1241
1242         ev = ldb_get_event_context(ldb);
1243
1244         ac = talloc_zero(ldb, struct ltdb_context);
1245         if (ac == NULL) {
1246                 ldb_oom(ldb);
1247                 return LDB_ERR_OPERATIONS_ERROR;
1248         }
1249
1250         ac->module = module;
1251         ac->req = req;
1252
1253         tv.tv_sec = 0;
1254         tv.tv_usec = 0;
1255         te = tevent_add_timer(ev, ac, tv, ltdb_callback, ac);
1256         if (NULL == te) {
1257                 talloc_free(ac);
1258                 return LDB_ERR_OPERATIONS_ERROR;
1259         }
1260
1261         tv.tv_sec = req->starttime + req->timeout;
1262         ac->timeout_event = tevent_add_timer(ev, ac, tv, ltdb_timeout, ac);
1263         if (NULL == ac->timeout_event) {
1264                 talloc_free(ac);
1265                 return LDB_ERR_OPERATIONS_ERROR;
1266         }
1267
1268         /* set a spy so that we do not try to use the request context
1269          * if it is freed before ltdb_callback fires */
1270         ac->spy = talloc(req, struct ltdb_req_spy);
1271         if (NULL == ac->spy) {
1272                 talloc_free(ac);
1273                 return LDB_ERR_OPERATIONS_ERROR;
1274         }
1275         ac->spy->ctx = ac;
1276
1277         talloc_set_destructor((TALLOC_CTX *)ac->spy, ltdb_request_destructor);
1278
1279         return LDB_SUCCESS;
1280 }
1281
1282 static const struct ldb_module_ops ltdb_ops = {
1283         .name              = "tdb",
1284         .search            = ltdb_handle_request,
1285         .add               = ltdb_handle_request,
1286         .modify            = ltdb_handle_request,
1287         .del               = ltdb_handle_request,
1288         .rename            = ltdb_handle_request,
1289         .extended          = ltdb_handle_request,
1290         .start_transaction = ltdb_start_trans,
1291         .end_transaction   = ltdb_end_trans,
1292         .prepare_commit    = ltdb_prepare_commit,
1293         .del_transaction   = ltdb_del_trans,
1294 };
1295
1296 /*
1297   connect to the database
1298 */
1299 static int ltdb_connect(struct ldb_context *ldb, const char *url,
1300                         unsigned int flags, const char *options[],
1301                         struct ldb_module **_module)
1302 {
1303         struct ldb_module *module;
1304         const char *path;
1305         int tdb_flags, open_flags;
1306         struct ltdb_private *ltdb;
1307
1308         /* parse the url */
1309         if (strchr(url, ':')) {
1310                 if (strncmp(url, "tdb://", 6) != 0) {
1311                         ldb_debug(ldb, LDB_DEBUG_ERROR,
1312                                   "Invalid tdb URL '%s'", url);
1313                         return -1;
1314                 }
1315                 path = url+6;
1316         } else {
1317                 path = url;
1318         }
1319
1320         tdb_flags = TDB_DEFAULT | TDB_SEQNUM;
1321
1322         /* check for the 'nosync' option */
1323         if (flags & LDB_FLG_NOSYNC) {
1324                 tdb_flags |= TDB_NOSYNC;
1325         }
1326
1327         /* and nommap option */
1328         if (flags & LDB_FLG_NOMMAP) {
1329                 tdb_flags |= TDB_NOMMAP;
1330         }
1331
1332         if (flags & LDB_FLG_RDONLY) {
1333                 open_flags = O_RDONLY;
1334         } else {
1335                 open_flags = O_CREAT | O_RDWR;
1336         }
1337
1338         ltdb = talloc_zero(ldb, struct ltdb_private);
1339         if (!ltdb) {
1340                 ldb_oom(ldb);
1341                 return -1;
1342         }
1343
1344         /* note that we use quite a large default hash size */
1345         ltdb->tdb = ltdb_wrap_open(ltdb, path, 10000,
1346                                    tdb_flags, open_flags,
1347                                    ldb_get_create_perms(ldb), ldb);
1348         if (!ltdb->tdb) {
1349                 ldb_debug(ldb, LDB_DEBUG_ERROR,
1350                           "Unable to open tdb '%s'", path);
1351                 talloc_free(ltdb);
1352                 return -1;
1353         }
1354
1355         ltdb->sequence_number = 0;
1356
1357         module = ldb_module_new(ldb, ldb, "ldb_tdb backend", &ltdb_ops);
1358         if (!module) {
1359                 talloc_free(ltdb);
1360                 return -1;
1361         }
1362         ldb_module_set_private(module, ltdb);
1363
1364         if (ltdb_cache_load(module) != 0) {
1365                 talloc_free(module);
1366                 talloc_free(ltdb);
1367                 return -1;
1368         }
1369
1370         *_module = module;
1371         return 0;
1372 }
1373
1374 const struct ldb_backend_ops ldb_tdb_backend_ops = {
1375         .name = "tdb",
1376         .connect_fn = ltdb_connect
1377 };