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