LDB ASYNC: core modules
[kai/samba.git] / source4 / lib / ldb / ldb_map / ldb_map_outbound.c
1 /*
2    ldb database mapping module
3
4    Copyright (C) Jelmer Vernooij 2005
5    Copyright (C) Martin Kuehl <mkhl@samba.org> 2006
6    Copyright (C) Andrew Bartlett <abartlet@samba.org> 2006
7    Copyright (C) Simo Sorce <idra@samba.org> 2008
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 #include "ldb_includes.h"
29
30 #include "ldb_map.h"
31 #include "ldb_map_private.h"
32
33
34 /* Mapping attributes
35  * ================== */
36
37 /* Select attributes that stay in the local partition. */
38 static const char **map_attrs_select_local(struct ldb_module *module, void *mem_ctx, const char * const *attrs)
39 {
40         const struct ldb_map_context *data = map_get_context(module);
41         const char **result;
42         int i, last;
43
44         if (attrs == NULL)
45                 return NULL;
46
47         last = 0;
48         result = talloc_array(mem_ctx, const char *, 1);
49         if (result == NULL) {
50                 goto failed;
51         }
52         result[0] = NULL;
53
54         for (i = 0; attrs[i]; i++) {
55                 /* Wildcards and ignored attributes are kept locally */
56                 if ((ldb_attr_cmp(attrs[i], "*") == 0) ||
57                     (!map_attr_check_remote(data, attrs[i]))) {
58                         result = talloc_realloc(mem_ctx, result, const char *, last+2);
59                         if (result == NULL) {
60                                 goto failed;
61                         }
62
63                         result[last] = talloc_strdup(result, attrs[i]);
64                         result[last+1] = NULL;
65                         last++;
66                 }
67         }
68
69         return result;
70
71 failed:
72         talloc_free(result);
73         map_oom(module);
74         return NULL;
75 }
76
77 /* Collect attributes that are mapped into the remote partition. */
78 static const char **map_attrs_collect_remote(struct ldb_module *module, void *mem_ctx, 
79                                              const char * const *attrs)
80 {
81         const struct ldb_map_context *data = map_get_context(module);
82         const char **result;
83         const struct ldb_map_attribute *map;
84         const char *name=NULL;
85         int i, j, last;
86         int ret;
87
88         last = 0;
89         result = talloc_array(mem_ctx, const char *, 1);
90         if (result == NULL) {
91                 goto failed;
92         }
93         result[0] = NULL;
94
95         for (i = 0; attrs[i]; i++) {
96                 /* Wildcards are kept remotely, too */
97                 if (ldb_attr_cmp(attrs[i], "*") == 0) {
98                         const char **new_attrs = NULL;
99                         ret = map_attrs_merge(module, mem_ctx, &new_attrs, attrs);
100                         if (ret != LDB_SUCCESS) {
101                                 goto failed;
102                         }
103                         ret = map_attrs_merge(module, mem_ctx, &new_attrs, data->wildcard_attributes);
104                         if (ret != LDB_SUCCESS) {
105                                 goto failed;
106                         }
107
108                         attrs = new_attrs;
109                         break;
110                 }
111         }
112
113         for (i = 0; attrs[i]; i++) {
114                 /* Wildcards are kept remotely, too */
115                 if (ldb_attr_cmp(attrs[i], "*") == 0) {
116                         /* Add all 'include in wildcard' attributes */
117                         name = attrs[i];
118                         goto named;
119                 }
120
121                 /* Add remote names of mapped attrs */
122                 map = map_attr_find_local(data, attrs[i]);
123                 if (map == NULL) {
124                         continue;
125                 }
126
127                 switch (map->type) {
128                 case MAP_IGNORE:
129                         continue;
130
131                 case MAP_KEEP:
132                         name = attrs[i];
133                         goto named;
134
135                 case MAP_RENAME:
136                 case MAP_CONVERT:
137                         name = map->u.rename.remote_name;
138                         goto named;
139
140                 case MAP_GENERATE:
141                         /* Add all remote names of "generate" attrs */
142                         for (j = 0; map->u.generate.remote_names[j]; j++) {
143                                 result = talloc_realloc(mem_ctx, result, const char *, last+2);
144                                 if (result == NULL) {
145                                         goto failed;
146                                 }
147
148                                 result[last] = talloc_strdup(result, map->u.generate.remote_names[j]);
149                                 result[last+1] = NULL;
150                                 last++;
151                         }
152                         continue;
153                 }
154
155         named:  /* We found a single remote name, add that */
156                 result = talloc_realloc(mem_ctx, result, const char *, last+2);
157                 if (result == NULL) {
158                         goto failed;
159                 }
160
161                 result[last] = talloc_strdup(result, name);
162                 result[last+1] = NULL;
163                 last++;
164         }
165
166         return result;
167
168 failed:
169         talloc_free(result);
170         map_oom(module);
171         return NULL;
172 }
173
174 /* Split attributes that stay in the local partition from those that
175  * are mapped into the remote partition. */
176 static int map_attrs_partition(struct ldb_module *module, void *mem_ctx, const char ***local_attrs, const char ***remote_attrs, const char * const *attrs)
177 {
178         *local_attrs = map_attrs_select_local(module, mem_ctx, attrs);
179         *remote_attrs = map_attrs_collect_remote(module, mem_ctx, attrs);
180
181         return 0;
182 }
183
184 /* Mapping message elements
185  * ======================== */
186
187 /* Add an element to a message, overwriting any old identically named elements. */
188 static int ldb_msg_replace(struct ldb_message *msg, const struct ldb_message_element *el)
189 {
190         struct ldb_message_element *old;
191
192         old = ldb_msg_find_element(msg, el->name);
193
194         /* no local result, add as new element */
195         if (old == NULL) {
196                 if (ldb_msg_add_empty(msg, el->name, 0, &old) != 0) {
197                         return -1;
198                 }
199                 talloc_free(discard_const_p(char, old->name));
200         }
201
202         /* copy new element */
203         *old = *el;
204
205         /* and make sure we reference the contents */
206         if (!talloc_reference(msg->elements, el->name)) {
207                 return -1;
208         }
209         if (!talloc_reference(msg->elements, el->values)) {
210                 return -1;
211         }
212
213         return 0;
214 }
215
216 /* Map a message element back into the local partition. */
217 static struct ldb_message_element *ldb_msg_el_map_remote(struct ldb_module *module, 
218                                                          void *mem_ctx, 
219                                                          const struct ldb_map_attribute *map, 
220                                                          const char *attr_name,
221                                                          const struct ldb_message_element *old)
222 {
223         struct ldb_message_element *el;
224         int i;
225
226         el = talloc_zero(mem_ctx, struct ldb_message_element);
227         if (el == NULL) {
228                 map_oom(module);
229                 return NULL;
230         }
231
232         el->values = talloc_array(el, struct ldb_val, old->num_values);
233         if (el->values == NULL) {
234                 talloc_free(el);
235                 map_oom(module);
236                 return NULL;
237         }
238
239         el->name = talloc_strdup(el, attr_name);
240         if (el->name == NULL) {
241                 talloc_free(el);
242                 map_oom(module);
243                 return NULL;
244         }
245
246         for (i = 0; i < old->num_values; i++) {
247                 el->values[i] = ldb_val_map_remote(module, el->values, map, &old->values[i]);
248                 /* Conversions might fail, in which case bail */
249                 if (!el->values[i].data) {
250                         talloc_free(el);
251                         return NULL;
252                 }
253                 el->num_values++;
254         }
255
256         return el;
257 }
258
259 /* Merge a remote message element into a local message. */
260 static int ldb_msg_el_merge(struct ldb_module *module, struct ldb_message *local, 
261                             struct ldb_message *remote, const char *attr_name)
262 {
263         const struct ldb_map_context *data = map_get_context(module);
264         const struct ldb_map_attribute *map;
265         struct ldb_message_element *old, *el=NULL;
266         const char *remote_name = NULL;
267
268         /* We handle wildcards in ldb_msg_el_merge_wildcard */
269         if (ldb_attr_cmp(attr_name, "*") == 0) {
270                 return LDB_SUCCESS;
271         }
272
273         map = map_attr_find_local(data, attr_name);
274
275         /* Unknown attribute in remote message:
276          * skip, attribute was probably auto-generated */
277         if (map == NULL) {
278                 return LDB_SUCCESS;
279         }
280
281         switch (map->type) {
282         case MAP_IGNORE:
283                 break;
284         case MAP_CONVERT:
285                 remote_name = map->u.convert.remote_name;
286                 break;
287         case MAP_KEEP:
288                 remote_name = attr_name;
289                 break;
290         case MAP_RENAME:
291                 remote_name = map->u.rename.remote_name;
292                 break;
293         case MAP_GENERATE:
294                 break;
295         }
296
297         switch (map->type) {
298         case MAP_IGNORE:
299                 return LDB_SUCCESS;
300
301         case MAP_CONVERT:
302                 if (map->u.convert.convert_remote == NULL) {
303                         ldb_debug(module->ldb, LDB_DEBUG_ERROR, "ldb_map: "
304                                   "Skipping attribute '%s': "
305                                   "'convert_remote' not set\n",
306                                   attr_name);
307                         return LDB_SUCCESS;
308                 }
309                 /* fall through */
310         case MAP_KEEP:
311         case MAP_RENAME:
312                 old = ldb_msg_find_element(remote, remote_name);
313                 if (old) {
314                         el = ldb_msg_el_map_remote(module, local, map, attr_name, old);
315                 } else {
316                         return LDB_ERR_NO_SUCH_ATTRIBUTE;
317                 }
318                 break;
319
320         case MAP_GENERATE:
321                 if (map->u.generate.generate_local == NULL) {
322                         ldb_debug(module->ldb, LDB_DEBUG_ERROR, "ldb_map: "
323                                   "Skipping attribute '%s': "
324                                   "'generate_local' not set\n",
325                                   attr_name);
326                         return LDB_SUCCESS;
327                 }
328
329                 el = map->u.generate.generate_local(module, local, attr_name, remote);
330                 if (!el) {
331                         /* Generation failure is probably due to lack of source attributes */
332                         return LDB_ERR_NO_SUCH_ATTRIBUTE;
333                 }
334                 break;
335         }
336
337         if (el == NULL) {
338                 return LDB_ERR_NO_SUCH_ATTRIBUTE;
339         }
340
341         return ldb_msg_replace(local, el);
342 }
343
344 /* Handle wildcard parts of merging a remote message element into a local message. */
345 static int ldb_msg_el_merge_wildcard(struct ldb_module *module, struct ldb_message *local, 
346                                      struct ldb_message *remote)
347 {
348         const struct ldb_map_context *data = map_get_context(module);
349         const struct ldb_map_attribute *map = map_attr_find_local(data, "*");
350         struct ldb_message_element *el=NULL;
351         int i, ret;
352
353         /* Perhaps we have a mapping for "*" */
354         if (map && map->type == MAP_KEEP) {
355                 /* We copy everything over, and hope that anything with a 
356                    more specific rule is overwritten */
357                 for (i = 0; i < remote->num_elements; i++) {
358                         el = ldb_msg_el_map_remote(module, local, map, remote->elements[i].name,
359                                                    &remote->elements[i]);
360                         if (el == NULL) {
361                                 return LDB_ERR_OPERATIONS_ERROR;
362                         }
363                         
364                         ret = ldb_msg_replace(local, el);
365                         if (ret) {
366                                 return ret;
367                         }
368                 }
369         }
370         
371         /* Now walk the list of possible mappings, and apply each */
372         for (i = 0; data->attribute_maps[i].local_name; i++) {
373                 ret = ldb_msg_el_merge(module, local, remote, 
374                                        data->attribute_maps[i].local_name);
375                 if (ret == LDB_ERR_NO_SUCH_ATTRIBUTE) {
376                         continue;
377                 } else if (ret) {
378                         return ret;
379                 } else {
380                         continue;
381                 }
382         }
383
384         return LDB_SUCCESS;
385 }
386
387 /* Mapping messages
388  * ================ */
389
390 /* Merge two local messages into a single one. */
391 static int ldb_msg_merge_local(struct ldb_module *module, struct ldb_message *msg1, struct ldb_message *msg2)
392 {
393         int i, ret;
394
395         for (i = 0; i < msg2->num_elements; i++) {
396                 ret = ldb_msg_replace(msg1, &msg2->elements[i]);
397                 if (ret) {
398                         return ret;
399                 }
400         }
401
402         return LDB_SUCCESS;
403 }
404
405 /* Merge a local and a remote message into a single local one. */
406 static int ldb_msg_merge_remote(struct map_context *ac, struct ldb_message *local, 
407                                 struct ldb_message *remote)
408 {
409         int i, ret;
410         const char * const *attrs = ac->all_attrs;
411         if (!attrs) {
412                 ret = ldb_msg_el_merge_wildcard(ac->module, local, remote);
413                 if (ret) {
414                         return ret;
415                 }
416         }
417
418         for (i = 0; attrs && attrs[i]; i++) {
419                 if (ldb_attr_cmp(attrs[i], "*") == 0) {
420                         ret = ldb_msg_el_merge_wildcard(ac->module, local, remote);
421                         if (ret) {
422                                 return ret;
423                         }
424                         break;
425                 }
426         }
427
428         /* Try to map each attribute back;
429          * Add to local message is possible,
430          * Overwrite old local attribute if necessary */
431         for (i = 0; attrs && attrs[i]; i++) {
432                 ret = ldb_msg_el_merge(ac->module, local, remote, 
433                                        attrs[i]);
434                 if (ret == LDB_ERR_NO_SUCH_ATTRIBUTE) {
435                 } else if (ret) {
436                         return ret;
437                 }
438         }
439
440         return LDB_SUCCESS;
441 }
442
443 /* Mapping search results
444  * ====================== */
445
446 /* Map a search result back into the local partition. */
447 static int map_reply_remote(struct map_context *ac, struct ldb_reply *ares)
448 {
449         struct ldb_message *msg;
450         struct ldb_dn *dn;
451         int ret;
452
453         /* There is no result message, skip */
454         if (ares->type != LDB_REPLY_ENTRY) {
455                 return 0;
456         }
457
458         /* Create a new result message */
459         msg = ldb_msg_new(ares);
460         if (msg == NULL) {
461                 map_oom(ac->module);
462                 return -1;
463         }
464
465         /* Merge remote message into new message */
466         ret = ldb_msg_merge_remote(ac, msg, ares->message);
467         if (ret) {
468                 talloc_free(msg);
469                 return ret;
470         }
471
472         /* Create corresponding local DN */
473         dn = ldb_dn_map_rebase_remote(ac->module, msg, ares->message->dn);
474         if (dn == NULL) {
475                 talloc_free(msg);
476                 return -1;
477         }
478         msg->dn = dn;
479
480         /* Store new message with new DN as the result */
481         talloc_free(ares->message);
482         ares->message = msg;
483
484         return 0;
485 }
486
487 /* Mapping parse trees
488  * =================== */
489
490 /* Check whether a parse tree can safely be split in two. */
491 static bool ldb_parse_tree_check_splittable(const struct ldb_parse_tree *tree)
492 {
493         const struct ldb_parse_tree *subtree = tree;
494         bool negate = false;
495
496         while (subtree) {
497                 switch (subtree->operation) {
498                 case LDB_OP_NOT:
499                         negate = !negate;
500                         subtree = subtree->u.isnot.child;
501                         continue;
502
503                 case LDB_OP_AND:
504                         return !negate; /* if negate: False */
505
506                 case LDB_OP_OR:
507                         return negate;  /* if negate: True */
508
509                 default:
510                         return true;    /* simple parse tree */
511                 }
512         }
513
514         return true;                    /* no parse tree */
515 }
516
517 /* Collect a list of attributes required to match a given parse tree. */
518 static int ldb_parse_tree_collect_attrs(struct ldb_module *module, void *mem_ctx, const char ***attrs, const struct ldb_parse_tree *tree)
519 {
520         const char **new_attrs;
521         int i, ret;
522
523         if (tree == NULL) {
524                 return 0;
525         }
526
527         switch (tree->operation) {
528         case LDB_OP_OR:
529         case LDB_OP_AND:                /* attributes stored in list of subtrees */
530                 for (i = 0; i < tree->u.list.num_elements; i++) {
531                         ret = ldb_parse_tree_collect_attrs(module, mem_ctx, 
532                                                            attrs, tree->u.list.elements[i]);
533                         if (ret) {
534                                 return ret;
535                         }
536                 }
537                 return 0;
538
539         case LDB_OP_NOT:                /* attributes stored in single subtree */
540                 return ldb_parse_tree_collect_attrs(module, mem_ctx, attrs, tree->u.isnot.child);
541
542         default:                        /* single attribute in tree */
543                 new_attrs = ldb_attr_list_copy_add(mem_ctx, *attrs, tree->u.equality.attr);
544                 talloc_free(*attrs);
545                 *attrs = new_attrs;
546                 return 0;
547         }
548
549         return -1;
550 }
551
552 static int map_subtree_select_local(struct ldb_module *module, void *mem_ctx, struct ldb_parse_tree **new, const struct ldb_parse_tree *tree);
553
554 /* Select a negated subtree that queries attributes in the local partition */
555 static int map_subtree_select_local_not(struct ldb_module *module, void *mem_ctx, struct ldb_parse_tree **new, const struct ldb_parse_tree *tree)
556 {
557         struct ldb_parse_tree *child;
558         int ret;
559
560         /* Prepare new tree */
561         *new = talloc_memdup(mem_ctx, tree, sizeof(struct ldb_parse_tree));
562         if (*new == NULL) {
563                 map_oom(module);
564                 return -1;
565         }
566
567         /* Generate new subtree */
568         ret = map_subtree_select_local(module, *new, &child, tree->u.isnot.child);
569         if (ret) {
570                 talloc_free(*new);
571                 return ret;
572         }
573
574         /* Prune tree without subtree */
575         if (child == NULL) {
576                 talloc_free(*new);
577                 *new = NULL;
578                 return 0;
579         }
580
581         (*new)->u.isnot.child = child;
582
583         return ret;
584 }
585
586 /* Select a list of subtrees that query attributes in the local partition */
587 static int map_subtree_select_local_list(struct ldb_module *module, void *mem_ctx, struct ldb_parse_tree **new, const struct ldb_parse_tree *tree)
588 {
589         int i, j, ret=0;
590
591         /* Prepare new tree */
592         *new = talloc_memdup(mem_ctx, tree, sizeof(struct ldb_parse_tree));
593         if (*new == NULL) {
594                 map_oom(module);
595                 return -1;
596         }
597
598         /* Prepare list of subtrees */
599         (*new)->u.list.num_elements = 0;
600         (*new)->u.list.elements = talloc_array(*new, struct ldb_parse_tree *, tree->u.list.num_elements);
601         if ((*new)->u.list.elements == NULL) {
602                 map_oom(module);
603                 talloc_free(*new);
604                 return -1;
605         }
606
607         /* Generate new list of subtrees */
608         j = 0;
609         for (i = 0; i < tree->u.list.num_elements; i++) {
610                 struct ldb_parse_tree *child;
611                 ret = map_subtree_select_local(module, *new, &child, tree->u.list.elements[i]);
612                 if (ret) {
613                         talloc_free(*new);
614                         return ret;
615                 }
616
617                 if (child) {
618                         (*new)->u.list.elements[j] = child;
619                         j++;
620                 }
621         }
622
623         /* Prune tree without subtrees */
624         if (j == 0) {
625                 talloc_free(*new);
626                 *new = NULL;
627                 return 0;
628         }
629
630         /* Fix subtree list size */
631         (*new)->u.list.num_elements = j;
632         (*new)->u.list.elements = talloc_realloc(*new, (*new)->u.list.elements, struct ldb_parse_tree *, (*new)->u.list.num_elements);
633
634         return ret;
635 }
636
637 /* Select a simple subtree that queries attributes in the local partition */
638 static int map_subtree_select_local_simple(struct ldb_module *module, void *mem_ctx, struct ldb_parse_tree **new, const struct ldb_parse_tree *tree)
639 {
640         /* Prepare new tree */
641         *new = talloc_memdup(mem_ctx, tree, sizeof(struct ldb_parse_tree));
642         if (*new == NULL) {
643                 map_oom(module);
644                 return -1;
645         }
646
647         return 0;
648 }
649
650 /* Select subtrees that query attributes in the local partition */
651 static int map_subtree_select_local(struct ldb_module *module, void *mem_ctx, struct ldb_parse_tree **new, const struct ldb_parse_tree *tree)
652 {
653         const struct ldb_map_context *data = map_get_context(module);
654
655         if (tree == NULL) {
656                 return 0;
657         }
658
659         if (tree->operation == LDB_OP_NOT) {
660                 return map_subtree_select_local_not(module, mem_ctx, new, tree);
661         }
662
663         if (tree->operation == LDB_OP_AND || tree->operation == LDB_OP_OR) {
664                 return map_subtree_select_local_list(module, mem_ctx, new, tree);
665         }
666
667         if (map_attr_check_remote(data, tree->u.equality.attr)) {
668                 *new = NULL;
669                 return 0;
670         }
671
672         return map_subtree_select_local_simple(module, mem_ctx, new, tree);
673 }
674
675 static int map_subtree_collect_remote(struct ldb_module *module, void *mem_ctx, struct ldb_parse_tree **new, const struct ldb_parse_tree *tree);
676
677 /* Collect a negated subtree that queries attributes in the remote partition */
678 static int map_subtree_collect_remote_not(struct ldb_module *module, void *mem_ctx, struct ldb_parse_tree **new, const struct ldb_parse_tree *tree)
679 {
680         struct ldb_parse_tree *child;
681         int ret;
682
683         /* Prepare new tree */
684         *new = talloc_memdup(mem_ctx, tree, sizeof(struct ldb_parse_tree));
685         if (*new == NULL) {
686                 map_oom(module);
687                 return -1;
688         }
689
690         /* Generate new subtree */
691         ret = map_subtree_collect_remote(module, *new, &child, tree->u.isnot.child);
692         if (ret) {
693                 talloc_free(*new);
694                 return ret;
695         }
696
697         /* Prune tree without subtree */
698         if (child == NULL) {
699                 talloc_free(*new);
700                 *new = NULL;
701                 return 0;
702         }
703
704         (*new)->u.isnot.child = child;
705
706         return ret;
707 }
708
709 /* Collect a list of subtrees that query attributes in the remote partition */
710 static int map_subtree_collect_remote_list(struct ldb_module *module, void *mem_ctx, struct ldb_parse_tree **new, const struct ldb_parse_tree *tree)
711 {
712         int i, j, ret=0;
713
714         /* Prepare new tree */
715         *new = talloc_memdup(mem_ctx, tree, sizeof(struct ldb_parse_tree));
716         if (*new == NULL) {
717                 map_oom(module);
718                 return -1;
719         }
720
721         /* Prepare list of subtrees */
722         (*new)->u.list.num_elements = 0;
723         (*new)->u.list.elements = talloc_array(*new, struct ldb_parse_tree *, tree->u.list.num_elements);
724         if ((*new)->u.list.elements == NULL) {
725                 map_oom(module);
726                 talloc_free(*new);
727                 return -1;
728         }
729
730         /* Generate new list of subtrees */
731         j = 0;
732         for (i = 0; i < tree->u.list.num_elements; i++) {
733                 struct ldb_parse_tree *child;
734                 ret = map_subtree_collect_remote(module, *new, &child, tree->u.list.elements[i]);
735                 if (ret) {
736                         talloc_free(*new);
737                         return ret;
738                 }
739
740                 if (child) {
741                         (*new)->u.list.elements[j] = child;
742                         j++;
743                 }
744         }
745
746         /* Prune tree without subtrees */
747         if (j == 0) {
748                 talloc_free(*new);
749                 *new = NULL;
750                 return 0;
751         }
752
753         /* Fix subtree list size */
754         (*new)->u.list.num_elements = j;
755         (*new)->u.list.elements = talloc_realloc(*new, (*new)->u.list.elements, struct ldb_parse_tree *, (*new)->u.list.num_elements);
756
757         return ret;
758 }
759
760 /* Collect a simple subtree that queries attributes in the remote partition */
761 int map_subtree_collect_remote_simple(struct ldb_module *module, void *mem_ctx, struct ldb_parse_tree **new, const struct ldb_parse_tree *tree, const struct ldb_map_attribute *map)
762 {
763         const char *attr;
764
765         /* Prepare new tree */
766         *new = talloc(mem_ctx, struct ldb_parse_tree);
767         if (*new == NULL) {
768                 map_oom(module);
769                 return -1;
770         }
771         **new = *tree;
772         
773         if (map->type == MAP_KEEP) {
774                 /* Nothing to do here */
775                 return 0;
776         }
777
778         /* Store attribute and value in new tree */
779         switch (tree->operation) {
780         case LDB_OP_PRESENT:
781                 attr = map_attr_map_local(*new, map, tree->u.present.attr);
782                 (*new)->u.present.attr = attr;
783                 break;
784         case LDB_OP_SUBSTRING:
785         {
786                 attr = map_attr_map_local(*new, map, tree->u.substring.attr);
787                 (*new)->u.substring.attr = attr;
788                 break;
789         }
790         case LDB_OP_EQUALITY:
791                 attr = map_attr_map_local(*new, map, tree->u.equality.attr);
792                 (*new)->u.equality.attr = attr;
793                 break;
794         case LDB_OP_LESS:
795         case LDB_OP_GREATER:
796         case LDB_OP_APPROX:
797                 attr = map_attr_map_local(*new, map, tree->u.comparison.attr);
798                 (*new)->u.comparison.attr = attr;
799                 break;
800         case LDB_OP_EXTENDED:
801                 attr = map_attr_map_local(*new, map, tree->u.extended.attr);
802                 (*new)->u.extended.attr = attr;
803                 break;
804         default:                        /* unknown kind of simple subtree */
805                 talloc_free(*new);
806                 return -1;
807         }
808
809         if (attr == NULL) {
810                 talloc_free(*new);
811                 *new = NULL;
812                 return 0;
813         }
814
815         if (map->type == MAP_RENAME) {
816                 /* Nothing more to do here, the attribute has been renamed */
817                 return 0;
818         }
819
820         /* Store attribute and value in new tree */
821         switch (tree->operation) {
822         case LDB_OP_PRESENT:
823                 break;
824         case LDB_OP_SUBSTRING:
825         {
826                 int i;
827                 /* Map value */
828                 (*new)->u.substring.chunks = NULL;
829                 for (i=0; tree->u.substring.chunks[i]; i++) {
830                         (*new)->u.substring.chunks = talloc_realloc(*new, (*new)->u.substring.chunks, struct ldb_val *, i+2);
831                         if (!(*new)->u.substring.chunks) {
832                                 talloc_free(*new);
833                                 *new = NULL;
834                                 return 0;
835                         }
836                         (*new)->u.substring.chunks[i] = talloc(*new, struct ldb_val);
837                         if (!(*new)->u.substring.chunks[i]) {
838                                 talloc_free(*new);
839                                 *new = NULL;
840                                 return 0;
841                         }
842                         *(*new)->u.substring.chunks[i] = ldb_val_map_local(module, *new, map, tree->u.substring.chunks[i]);
843                         (*new)->u.substring.chunks[i+1] = NULL;
844                 }
845                 break;
846         }
847         case LDB_OP_EQUALITY:
848                 (*new)->u.equality.value = ldb_val_map_local(module, *new, map, &tree->u.equality.value);
849                 break;
850         case LDB_OP_LESS:
851         case LDB_OP_GREATER:
852         case LDB_OP_APPROX:
853                 (*new)->u.comparison.value = ldb_val_map_local(module, *new, map, &tree->u.comparison.value);
854                 break;
855         case LDB_OP_EXTENDED:
856                 (*new)->u.extended.value = ldb_val_map_local(module, *new, map, &tree->u.extended.value);
857                 (*new)->u.extended.rule_id = talloc_strdup(*new, tree->u.extended.rule_id);
858                 break;
859         default:                        /* unknown kind of simple subtree */
860                 talloc_free(*new);
861                 return -1;
862         }
863
864         return 0;
865 }
866
867 /* Collect subtrees that query attributes in the remote partition */
868 static int map_subtree_collect_remote(struct ldb_module *module, void *mem_ctx, struct ldb_parse_tree **new, const struct ldb_parse_tree *tree)
869 {
870         const struct ldb_map_context *data = map_get_context(module);
871         const struct ldb_map_attribute *map;
872
873         if (tree == NULL) {
874                 return 0;
875         }
876
877         if (tree->operation == LDB_OP_NOT) {
878                 return map_subtree_collect_remote_not(module, mem_ctx, new, tree);
879         }
880
881         if ((tree->operation == LDB_OP_AND) || (tree->operation == LDB_OP_OR)) {
882                 return map_subtree_collect_remote_list(module, mem_ctx, new, tree);
883         }
884
885         if (!map_attr_check_remote(data, tree->u.equality.attr)) {
886                 *new = NULL;
887                 return 0;
888         }
889
890         map = map_attr_find_local(data, tree->u.equality.attr);
891         if (map->convert_operator) {
892                 return map->convert_operator(module, mem_ctx, new, tree);
893         }
894
895         if (map->type == MAP_GENERATE) {
896                 ldb_debug(module->ldb, LDB_DEBUG_WARNING, "ldb_map: "
897                           "Skipping attribute '%s': "
898                           "'convert_operator' not set\n",
899                           tree->u.equality.attr);
900                 *new = NULL;
901                 return 0;
902         }
903
904         return map_subtree_collect_remote_simple(module, mem_ctx, new, tree, map);
905 }
906
907 /* Split subtrees that query attributes in the local partition from
908  * those that query the remote partition. */
909 static int ldb_parse_tree_partition(struct ldb_module *module,
910                                         void *mem_ctx,
911                                         struct ldb_parse_tree **local_tree,
912                                         struct ldb_parse_tree **remote_tree,
913                                         const struct ldb_parse_tree *tree)
914 {
915         int ret;
916
917         *local_tree = NULL;
918         *remote_tree = NULL;
919
920         /* No original tree */
921         if (tree == NULL) {
922                 return 0;
923         }
924
925         /* Generate local tree */
926         ret = map_subtree_select_local(module, mem_ctx, local_tree, tree);
927         if (ret) {
928                 return ret;
929         }
930
931         /* Generate remote tree */
932         ret = map_subtree_collect_remote(module, mem_ctx, remote_tree, tree);
933         if (ret) {
934                 talloc_free(*local_tree);
935                 return ret;
936         }
937
938         return 0;
939 }
940
941 /* Collect a list of attributes required either explicitly from a
942  * given list or implicitly  from a given parse tree; split the
943  * collected list into local and remote parts. */
944 static int map_attrs_collect_and_partition(struct ldb_module *module, struct map_context *ac,
945                                            const char * const *search_attrs, 
946                                            const struct ldb_parse_tree *tree)
947 {
948         void *tmp_ctx;
949         const char **tree_attrs;
950         const char **remote_attrs;
951         const char **local_attrs;
952         int ret;
953
954         /* There is no tree, just partition the searched attributes */
955         if (tree == NULL) {
956                 ret = map_attrs_partition(module, ac, 
957                                           &local_attrs, &remote_attrs, search_attrs);
958                 if (ret == 0) {
959                         ac->local_attrs = local_attrs;
960                         ac->remote_attrs = remote_attrs;
961                         ac->all_attrs = search_attrs;
962                 }
963                 return ret; 
964         }
965
966         /* Create context for temporary memory */
967         tmp_ctx = talloc_new(ac);
968         if (tmp_ctx == NULL) {
969                 goto oom;
970         }
971
972         /* Prepare list of attributes from tree */
973         tree_attrs = talloc_array(tmp_ctx, const char *, 1);
974         if (tree_attrs == NULL) {
975                 talloc_free(tmp_ctx);
976                 goto oom;
977         }
978         tree_attrs[0] = NULL;
979
980         /* Collect attributes from tree */
981         ret = ldb_parse_tree_collect_attrs(module, tmp_ctx, &tree_attrs, tree);
982         if (ret) {
983                 goto done;
984         }
985
986         /* Merge attributes from search operation */
987         ret = map_attrs_merge(module, tmp_ctx, &tree_attrs, search_attrs);
988         if (ret) {
989                 goto done;
990         }
991
992         /* Split local from remote attributes */
993         ret = map_attrs_partition(module, ac, &local_attrs, 
994                                   &remote_attrs, tree_attrs);
995         
996         if (ret == 0) {
997                 ac->local_attrs = local_attrs;
998                 ac->remote_attrs = remote_attrs;
999                 talloc_steal(ac, tree_attrs);
1000                 ac->all_attrs = tree_attrs;
1001         }
1002 done:
1003         /* Free temporary memory */
1004         talloc_free(tmp_ctx);
1005         return ret;
1006
1007 oom:
1008         map_oom(module);
1009         return -1;
1010 }
1011
1012
1013 /* Outbound requests: search
1014  * ========================= */
1015
1016 static int map_remote_search_callback(struct ldb_request *req,
1017                                         struct ldb_reply *ares);
1018 static int map_local_merge_callback(struct ldb_request *req,
1019                                         struct ldb_reply *ares);
1020 static int map_search_local(struct map_context *ac);
1021
1022 static int map_save_entry(struct map_context *ac, struct ldb_reply *ares)
1023 {
1024         struct map_reply *mr;
1025
1026         mr = talloc_zero(ac, struct map_reply);
1027         if (mr == NULL) {
1028                 map_oom(ac->module);
1029                 return LDB_ERR_OPERATIONS_ERROR;
1030         }
1031         mr->remote = talloc_steal(mr, ares);
1032         if (ac->r_current) {
1033                 ac->r_current->next = mr;
1034         } else {
1035                 /* first entry */
1036                 ac->r_list = mr;
1037         }
1038         ac->r_current = mr;
1039
1040         return LDB_SUCCESS;
1041 }
1042
1043 /* Pass a merged search result up the callback chain. */
1044 int map_return_entry(struct map_context *ac, struct ldb_reply *ares)
1045 {
1046         struct ldb_message_element *el;
1047         const char * const *attrs;
1048         int i;
1049
1050         /* Merged result doesn't match original query, skip */
1051         if (!ldb_match_msg(ac->module->ldb, ares->message,
1052                            ac->req->op.search.tree,
1053                            ac->req->op.search.base,
1054                            ac->req->op.search.scope)) {
1055                 ldb_debug(ac->module->ldb, LDB_DEBUG_TRACE, "ldb_map: "
1056                           "Skipping record '%s': "
1057                           "doesn't match original search\n",
1058                           ldb_dn_get_linearized(ares->message->dn));
1059                 return LDB_SUCCESS;
1060         }
1061
1062         /* Limit result to requested attrs */
1063         if (ac->req->op.search.attrs &&
1064             (! ldb_attr_in_list(ac->req->op.search.attrs, "*"))) {
1065
1066                 attrs = ac->req->op.search.attrs;
1067                 i = 0;
1068
1069                 while (i < ares->message->num_elements) {
1070
1071                         el = &ares->message->elements[i];
1072                         if ( ! ldb_attr_in_list(attrs, el->name)) {
1073                                 ldb_msg_remove_element(ares->message, el);
1074                         } else {
1075                                 i++;
1076                         }
1077                 }
1078         }
1079
1080         return ldb_module_send_entry(ac->req, ares->message);
1081 }
1082
1083 /* Search a record. */
1084 int map_search(struct ldb_module *module, struct ldb_request *req)
1085 {
1086         struct ldb_parse_tree *remote_tree;
1087         struct ldb_parse_tree *local_tree;
1088         struct ldb_request *remote_req;
1089         struct map_context *ac;
1090         int ret;
1091
1092         const char *wildcard[] = { "*", NULL };
1093         const char * const *attrs;
1094
1095         if (!module->private_data) /* if we're not yet initialized, go to the next module */
1096                 return ldb_next_request(module, req);
1097
1098         /* Do not manipulate our control entries */
1099         if (ldb_dn_is_special(req->op.search.base)) {
1100                 return ldb_next_request(module, req);
1101         }
1102
1103         /* No mapping requested, skip to next module */
1104         if ((req->op.search.base) && (!ldb_dn_check_local(module, req->op.search.base))) {
1105                 return ldb_next_request(module, req);
1106         }
1107
1108         /* TODO: How can we be sure about which partition we are
1109          *       targetting when there is no search base? */
1110
1111         /* Prepare context and handle */
1112         ac = map_init_context(module, req);
1113         if (ac == NULL) {
1114                 return LDB_ERR_OPERATIONS_ERROR;
1115         }
1116
1117         /* It is easier to deal with the two different ways of
1118          * expressing the wildcard in the same codepath */
1119         attrs = req->op.search.attrs;
1120         if (attrs == NULL) {
1121                 attrs = wildcard;
1122         }
1123
1124         /* Split local from remote attrs */
1125         ret = map_attrs_collect_and_partition(module, ac, 
1126                                               attrs, req->op.search.tree);
1127         if (ret) {
1128                 return LDB_ERR_OPERATIONS_ERROR;
1129         }
1130
1131         /* Split local from remote tree */
1132         ret = ldb_parse_tree_partition(module, ac,
1133                                        &local_tree, &remote_tree,
1134                                        req->op.search.tree);
1135         if (ret) {
1136                 return LDB_ERR_OPERATIONS_ERROR;
1137         }
1138
1139         if (((local_tree != NULL) && (remote_tree != NULL)) &&
1140             (!ldb_parse_tree_check_splittable(req->op.search.tree))) {
1141                 /* The query can't safely be split, enumerate the remote partition */
1142                 local_tree = NULL;
1143                 remote_tree = NULL;
1144         }
1145
1146         if (local_tree == NULL) {
1147                 /* Construct default local parse tree */
1148                 local_tree = talloc_zero(ac, struct ldb_parse_tree);
1149                 if (local_tree == NULL) {
1150                         map_oom(ac->module);
1151                         return LDB_ERR_OPERATIONS_ERROR;
1152                 }
1153
1154                 local_tree->operation = LDB_OP_PRESENT;
1155                 local_tree->u.present.attr = talloc_strdup(local_tree, IS_MAPPED);
1156         }
1157         if (remote_tree == NULL) {
1158                 /* Construct default remote parse tree */
1159                 remote_tree = ldb_parse_tree(ac, NULL);
1160                 if (remote_tree == NULL) {
1161                         return LDB_ERR_OPERATIONS_ERROR;
1162                 }
1163         }
1164
1165         ac->local_tree = local_tree;
1166
1167         /* Prepare the remote operation */
1168         ret = ldb_build_search_req_ex(&remote_req, module->ldb, ac,
1169                                       req->op.search.base,
1170                                       req->op.search.scope,
1171                                       remote_tree,
1172                                       ac->remote_attrs,
1173                                       req->controls,
1174                                       ac, map_remote_search_callback,
1175                                       req);
1176         if (ret != LDB_SUCCESS) {
1177                 return LDB_ERR_OPERATIONS_ERROR;
1178         }
1179
1180         return ldb_next_remote_request(module, remote_req);
1181 }
1182
1183 /* Now, search the local part of a remote search result. */
1184 static int map_remote_search_callback(struct ldb_request *req,
1185                                         struct ldb_reply *ares)
1186 {
1187         struct map_context *ac;
1188         int ret;
1189
1190         ac = talloc_get_type(req->context, struct map_context);
1191
1192         if (!ares) {
1193                 return ldb_module_done(ac->req, NULL, NULL,
1194                                         LDB_ERR_OPERATIONS_ERROR);
1195         }
1196         if (ares->error != LDB_SUCCESS) {
1197                 return ldb_module_done(ac->req, ares->controls,
1198                                         ares->response, ares->error);
1199         }
1200
1201         switch (ares->type) {
1202         case LDB_REPLY_REFERRAL:
1203
1204                 /* ignore referrals */
1205                 talloc_free(ares);
1206                 return LDB_SUCCESS;
1207
1208         case LDB_REPLY_ENTRY:
1209
1210                 /* Map result record into a local message */
1211                 ret = map_reply_remote(ac, ares);
1212                 if (ret) {
1213                         talloc_free(ares);
1214                         return ldb_module_done(ac->req, NULL, NULL,
1215                                                 LDB_ERR_OPERATIONS_ERROR);
1216                 }
1217
1218                 /* if we have no local db, then we can just return the reply to
1219                  * the upper layer, otherwise we must save it and process it
1220                  * when all replies ahve been gathered */
1221                 if ( ! map_check_local_db(ac->module)) {
1222                         ret = map_return_entry(ac, ares);
1223                 } else {
1224                         ret = map_save_entry(ac,ares);
1225                 }
1226
1227                 if (ret != LDB_SUCCESS) {
1228                         talloc_free(ares);
1229                         return ldb_module_done(ac->req, NULL, NULL,
1230                                                 LDB_ERR_OPERATIONS_ERROR);
1231                 }
1232                 break;
1233
1234         case LDB_REPLY_DONE:
1235
1236                 if ( ! map_check_local_db(ac->module)) {
1237                         return ldb_module_done(ac->req, ares->controls,
1238                                                 ares->response, LDB_SUCCESS);
1239                 }
1240
1241                 talloc_free(ares);
1242
1243                 /* reset the pointer to the start of the list */
1244                 ac->r_current = ac->r_list;
1245
1246                 /* no entry just return */
1247                 if (ac->r_current == NULL) {
1248                         return ldb_module_done(ac->req, ares->controls,
1249                                                 ares->response, LDB_SUCCESS);
1250                 }
1251
1252                 ret = map_search_local(ac);
1253                 if (ret != LDB_SUCCESS) {
1254                         return ldb_module_done(ac->req, NULL, NULL, ret);
1255                 }
1256         }
1257
1258         return LDB_SUCCESS;
1259 }
1260
1261 static int map_search_local(struct map_context *ac)
1262 {
1263         struct ldb_request *search_req;
1264
1265         if (ac->r_current == NULL || ac->r_current->remote == NULL) {
1266                 return LDB_ERR_OPERATIONS_ERROR;
1267         }
1268
1269         /* Prepare local search request */
1270         /* TODO: use GUIDs here instead? */
1271         search_req = map_search_base_req(ac,
1272                                          ac->r_current->remote->message->dn,
1273                                          NULL, NULL,
1274                                          ac, map_local_merge_callback);
1275         if (search_req == NULL) {
1276                 return LDB_ERR_OPERATIONS_ERROR;
1277         }
1278
1279         return ldb_next_request(ac->module, search_req);
1280 }
1281
1282 /* Merge the remote and local parts of a search result. */
1283 int map_local_merge_callback(struct ldb_request *req, struct ldb_reply *ares)
1284 {
1285         struct map_context *ac;
1286         int ret;
1287
1288         ac = talloc_get_type(req->context, struct map_context);
1289
1290         if (!ares) {
1291                 return ldb_module_done(ac->req, NULL, NULL,
1292                                         LDB_ERR_OPERATIONS_ERROR);
1293         }
1294         if (ares->error != LDB_SUCCESS) {
1295                 return ldb_module_done(ac->req, ares->controls,
1296                                         ares->response, ares->error);
1297         }
1298
1299         switch (ares->type) {
1300         case LDB_REPLY_ENTRY:
1301                 /* We have already found a local record */
1302                 if (ac->r_current->local) {
1303                         talloc_free(ares);
1304                         ldb_set_errstring(ac->module->ldb, "ldb_map: Too many results!");
1305                         return ldb_module_done(ac->req, NULL, NULL,
1306                                                 LDB_ERR_OPERATIONS_ERROR);
1307                 }
1308
1309                 /* Store local result */
1310                 ac->r_current->local = talloc_steal(ac->r_current, ares);
1311
1312                 break;
1313
1314         case LDB_REPLY_REFERRAL:
1315                 /* ignore referrals */
1316                 talloc_free(ares);
1317                 break;
1318
1319         case LDB_REPLY_DONE:
1320                 talloc_free(ares);
1321
1322                 /* No local record found, map and send remote record */
1323                 if (ac->r_current->local != NULL) {
1324                         /* Merge remote into local message */
1325                         ret = ldb_msg_merge_local(ac->module,
1326                                                   ac->r_current->local->message,
1327                                                   ac->r_current->remote->message);
1328                         if (ret == LDB_SUCCESS) {
1329                                 ret = map_return_entry(ac, ac->r_current->local);
1330                         }
1331                         if (ret != LDB_SUCCESS) {
1332                                 return ldb_module_done(ac->req, NULL, NULL,
1333                                                         LDB_ERR_OPERATIONS_ERROR);
1334                         }
1335                 } else {
1336                         ret = map_return_entry(ac, ac->r_current->remote);
1337                         if (ret != LDB_SUCCESS) {
1338                                 return ldb_module_done(ac->req,
1339                                                         NULL, NULL, ret);
1340                         }
1341                 }
1342
1343                 if (ac->r_current->next != NULL) {
1344                         ac->r_current = ac->r_current->next;
1345                         if (ac->r_current->remote->type == LDB_REPLY_ENTRY) {
1346                                 ret = map_search_local(ac);
1347                                 if (ret != LDB_SUCCESS) {
1348                                         return ldb_module_done(ac->req,
1349                                                                NULL, NULL, ret);
1350                                 }
1351                                 break;
1352                         }
1353                 }
1354
1355                 /* ok we are done with all search, finally it is time to
1356                  * finish operations for this module */
1357                 return ldb_module_done(ac->req,
1358                                         ac->r_current->remote->controls,
1359                                         ac->r_current->remote->response,
1360                                         ac->r_current->remote->error);
1361         }
1362
1363         return LDB_SUCCESS;
1364 }