?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
s4:dsdb: passdown DSDB_CONTROL_REPLICATED_UPDATE_OID for replicated updates
[Samba/nascimento.git] / source4 / dsdb / samdb / ldb_modules / repl_meta_data.c
blobfbaf461a3fa11d0da109c52e1a687b88324c0374
1 /*
2 ldb database library
3
4 Copyright (C) Simo Sorce 2004-2006
5 Copyright (C) Andrew Bartlett <abartlet@samba.org> 2005
6 Copyright (C) Andrew Tridgell 2005
7 Copyright (C) Stefan Metzmacher <metze@samba.org> 2007
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
29 *
30 * Component: ldb repl_meta_data module
31 *
32 * Description: - add a unique objectGUID onto every new record,
33 * - handle whenCreated, whenChanged timestamps
34 * - handle uSNCreated, uSNChanged numbers
35 * - handle replPropertyMetaData attribute
36 *
37 * Author: Simo Sorce
38 * Author: Stefan Metzmacher
39 */
40
41 #include "includes.h"
42 #include "lib/ldb/include/ldb.h"
43 #include "lib/ldb/include/ldb_errors.h"
44 #include "lib/ldb/include/ldb_private.h"
45 #include "dsdb/samdb/samdb.h"
46 #include "dsdb/common/flags.h"
47 #include "librpc/gen_ndr/ndr_misc.h"
48 #include "librpc/gen_ndr/ndr_drsuapi.h"
49 #include "librpc/gen_ndr/ndr_drsblobs.h"
50 #include "param/param.h"
51
52 struct replmd_replicated_request {
53 struct ldb_module *module;
54 struct ldb_handle *handle;
55 struct ldb_request *orig_req;
56
57 const struct dsdb_schema *schema;
58
59 struct dsdb_extended_replicated_objects *objs;
60
61 /* the controls we pass down */
62 struct ldb_control **controls;
63
64 uint32_t index_current;
65
66 struct {
67 TALLOC_CTX *mem_ctx;
68 struct ldb_request *search_req;
69 struct ldb_message *search_msg;
70 int search_ret;
71 struct ldb_request *change_req;
72 int change_ret;
73 } sub;
74 };
75
76 static struct replmd_replicated_request *replmd_replicated_init_handle(struct ldb_module *module,
77 struct ldb_request *req,
78 struct dsdb_extended_replicated_objects *objs)
79 {
80 struct replmd_replicated_request *ar;
81 struct ldb_handle *h;
82 const struct dsdb_schema *schema;
83
84 schema = dsdb_get_schema(module->ldb);
85 if (!schema) {
86 ldb_debug_set(module->ldb, LDB_DEBUG_FATAL,
87 "replmd_replicated_init_handle: no loaded schema found\n");
88 return NULL;
89 }
90
91 h = talloc_zero(req, struct ldb_handle);
92 if (h == NULL) {
93 ldb_set_errstring(module->ldb, "Out of Memory");
94 return NULL;
95 }
96
97 h->module = module;
98 h->state = LDB_ASYNC_PENDING;
99 h->status = LDB_SUCCESS;
100
101 ar = talloc_zero(h, struct replmd_replicated_request);
102 if (ar == NULL) {
103 ldb_set_errstring(module->ldb, "Out of Memory");
104 talloc_free(h);
105 return NULL;
106 }
107
108 h->private_data = ar;
109
110 ar->module = module;
111 ar->handle = h;
112 ar->orig_req = req;
113 ar->schema = schema;
114 ar->objs = objs;
115
116 req->handle = h;
117
118 return ar;
119 }
120
121 /*
122 add a time element to a record
123 */
124 static int add_time_element(struct ldb_message *msg, const char *attr, time_t t)
125 {
126 struct ldb_message_element *el;
127 char *s;
128
129 if (ldb_msg_find_element(msg, attr) != NULL) {
130 return 0;
131 }
132
133 s = ldb_timestring(msg, t);
134 if (s == NULL) {
135 return -1;
136 }
137
138 if (ldb_msg_add_string(msg, attr, s) != 0) {
139 return -1;
140 }
141
142 el = ldb_msg_find_element(msg, attr);
143 /* always set as replace. This works because on add ops, the flag
144 is ignored */
145 el->flags = LDB_FLAG_MOD_REPLACE;
146
147 return 0;
148 }
149
150 /*
151 add a uint64_t element to a record
152 */
153 static int add_uint64_element(struct ldb_message *msg, const char *attr, uint64_t v)
154 {
155 struct ldb_message_element *el;
156
157 if (ldb_msg_find_element(msg, attr) != NULL) {
158 return 0;
159 }
160
161 if (ldb_msg_add_fmt(msg, attr, "%llu", (unsigned long long)v) != 0) {
162 return -1;
163 }
164
165 el = ldb_msg_find_element(msg, attr);
166 /* always set as replace. This works because on add ops, the flag
167 is ignored */
168 el->flags = LDB_FLAG_MOD_REPLACE;
169
170 return 0;
171 }
172
173 static int replmd_replPropertyMetaData1_attid_sort(const struct replPropertyMetaData1 *m1,
174 const struct replPropertyMetaData1 *m2,
175 const uint32_t *rdn_attid)
176 {
177 if (m1->attid == m2->attid) {
178 return 0;
179 }
180
181 /*
182 * the rdn attribute should be at the end!
183 * so we need to return a value greater than zero
184 * which means m1 is greater than m2
185 */
186 if (m1->attid == *rdn_attid) {
187 return 1;
188 }
189
190 /*
191 * the rdn attribute should be at the end!
192 * so we need to return a value less than zero
193 * which means m2 is greater than m1
194 */
195 if (m2->attid == *rdn_attid) {
196 return -1;
197 }
198
199 return m1->attid - m2->attid;
200 }
201
202 static void replmd_replPropertyMetaDataCtr1_sort(struct replPropertyMetaDataCtr1 *ctr1,
203 const uint32_t *rdn_attid)
204 {
205 ldb_qsort(ctr1->array, ctr1->count, sizeof(struct replPropertyMetaData1),
206 discard_const_p(void, rdn_attid), (ldb_qsort_cmp_fn_t)replmd_replPropertyMetaData1_attid_sort);
207 }
208
209 static int replmd_ldb_message_element_attid_sort(const struct ldb_message_element *e1,
210 const struct ldb_message_element *e2,
211 const struct dsdb_schema *schema)
212 {
213 const struct dsdb_attribute *a1;
214 const struct dsdb_attribute *a2;
215
216 /*
217 * TODO: make this faster by caching the dsdb_attribute pointer
218 * on the ldb_messag_element
219 */
220
221 a1 = dsdb_attribute_by_lDAPDisplayName(schema, e1->name);
222 a2 = dsdb_attribute_by_lDAPDisplayName(schema, e2->name);
223
224 /*
225 * TODO: remove this check, we should rely on e1 and e2 having valid attribute names
226 * in the schema
227 */
228 if (!a1 || !a2) {
229 return strcasecmp(e1->name, e2->name);
230 }
231
232 return a1->attributeID_id - a2->attributeID_id;
233 }
234
235 static void replmd_ldb_message_sort(struct ldb_message *msg,
236 const struct dsdb_schema *schema)
237 {
238 ldb_qsort(msg->elements, msg->num_elements, sizeof(struct ldb_message_element),
239 discard_const_p(void, schema), (ldb_qsort_cmp_fn_t)replmd_ldb_message_element_attid_sort);
240 }
241
242 static int replmd_prepare_originating(struct ldb_module *module, struct ldb_request *req,
243 struct ldb_dn *dn, const char *fn_name,
244 int (*fn)(struct ldb_module *,
245 struct ldb_request *,
246 const struct dsdb_schema *))
247 {
248 const struct dsdb_schema *schema;
249
250 /* do not manipulate our control entries */
251 if (ldb_dn_is_special(dn)) {
252 return ldb_next_request(module, req);
253 }
254
255 schema = dsdb_get_schema(module->ldb);
256 if (!schema) {
257 ldb_debug_set(module->ldb, LDB_DEBUG_FATAL,
258 "%s: no dsdb_schema loaded",
259 fn_name);
260 return LDB_ERR_CONSTRAINT_VIOLATION;
261 }
262
263 return fn(module, req, schema);
264 }
265
266 static int replmd_add_originating(struct ldb_module *module,
267 struct ldb_request *req,
268 const struct dsdb_schema *schema)
269 {
270 enum ndr_err_code ndr_err;
271 struct ldb_request *down_req;
272 struct ldb_message *msg;
273 const struct dsdb_attribute *rdn_attr = NULL;
274 struct GUID guid;
275 struct ldb_val guid_value;
276 struct replPropertyMetaDataBlob nmd;
277 struct ldb_val nmd_value;
278 uint64_t seq_num;
279 const struct GUID *our_invocation_id;
280 time_t t = time(NULL);
281 NTTIME now;
282 char *time_str;
283 int ret;
284 uint32_t i, ni=0;
285
286 ldb_debug(module->ldb, LDB_DEBUG_TRACE, "replmd_add_originating\n");
287
288 if (ldb_msg_find_element(req->op.add.message, "objectGUID")) {
289 ldb_debug_set(module->ldb, LDB_DEBUG_ERROR,
290 "replmd_add_originating: it's not allowed to add an object with objectGUID\n");
291 return LDB_ERR_UNWILLING_TO_PERFORM;
292 }
293
294 /* Get a sequence number from the backend */
295 ret = ldb_sequence_number(module->ldb, LDB_SEQ_NEXT, &seq_num);
296 if (ret != LDB_SUCCESS) {
297 return ret;
298 }
299
300 /* a new GUID */
301 guid = GUID_random();
302
303 /* get our invicationId */
304 our_invocation_id = samdb_ntds_invocation_id(module->ldb);
305 if (!our_invocation_id) {
306 ldb_debug_set(module->ldb, LDB_DEBUG_ERROR,
307 "replmd_add_originating: unable to find invocationId\n");
308 return LDB_ERR_OPERATIONS_ERROR;
309 }
310
311 /* create a copy of the request */
312 down_req = talloc(req, struct ldb_request);
313 if (down_req == NULL) {
314 ldb_oom(module->ldb);
315 return LDB_ERR_OPERATIONS_ERROR;
316 }
317 *down_req = *req;
318
319 /* we have to copy the message as the caller might have it as a const */
320 down_req->op.add.message = msg = ldb_msg_copy_shallow(down_req, req->op.add.message);
321 if (msg == NULL) {
322 talloc_free(down_req);
323 ldb_oom(module->ldb);
324 return LDB_ERR_OPERATIONS_ERROR;
325 }
326
327 /* generated times */
328 unix_to_nt_time(&now, t);
329 time_str = ldb_timestring(msg, t);
330 if (!time_str) {
331 talloc_free(down_req);
332 return LDB_ERR_OPERATIONS_ERROR;
333 }
334
335 /*
336 * remove autogenerated attributes
337 */
338 ldb_msg_remove_attr(msg, "whenCreated");
339 ldb_msg_remove_attr(msg, "whenChanged");
340 ldb_msg_remove_attr(msg, "uSNCreated");
341 ldb_msg_remove_attr(msg, "uSNChanged");
342 ldb_msg_remove_attr(msg, "replPropertyMetaData");
343
344 /*
345 * readd replicated attributes
346 */
347 ret = ldb_msg_add_string(msg, "whenCreated", time_str);
348 if (ret != LDB_SUCCESS) {
349 talloc_free(down_req);
350 ldb_oom(module->ldb);
351 return LDB_ERR_OPERATIONS_ERROR;
352 }
353
354 /* build the replication meta_data */
355 ZERO_STRUCT(nmd);
356 nmd.version = 1;
357 nmd.ctr.ctr1.count = msg->num_elements;
358 nmd.ctr.ctr1.array = talloc_array(msg,
359 struct replPropertyMetaData1,
360 nmd.ctr.ctr1.count);
361 if (!nmd.ctr.ctr1.array) {
362 talloc_free(down_req);
363 ldb_oom(module->ldb);
364 return LDB_ERR_OPERATIONS_ERROR;
365 }
366
367 for (i=0; i < msg->num_elements; i++) {
368 struct ldb_message_element *e = &msg->elements[i];
369 struct replPropertyMetaData1 *m = &nmd.ctr.ctr1.array[ni];
370 const struct dsdb_attribute *sa;
371
372 if (e->name[0] == '@') continue;
373
374 sa = dsdb_attribute_by_lDAPDisplayName(schema, e->name);
375 if (!sa) {
376 ldb_debug_set(module->ldb, LDB_DEBUG_ERROR,
377 "replmd_add_originating: attribute '%s' not defined in schema\n",
378 e->name);
379 talloc_free(down_req);
380 return LDB_ERR_NO_SUCH_ATTRIBUTE;
381 }
382
383 if ((sa->systemFlags & 0x00000001) || (sa->systemFlags & 0x00000004)) {
384 /* if the attribute is not replicated (0x00000001)
385 * or constructed (0x00000004) it has no metadata
386 */
387 continue;
388 }
389
390 m->attid = sa->attributeID_id;
391 m->version = 1;
392 m->originating_change_time = now;
393 m->originating_invocation_id = *our_invocation_id;
394 m->originating_usn = seq_num;
395 m->local_usn = seq_num;
396 ni++;
397
398 if (ldb_attr_cmp(e->name, ldb_dn_get_rdn_name(msg->dn))) {
399 rdn_attr = sa;
400 }
401 }
402
403 /* fix meta data count */
404 nmd.ctr.ctr1.count = ni;
405
406 /*
407 * sort meta data array, and move the rdn attribute entry to the end
408 */
409 replmd_replPropertyMetaDataCtr1_sort(&nmd.ctr.ctr1, &rdn_attr->attributeID_id);
410
411 /* generated NDR encoded values */
412 ndr_err = ndr_push_struct_blob(&guid_value, msg,
413 NULL,
414 &guid,
415 (ndr_push_flags_fn_t)ndr_push_GUID);
416 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
417 ldb_oom(module->ldb);
418 return LDB_ERR_OPERATIONS_ERROR;
419 }
420 ndr_err = ndr_push_struct_blob(&nmd_value, msg,
421 lp_iconv_convenience(ldb_get_opaque(module->ldb, "loadparm")),
422 &nmd,
423 (ndr_push_flags_fn_t)ndr_push_replPropertyMetaDataBlob);
424 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
425 talloc_free(down_req);
426 ldb_oom(module->ldb);
427 return LDB_ERR_OPERATIONS_ERROR;
428 }
429
430 /*
431 * add the autogenerated values
432 */
433 ret = ldb_msg_add_value(msg, "objectGUID", &guid_value, NULL);
434 if (ret != LDB_SUCCESS) {
435 talloc_free(down_req);
436 ldb_oom(module->ldb);
437 return LDB_ERR_OPERATIONS_ERROR;
438 }
439 ret = ldb_msg_add_string(msg, "whenChanged", time_str);
440 if (ret != LDB_SUCCESS) {
441 talloc_free(down_req);
442 ldb_oom(module->ldb);
443 return LDB_ERR_OPERATIONS_ERROR;
444 }
445 ret = samdb_msg_add_uint64(module->ldb, msg, msg, "uSNCreated", seq_num);
446 if (ret != LDB_SUCCESS) {
447 talloc_free(down_req);
448 ldb_oom(module->ldb);
449 return LDB_ERR_OPERATIONS_ERROR;
450 }
451 ret = samdb_msg_add_uint64(module->ldb, msg, msg, "uSNChanged", seq_num);
452 if (ret != LDB_SUCCESS) {
453 talloc_free(down_req);
454 ldb_oom(module->ldb);
455 return LDB_ERR_OPERATIONS_ERROR;
456 }
457 ret = ldb_msg_add_value(msg, "replPropertyMetaData", &nmd_value, NULL);
458 if (ret != LDB_SUCCESS) {
459 talloc_free(down_req);
460 ldb_oom(module->ldb);
461 return LDB_ERR_OPERATIONS_ERROR;
462 }
463
464 /*
465 * sort the attributes by attid before storing the object
466 */
467 replmd_ldb_message_sort(msg, schema);
468
469 ldb_set_timeout_from_prev_req(module->ldb, req, down_req);
470
471 /* go on with the call chain */
472 ret = ldb_next_request(module, down_req);
473
474 /* do not free down_req as the call results may be linked to it,
475 * it will be freed when the upper level request get freed */
476 if (ret == LDB_SUCCESS) {
477 req->handle = down_req->handle;
478 }
479
480 return ret;
481 }
482
483 static int replmd_add(struct ldb_module *module, struct ldb_request *req)
484 {
485 return replmd_prepare_originating(module, req, req->op.add.message->dn,
486 "replmd_add", replmd_add_originating);
487 }
488
489 static int replmd_modify_originating(struct ldb_module *module,
490 struct ldb_request *req,
491 const struct dsdb_schema *schema)
492 {
493 struct ldb_request *down_req;
494 struct ldb_message *msg;
495 int ret;
496 time_t t = time(NULL);
497 uint64_t seq_num;
498
499 ldb_debug(module->ldb, LDB_DEBUG_TRACE, "replmd_modify_originating\n");
500
501 down_req = talloc(req, struct ldb_request);
502 if (down_req == NULL) {
503 return LDB_ERR_OPERATIONS_ERROR;
504 }
505
506 *down_req = *req;
507
508 /* we have to copy the message as the caller might have it as a const */
509 down_req->op.mod.message = msg = ldb_msg_copy_shallow(down_req, req->op.mod.message);
510 if (msg == NULL) {
511 talloc_free(down_req);
512 return LDB_ERR_OPERATIONS_ERROR;
513 }
514
515 /* TODO:
516 * - get the whole old object
517 * - if the old object doesn't exist report an error
518 * - give an error when a readonly attribute should
519 * be modified
520 * - merge the changed into the old object
521 * if the caller set values to the same value
522 * ignore the attribute, return success when no
523 * attribute was changed
524 * - calculate the new replPropertyMetaData attribute
525 */
526
527 if (add_time_element(msg, "whenChanged", t) != 0) {
528 talloc_free(down_req);
529 return LDB_ERR_OPERATIONS_ERROR;
530 }
531
532 /* Get a sequence number from the backend */
533 ret = ldb_sequence_number(module->ldb, LDB_SEQ_NEXT, &seq_num);
534 if (ret == LDB_SUCCESS) {
535 if (add_uint64_element(msg, "uSNChanged", seq_num) != 0) {
536 talloc_free(down_req);
537 return LDB_ERR_OPERATIONS_ERROR;
538 }
539 }
540
541 /* TODO:
542 * - sort the attributes by attid with replmd_ldb_message_sort()
543 * - replace the old object with the newly constructed one
544 */
545
546 ldb_set_timeout_from_prev_req(module->ldb, req, down_req);
547
548 /* go on with the call chain */
549 ret = ldb_next_request(module, down_req);
550
551 /* do not free down_req as the call results may be linked to it,
552 * it will be freed when the upper level request get freed */
553 if (ret == LDB_SUCCESS) {
554 req->handle = down_req->handle;
555 }
556
557 return ret;
558 }
559
560 static int replmd_modify(struct ldb_module *module, struct ldb_request *req)
561 {
562 return replmd_prepare_originating(module, req, req->op.mod.message->dn,
563 "replmd_modify", replmd_modify_originating);
564 }
565
566 static int replmd_replicated_request_reply_helper(struct replmd_replicated_request *ar, int ret)
567 {
568 struct ldb_reply *ares = NULL;
569
570 ar->handle->status = ret;
571 ar->handle->state = LDB_ASYNC_DONE;
572
573 if (!ar->orig_req->callback) {
574 return LDB_SUCCESS;
575 }
576
577 /* we're done and need to report the success to the caller */
578 ares = talloc_zero(ar, struct ldb_reply);
579 if (!ares) {
580 ar->handle->status = LDB_ERR_OPERATIONS_ERROR;
581 ar->handle->state = LDB_ASYNC_DONE;
582 return LDB_ERR_OPERATIONS_ERROR;
583 }
584
585 ares->type = LDB_REPLY_EXTENDED;
586 ares->response = NULL;
587
588 return ar->orig_req->callback(ar->module->ldb, ar->orig_req->context, ares);
589 }
590
591 static int replmd_replicated_request_done(struct replmd_replicated_request *ar)
592 {
593 return replmd_replicated_request_reply_helper(ar, LDB_SUCCESS);
594 }
595
596 static int replmd_replicated_request_error(struct replmd_replicated_request *ar, int ret)
597 {
598 return replmd_replicated_request_reply_helper(ar, ret);
599 }
600
601 static int replmd_replicated_request_werror(struct replmd_replicated_request *ar, WERROR status)
602 {
603 int ret = LDB_ERR_OTHER;
604 /* TODO: do some error mapping */
605 return replmd_replicated_request_reply_helper(ar, ret);
606 }
607
608 static int replmd_replicated_apply_next(struct replmd_replicated_request *ar);
609
610 static int replmd_replicated_apply_add_callback(struct ldb_context *ldb,
611 void *private_data,
612 struct ldb_reply *ares)
613 {
614 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
615 struct replmd_replicated_request *ar = talloc_get_type(private_data,
616 struct replmd_replicated_request);
617
618 ar->sub.change_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
619 if (ar->sub.change_ret != LDB_SUCCESS) {
620 return replmd_replicated_request_error(ar, ar->sub.change_ret);
621 }
622
623 talloc_free(ar->sub.mem_ctx);
624 ZERO_STRUCT(ar->sub);
625
626 ar->index_current++;
627
628 return replmd_replicated_apply_next(ar);
629 #else
630 return LDB_SUCCESS;
631 #endif
632 }
633
634 static int replmd_replicated_apply_add(struct replmd_replicated_request *ar)
635 {
636 enum ndr_err_code ndr_err;
637 struct ldb_message *msg;
638 struct replPropertyMetaDataBlob *md;
639 struct ldb_val md_value;
640 uint32_t i;
641 uint64_t seq_num;
642 int ret;
643
644 /*
645 * TODO: check if the parent object exist
646 */
647
648 /*
649 * TODO: handle the conflict case where an object with the
650 * same name exist
651 */
652
653 msg = ar->objs->objects[ar->index_current].msg;
654 md = ar->objs->objects[ar->index_current].meta_data;
655
656 ret = ldb_sequence_number(ar->module->ldb, LDB_SEQ_NEXT, &seq_num);
657 if (ret != LDB_SUCCESS) {
658 return replmd_replicated_request_error(ar, ret);
659 }
660
661 ret = ldb_msg_add_value(msg, "objectGUID", &ar->objs->objects[ar->index_current].guid_value, NULL);
662 if (ret != LDB_SUCCESS) {
663 return replmd_replicated_request_error(ar, ret);
664 }
665
666 ret = ldb_msg_add_string(msg, "whenChanged", ar->objs->objects[ar->index_current].when_changed);
667 if (ret != LDB_SUCCESS) {
668 return replmd_replicated_request_error(ar, ret);
669 }
670
671 ret = samdb_msg_add_uint64(ar->module->ldb, msg, msg, "uSNCreated", seq_num);
672 if (ret != LDB_SUCCESS) {
673 return replmd_replicated_request_error(ar, ret);
674 }
675
676 ret = samdb_msg_add_uint64(ar->module->ldb, msg, msg, "uSNChanged", seq_num);
677 if (ret != LDB_SUCCESS) {
678 return replmd_replicated_request_error(ar, ret);
679 }
680
681 /*
682 * the meta data array is already sorted by the caller
683 */
684 for (i=0; i < md->ctr.ctr1.count; i++) {
685 md->ctr.ctr1.array[i].local_usn = seq_num;
686 }
687 ndr_err = ndr_push_struct_blob(&md_value, msg,
688 lp_iconv_convenience(ldb_get_opaque(ar->module->ldb, "loadparm")),
689 md,
690 (ndr_push_flags_fn_t)ndr_push_replPropertyMetaDataBlob);
691 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
692 NTSTATUS nt_status = ndr_map_error2ntstatus(ndr_err);
693 return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
694 }
695 ret = ldb_msg_add_value(msg, "replPropertyMetaData", &md_value, NULL);
696 if (ret != LDB_SUCCESS) {
697 return replmd_replicated_request_error(ar, ret);
698 }
699
700 replmd_ldb_message_sort(msg, ar->schema);
701
702 ret = ldb_build_add_req(&ar->sub.change_req,
703 ar->module->ldb,
704 ar->sub.mem_ctx,
705 msg,
706 ar->controls,
707 ar,
708 replmd_replicated_apply_add_callback);
709 if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
710
711 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
712 return ldb_next_request(ar->module, ar->sub.change_req);
713 #else
714 ret = ldb_next_request(ar->module, ar->sub.change_req);
715 if (ret != LDB_SUCCESS) {
716 ldb_asprintf_errstring(ar->module->ldb, "Failed to add replicated object %s: %s", ldb_dn_get_linearized(ar->sub.change_req->op.add.message->dn),
717 ldb_errstring(ar->module->ldb));
718 return replmd_replicated_request_error(ar, ret);
719 }
720
721 ar->sub.change_ret = ldb_wait(ar->sub.change_req->handle, LDB_WAIT_ALL);
722 if (ar->sub.change_ret != LDB_SUCCESS) {
723 ldb_asprintf_errstring(ar->module->ldb, "Failed while waiting on add replicated object %s: %s", ldb_dn_get_linearized(ar->sub.change_req->op.add.message->dn),
724 ldb_errstring(ar->module->ldb));
725 return replmd_replicated_request_error(ar, ar->sub.change_ret);
726 }
727
728 talloc_free(ar->sub.mem_ctx);
729 ZERO_STRUCT(ar->sub);
730
731 ar->index_current++;
732
733 return LDB_SUCCESS;
734 #endif
735 }
736
737 static int replmd_replPropertyMetaData1_conflict_compare(struct replPropertyMetaData1 *m1,
738 struct replPropertyMetaData1 *m2)
739 {
740 int ret;
741
742 if (m1->version != m2->version) {
743 return m1->version - m2->version;
744 }
745
746 if (m1->originating_change_time != m2->originating_change_time) {
747 return m1->originating_change_time - m2->originating_change_time;
748 }
749
750 ret = GUID_compare(&m1->originating_invocation_id, &m2->originating_invocation_id);
751 if (ret != 0) {
752 return ret;
753 }
754
755 return m1->originating_usn - m2->originating_usn;
756 }
757
758 static int replmd_replicated_apply_merge_callback(struct ldb_context *ldb,
759 void *private_data,
760 struct ldb_reply *ares)
761 {
762 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
763 struct replmd_replicated_request *ar = talloc_get_type(private_data,
764 struct replmd_replicated_request);
765
766 ret = ldb_next_request(ar->module, ar->sub.change_req);
767 if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
768
769 ar->sub.change_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
770 if (ar->sub.change_ret != LDB_SUCCESS) {
771 return replmd_replicated_request_error(ar, ar->sub.change_ret);
772 }
773
774 talloc_free(ar->sub.mem_ctx);
775 ZERO_STRUCT(ar->sub);
776
777 ar->index_current++;
778
779 return LDB_SUCCESS;
780 #else
781 return LDB_SUCCESS;
782 #endif
783 }
784
785 static int replmd_replicated_apply_merge(struct replmd_replicated_request *ar)
786 {
787 enum ndr_err_code ndr_err;
788 struct ldb_message *msg;
789 struct replPropertyMetaDataBlob *rmd;
790 struct replPropertyMetaDataBlob omd;
791 const struct ldb_val *omd_value;
792 struct replPropertyMetaDataBlob nmd;
793 struct ldb_val nmd_value;
794 uint32_t i,j,ni=0;
795 uint32_t removed_attrs = 0;
796 uint64_t seq_num;
797 int ret;
798
799 msg = ar->objs->objects[ar->index_current].msg;
800 rmd = ar->objs->objects[ar->index_current].meta_data;
801 ZERO_STRUCT(omd);
802 omd.version = 1;
803
804 /*
805 * TODO: add rename conflict handling
806 */
807 if (ldb_dn_compare(msg->dn, ar->sub.search_msg->dn) != 0) {
808 ldb_debug_set(ar->module->ldb, LDB_DEBUG_FATAL, "replmd_replicated_apply_merge[%u]: rename not supported",
809 ar->index_current);
810 ldb_debug(ar->module->ldb, LDB_DEBUG_FATAL, "%s => %s\n",
811 ldb_dn_get_linearized(ar->sub.search_msg->dn),
812 ldb_dn_get_linearized(msg->dn));
813 return replmd_replicated_request_werror(ar, WERR_NOT_SUPPORTED);
814 }
815
816 ret = ldb_sequence_number(ar->module->ldb, LDB_SEQ_NEXT, &seq_num);
817 if (ret != LDB_SUCCESS) {
818 return replmd_replicated_request_error(ar, ret);
819 }
820
821 /* find existing meta data */
822 omd_value = ldb_msg_find_ldb_val(ar->sub.search_msg, "replPropertyMetaData");
823 if (omd_value) {
824 ndr_err = ndr_pull_struct_blob(omd_value, ar->sub.mem_ctx,
825 lp_iconv_convenience(ldb_get_opaque(ar->module->ldb, "loadparm")), &omd,
826 (ndr_pull_flags_fn_t)ndr_pull_replPropertyMetaDataBlob);
827 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
828 NTSTATUS nt_status = ndr_map_error2ntstatus(ndr_err);
829 return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
830 }
831
832 if (omd.version != 1) {
833 return replmd_replicated_request_werror(ar, WERR_DS_DRA_INTERNAL_ERROR);
834 }
835 }
836
837 ZERO_STRUCT(nmd);
838 nmd.version = 1;
839 nmd.ctr.ctr1.count = omd.ctr.ctr1.count + rmd->ctr.ctr1.count;
840 nmd.ctr.ctr1.array = talloc_array(ar->sub.mem_ctx,
841 struct replPropertyMetaData1,
842 nmd.ctr.ctr1.count);
843 if (!nmd.ctr.ctr1.array) return replmd_replicated_request_werror(ar, WERR_NOMEM);
844
845 /* first copy the old meta data */
846 for (i=0; i < omd.ctr.ctr1.count; i++) {
847 nmd.ctr.ctr1.array[ni] = omd.ctr.ctr1.array[i];
848 ni++;
849 }
850
851 /* now merge in the new meta data */
852 for (i=0; i < rmd->ctr.ctr1.count; i++) {
853 bool found = false;
854
855 rmd->ctr.ctr1.array[i].local_usn = seq_num;
856
857 for (j=0; j < ni; j++) {
858 int cmp;
859
860 if (rmd->ctr.ctr1.array[i].attid != nmd.ctr.ctr1.array[j].attid) {
861 continue;
862 }
863
864 cmp = replmd_replPropertyMetaData1_conflict_compare(&rmd->ctr.ctr1.array[i],
865 &nmd.ctr.ctr1.array[j]);
866 if (cmp > 0) {
867 /* replace the entry */
868 nmd.ctr.ctr1.array[j] = rmd->ctr.ctr1.array[i];
869 found = true;
870 break;
871 }
872
873 /* we don't want to apply this change so remove the attribute */
874 ldb_msg_remove_element(msg, &msg->elements[i-removed_attrs]);
875 removed_attrs++;
876
877 found = true;
878 break;
879 }
880
881 if (found) continue;
882
883 nmd.ctr.ctr1.array[ni] = rmd->ctr.ctr1.array[i];
884 ni++;
885 }
886
887 /*
888 * finally correct the size of the meta_data array
889 */
890 nmd.ctr.ctr1.count = ni;
891
892 /*
893 * the rdn attribute (the alias for the name attribute),
894 * 'cn' for most objects is the last entry in the meta data array
895 * we have stored
896 *
897 * sort the new meta data array
898 */
899 {
900 struct replPropertyMetaData1 *rdn_p;
901 uint32_t rdn_idx = omd.ctr.ctr1.count - 1;
902
903 rdn_p = &nmd.ctr.ctr1.array[rdn_idx];
904 replmd_replPropertyMetaDataCtr1_sort(&nmd.ctr.ctr1, &rdn_p->attid);
905 }
906
907 /* create the meta data value */
908 ndr_err = ndr_push_struct_blob(&nmd_value, msg,
909 lp_iconv_convenience(ldb_get_opaque(ar->module->ldb, "loadparm")),
910 &nmd,
911 (ndr_push_flags_fn_t)ndr_push_replPropertyMetaDataBlob);
912 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
913 NTSTATUS nt_status = ndr_map_error2ntstatus(ndr_err);
914 return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
915 }
916
917 /*
918 * check if some replicated attributes left, otherwise skip the ldb_modify() call
919 */
920 if (msg->num_elements == 0) {
921 ldb_debug(ar->module->ldb, LDB_DEBUG_TRACE, "replmd_replicated_apply_merge[%u]: skip replace\n",
922 ar->index_current);
923 goto next_object;
924 }
925
926 ldb_debug(ar->module->ldb, LDB_DEBUG_TRACE, "replmd_replicated_apply_merge[%u]: replace %u attributes\n",
927 ar->index_current, msg->num_elements);
928
929 /*
930 * when we now that we'll modify the record, add the whenChanged, uSNChanged
931 * and replPopertyMetaData attributes
932 */
933 ret = ldb_msg_add_string(msg, "whenChanged", ar->objs->objects[ar->index_current].when_changed);
934 if (ret != LDB_SUCCESS) {
935 return replmd_replicated_request_error(ar, ret);
936 }
937 ret = samdb_msg_add_uint64(ar->module->ldb, msg, msg, "uSNChanged", seq_num);
938 if (ret != LDB_SUCCESS) {
939 return replmd_replicated_request_error(ar, ret);
940 }
941 ret = ldb_msg_add_value(msg, "replPropertyMetaData", &nmd_value, NULL);
942 if (ret != LDB_SUCCESS) {
943 return replmd_replicated_request_error(ar, ret);
944 }
945
946 replmd_ldb_message_sort(msg, ar->schema);
947
948 /* we want to replace the old values */
949 for (i=0; i < msg->num_elements; i++) {
950 msg->elements[i].flags = LDB_FLAG_MOD_REPLACE;
951 }
952
953 ret = ldb_build_mod_req(&ar->sub.change_req,
954 ar->module->ldb,
955 ar->sub.mem_ctx,
956 msg,
957 ar->controls,
958 ar,
959 replmd_replicated_apply_merge_callback);
960 if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
961
962 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
963 return ldb_next_request(ar->module, ar->sub.change_req);
964 #else
965 ret = ldb_next_request(ar->module, ar->sub.change_req);
966 if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
967
968 ar->sub.change_ret = ldb_wait(ar->sub.change_req->handle, LDB_WAIT_ALL);
969 if (ar->sub.change_ret != LDB_SUCCESS) {
970 return replmd_replicated_request_error(ar, ar->sub.change_ret);
971 }
972
973 next_object:
974 talloc_free(ar->sub.mem_ctx);
975 ZERO_STRUCT(ar->sub);
976
977 ar->index_current++;
978
979 return LDB_SUCCESS;
980 #endif
981 }
982
983 static int replmd_replicated_apply_search_callback(struct ldb_context *ldb,
984 void *private_data,
985 struct ldb_reply *ares)
986 {
987 struct replmd_replicated_request *ar = talloc_get_type(private_data,
988 struct replmd_replicated_request);
989 bool is_done = false;
990
991 switch (ares->type) {
992 case LDB_REPLY_ENTRY:
993 ar->sub.search_msg = talloc_steal(ar->sub.mem_ctx, ares->message);
994 break;
995 case LDB_REPLY_REFERRAL:
996 /* we ignore referrals */
997 break;
998 case LDB_REPLY_EXTENDED:
999 case LDB_REPLY_DONE:
1000 is_done = true;
1001 }
1002
1003 talloc_free(ares);
1004
1005 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1006 if (is_done) {
1007 ar->sub.search_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
1008 if (ar->sub.search_ret != LDB_SUCCESS) {
1009 return replmd_replicated_request_error(ar, ar->sub.search_ret);
1010 }
1011 if (ar->sub.search_msg) {
1012 return replmd_replicated_apply_merge(ar);
1013 }
1014 return replmd_replicated_apply_add(ar);
1015 }
1016 #endif
1017 return LDB_SUCCESS;
1018 }
1019
1020 static int replmd_replicated_apply_search(struct replmd_replicated_request *ar)
1021 {
1022 int ret;
1023 char *tmp_str;
1024 char *filter;
1025
1026 tmp_str = ldb_binary_encode(ar->sub.mem_ctx, ar->objs->objects[ar->index_current].guid_value);
1027 if (!tmp_str) return replmd_replicated_request_werror(ar, WERR_NOMEM);
1028
1029 filter = talloc_asprintf(ar->sub.mem_ctx, "(objectGUID=%s)", tmp_str);
1030 if (!filter) return replmd_replicated_request_werror(ar, WERR_NOMEM);
1031 talloc_free(tmp_str);
1032
1033 ret = ldb_build_search_req(&ar->sub.search_req,
1034 ar->module->ldb,
1035 ar->sub.mem_ctx,
1036 ar->objs->partition_dn,
1037 LDB_SCOPE_SUBTREE,
1038 filter,
1039 NULL,
1040 NULL,
1041 ar,
1042 replmd_replicated_apply_search_callback);
1043 if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1044
1045 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1046 return ldb_next_request(ar->module, ar->sub.search_req);
1047 #else
1048 ret = ldb_next_request(ar->module, ar->sub.search_req);
1049 if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1050
1051 ar->sub.search_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
1052 if (ar->sub.search_ret != LDB_SUCCESS && ar->sub.search_ret != LDB_ERR_NO_SUCH_OBJECT) {
1053 return replmd_replicated_request_error(ar, ar->sub.search_ret);
1054 }
1055 if (ar->sub.search_msg) {
1056 return replmd_replicated_apply_merge(ar);
1057 }
1058
1059 return replmd_replicated_apply_add(ar);
1060 #endif
1061 }
1062
1063 static int replmd_replicated_apply_next(struct replmd_replicated_request *ar)
1064 {
1065 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1066 if (ar->index_current >= ar->objs->num_objects) {
1067 return replmd_replicated_uptodate_vector(ar);
1068 }
1069 #endif
1070
1071 ar->sub.mem_ctx = talloc_new(ar);
1072 if (!ar->sub.mem_ctx) return replmd_replicated_request_werror(ar, WERR_NOMEM);
1073
1074 return replmd_replicated_apply_search(ar);
1075 }
1076
1077 static int replmd_replicated_uptodate_modify_callback(struct ldb_context *ldb,
1078 void *private_data,
1079 struct ldb_reply *ares)
1080 {
1081 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1082 struct replmd_replicated_request *ar = talloc_get_type(private_data,
1083 struct replmd_replicated_request);
1084
1085 ar->sub.change_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
1086 if (ar->sub.change_ret != LDB_SUCCESS) {
1087 return replmd_replicated_request_error(ar, ar->sub.change_ret);
1088 }
1089
1090 talloc_free(ar->sub.mem_ctx);
1091 ZERO_STRUCT(ar->sub);
1092
1093 return replmd_replicated_request_done(ar);
1094 #else
1095 return LDB_SUCCESS;
1096 #endif
1097 }
1098
1099 static int replmd_drsuapi_DsReplicaCursor2_compare(const struct drsuapi_DsReplicaCursor2 *c1,
1100 const struct drsuapi_DsReplicaCursor2 *c2)
1101 {
1102 return GUID_compare(&c1->source_dsa_invocation_id, &c2->source_dsa_invocation_id);
1103 }
1104
1105 static int replmd_replicated_uptodate_modify(struct replmd_replicated_request *ar)
1106 {
1107 enum ndr_err_code ndr_err;
1108 struct ldb_message *msg;
1109 struct replUpToDateVectorBlob ouv;
1110 const struct ldb_val *ouv_value;
1111 const struct drsuapi_DsReplicaCursor2CtrEx *ruv;
1112 struct replUpToDateVectorBlob nuv;
1113 struct ldb_val nuv_value;
1114 struct ldb_message_element *nuv_el = NULL;
1115 const struct GUID *our_invocation_id;
1116 struct ldb_message_element *orf_el = NULL;
1117 struct repsFromToBlob nrf;
1118 struct ldb_val *nrf_value = NULL;
1119 struct ldb_message_element *nrf_el = NULL;
1120 uint32_t i,j,ni=0;
1121 uint64_t seq_num;
1122 bool found = false;
1123 time_t t = time(NULL);
1124 NTTIME now;
1125 int ret;
1126
1127 ruv = ar->objs->uptodateness_vector;
1128 ZERO_STRUCT(ouv);
1129 ouv.version = 2;
1130 ZERO_STRUCT(nuv);
1131 nuv.version = 2;
1132
1133 unix_to_nt_time(&now, t);
1134
1135 /*
1136 * we use the next sequence number for our own highest_usn
1137 * because we will do a modify request and this will increment
1138 * our highest_usn
1139 */
1140 ret = ldb_sequence_number(ar->module->ldb, LDB_SEQ_NEXT, &seq_num);
1141 if (ret != LDB_SUCCESS) {
1142 return replmd_replicated_request_error(ar, ret);
1143 }
1144
1145 /*
1146 * first create the new replUpToDateVector
1147 */
1148 ouv_value = ldb_msg_find_ldb_val(ar->sub.search_msg, "replUpToDateVector");
1149 if (ouv_value) {
1150 ndr_err = ndr_pull_struct_blob(ouv_value, ar->sub.mem_ctx,
1151 lp_iconv_convenience(ldb_get_opaque(ar->module->ldb, "loadparm")), &ouv,
1152 (ndr_pull_flags_fn_t)ndr_pull_replUpToDateVectorBlob);
1153 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
1154 NTSTATUS nt_status = ndr_map_error2ntstatus(ndr_err);
1155 return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
1156 }
1157
1158 if (ouv.version != 2) {
1159 return replmd_replicated_request_werror(ar, WERR_DS_DRA_INTERNAL_ERROR);
1160 }
1161 }
1162
1163 /*
1164 * the new uptodateness vector will at least
1165 * contain 1 entry, one for the source_dsa
1166 *
1167 * plus optional values from our old vector and the one from the source_dsa
1168 */
1169 nuv.ctr.ctr2.count = 1 + ouv.ctr.ctr2.count;
1170 if (ruv) nuv.ctr.ctr2.count += ruv->count;
1171 nuv.ctr.ctr2.cursors = talloc_array(ar->sub.mem_ctx,
1172 struct drsuapi_DsReplicaCursor2,
1173 nuv.ctr.ctr2.count);
1174 if (!nuv.ctr.ctr2.cursors) return replmd_replicated_request_werror(ar, WERR_NOMEM);
1175
1176 /* first copy the old vector */
1177 for (i=0; i < ouv.ctr.ctr2.count; i++) {
1178 nuv.ctr.ctr2.cursors[ni] = ouv.ctr.ctr2.cursors[i];
1179 ni++;
1180 }
1181
1182 /* get our invocation_id if we have one already attached to the ldb */
1183 our_invocation_id = samdb_ntds_invocation_id(ar->module->ldb);
1184
1185 /* merge in the source_dsa vector is available */
1186 for (i=0; (ruv && i < ruv->count); i++) {
1187 found = false;
1188
1189 if (our_invocation_id &&
1190 GUID_equal(&ruv->cursors[i].source_dsa_invocation_id,
1191 our_invocation_id)) {
1192 continue;
1193 }
1194
1195 for (j=0; j < ni; j++) {
1196 if (!GUID_equal(&ruv->cursors[i].source_dsa_invocation_id,
1197 &nuv.ctr.ctr2.cursors[j].source_dsa_invocation_id)) {
1198 continue;
1199 }
1200
1201 found = true;
1202
1203 /*
1204 * we update only the highest_usn and not the latest_sync_success time,
1205 * because the last success stands for direct replication
1206 */
1207 if (ruv->cursors[i].highest_usn > nuv.ctr.ctr2.cursors[j].highest_usn) {
1208 nuv.ctr.ctr2.cursors[j].highest_usn = ruv->cursors[i].highest_usn;
1209 }
1210 break;
1211 }
1212
1213 if (found) continue;
1214
1215 /* if it's not there yet, add it */
1216 nuv.ctr.ctr2.cursors[ni] = ruv->cursors[i];
1217 ni++;
1218 }
1219
1220 /*
1221 * merge in the current highwatermark for the source_dsa
1222 */
1223 found = false;
1224 for (j=0; j < ni; j++) {
1225 if (!GUID_equal(&ar->objs->source_dsa->source_dsa_invocation_id,
1226 &nuv.ctr.ctr2.cursors[j].source_dsa_invocation_id)) {
1227 continue;
1228 }
1229
1230 found = true;
1231
1232 /*
1233 * here we update the highest_usn and last_sync_success time
1234 * because we're directly replicating from the source_dsa
1235 *
1236 * and use the tmp_highest_usn because this is what we have just applied
1237 * to our ldb
1238 */
1239 nuv.ctr.ctr2.cursors[j].highest_usn = ar->objs->source_dsa->highwatermark.tmp_highest_usn;
1240 nuv.ctr.ctr2.cursors[j].last_sync_success = now;
1241 break;
1242 }
1243 if (!found) {
1244 /*
1245 * here we update the highest_usn and last_sync_success time
1246 * because we're directly replicating from the source_dsa
1247 *
1248 * and use the tmp_highest_usn because this is what we have just applied
1249 * to our ldb
1250 */
1251 nuv.ctr.ctr2.cursors[ni].source_dsa_invocation_id= ar->objs->source_dsa->source_dsa_invocation_id;
1252 nuv.ctr.ctr2.cursors[ni].highest_usn = ar->objs->source_dsa->highwatermark.tmp_highest_usn;
1253 nuv.ctr.ctr2.cursors[ni].last_sync_success = now;
1254 ni++;
1255 }
1256
1257 /*
1258 * finally correct the size of the cursors array
1259 */
1260 nuv.ctr.ctr2.count = ni;
1261
1262 /*
1263 * sort the cursors
1264 */
1265 qsort(nuv.ctr.ctr2.cursors, nuv.ctr.ctr2.count,
1266 sizeof(struct drsuapi_DsReplicaCursor2),
1267 (comparison_fn_t)replmd_drsuapi_DsReplicaCursor2_compare);
1268
1269 /*
1270 * create the change ldb_message
1271 */
1272 msg = ldb_msg_new(ar->sub.mem_ctx);
1273 if (!msg) return replmd_replicated_request_werror(ar, WERR_NOMEM);
1274 msg->dn = ar->sub.search_msg->dn;
1275
1276 ndr_err = ndr_push_struct_blob(&nuv_value, msg,
1277 lp_iconv_convenience(ldb_get_opaque(ar->module->ldb, "loadparm")),
1278 &nuv,
1279 (ndr_push_flags_fn_t)ndr_push_replUpToDateVectorBlob);
1280 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
1281 NTSTATUS nt_status = ndr_map_error2ntstatus(ndr_err);
1282 return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
1283 }
1284 ret = ldb_msg_add_value(msg, "replUpToDateVector", &nuv_value, &nuv_el);
1285 if (ret != LDB_SUCCESS) {
1286 return replmd_replicated_request_error(ar, ret);
1287 }
1288 nuv_el->flags = LDB_FLAG_MOD_REPLACE;
1289
1290 /*
1291 * now create the new repsFrom value from the given repsFromTo1 structure
1292 */
1293 ZERO_STRUCT(nrf);
1294 nrf.version = 1;
1295 nrf.ctr.ctr1 = *ar->objs->source_dsa;
1296 /* and fix some values... */
1297 nrf.ctr.ctr1.consecutive_sync_failures = 0;
1298 nrf.ctr.ctr1.last_success = now;
1299 nrf.ctr.ctr1.last_attempt = now;
1300 nrf.ctr.ctr1.result_last_attempt = WERR_OK;
1301 nrf.ctr.ctr1.highwatermark.highest_usn = nrf.ctr.ctr1.highwatermark.tmp_highest_usn;
1302
1303 /*
1304 * first see if we already have a repsFrom value for the current source dsa
1305 * if so we'll later replace this value
1306 */
1307 orf_el = ldb_msg_find_element(ar->sub.search_msg, "repsFrom");
1308 if (orf_el) {
1309 for (i=0; i < orf_el->num_values; i++) {
1310 struct repsFromToBlob *trf;
1311
1312 trf = talloc(ar->sub.mem_ctx, struct repsFromToBlob);
1313 if (!trf) return replmd_replicated_request_werror(ar, WERR_NOMEM);
1314
1315 ndr_err = ndr_pull_struct_blob(&orf_el->values[i], trf, lp_iconv_convenience(ldb_get_opaque(ar->module->ldb, "loadparm")), trf,
1316 (ndr_pull_flags_fn_t)ndr_pull_repsFromToBlob);
1317 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
1318 NTSTATUS nt_status = ndr_map_error2ntstatus(ndr_err);
1319 return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
1320 }
1321
1322 if (trf->version != 1) {
1323 return replmd_replicated_request_werror(ar, WERR_DS_DRA_INTERNAL_ERROR);
1324 }
1325
1326 /*
1327 * we compare the source dsa objectGUID not the invocation_id
1328 * because we want only one repsFrom value per source dsa
1329 * and when the invocation_id of the source dsa has changed we don't need
1330 * the old repsFrom with the old invocation_id
1331 */
1332 if (!GUID_equal(&trf->ctr.ctr1.source_dsa_obj_guid,
1333 &ar->objs->source_dsa->source_dsa_obj_guid)) {
1334 talloc_free(trf);
1335 continue;
1336 }
1337
1338 talloc_free(trf);
1339 nrf_value = &orf_el->values[i];
1340 break;
1341 }
1342
1343 /*
1344 * copy over all old values to the new ldb_message
1345 */
1346 ret = ldb_msg_add_empty(msg, "repsFrom", 0, &nrf_el);
1347 if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1348 *nrf_el = *orf_el;
1349 }
1350
1351 /*
1352 * if we haven't found an old repsFrom value for the current source dsa
1353 * we'll add a new value
1354 */
1355 if (!nrf_value) {
1356 struct ldb_val zero_value;
1357 ZERO_STRUCT(zero_value);
1358 ret = ldb_msg_add_value(msg, "repsFrom", &zero_value, &nrf_el);
1359 if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1360
1361 nrf_value = &nrf_el->values[nrf_el->num_values - 1];
1362 }
1363
1364 /* we now fill the value which is already attached to ldb_message */
1365 ndr_err = ndr_push_struct_blob(nrf_value, msg,
1366 lp_iconv_convenience(ldb_get_opaque(ar->module->ldb, "loadparm")),
1367 &nrf,
1368 (ndr_push_flags_fn_t)ndr_push_repsFromToBlob);
1369 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
1370 NTSTATUS nt_status = ndr_map_error2ntstatus(ndr_err);
1371 return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
1372 }
1373
1374 /*
1375 * the ldb_message_element for the attribute, has all the old values and the new one
1376 * so we'll replace the whole attribute with all values
1377 */
1378 nrf_el->flags = LDB_FLAG_MOD_REPLACE;
1379
1380 /* prepare the ldb_modify() request */
1381 ret = ldb_build_mod_req(&ar->sub.change_req,
1382 ar->module->ldb,
1383 ar->sub.mem_ctx,
1384 msg,
1385 ar->controls,
1386 ar,
1387 replmd_replicated_uptodate_modify_callback);
1388 if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1389
1390 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1391 return ldb_next_request(ar->module, ar->sub.change_req);
1392 #else
1393 ret = ldb_next_request(ar->module, ar->sub.change_req);
1394 if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1395
1396 ar->sub.change_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
1397 if (ar->sub.change_ret != LDB_SUCCESS) {
1398 return replmd_replicated_request_error(ar, ar->sub.change_ret);
1399 }
1400
1401 talloc_free(ar->sub.mem_ctx);
1402 ZERO_STRUCT(ar->sub);
1403
1404 return replmd_replicated_request_done(ar);
1405 #endif
1406 }
1407
1408 static int replmd_replicated_uptodate_search_callback(struct ldb_context *ldb,
1409 void *private_data,
1410 struct ldb_reply *ares)
1411 {
1412 struct replmd_replicated_request *ar = talloc_get_type(private_data,
1413 struct replmd_replicated_request);
1414 bool is_done = false;
1415
1416 switch (ares->type) {
1417 case LDB_REPLY_ENTRY:
1418 ar->sub.search_msg = talloc_steal(ar->sub.mem_ctx, ares->message);
1419 break;
1420 case LDB_REPLY_REFERRAL:
1421 /* we ignore referrals */
1422 break;
1423 case LDB_REPLY_EXTENDED:
1424 case LDB_REPLY_DONE:
1425 is_done = true;
1426 }
1427
1428 talloc_free(ares);
1429
1430 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1431 if (is_done) {
1432 ar->sub.search_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
1433 if (ar->sub.search_ret != LDB_SUCCESS) {
1434 return replmd_replicated_request_error(ar, ar->sub.search_ret);
1435 }
1436 if (!ar->sub.search_msg) {
1437 return replmd_replicated_request_werror(ar, WERR_DS_DRA_INTERNAL_ERROR);
1438 }
1439
1440 return replmd_replicated_uptodate_modify(ar);
1441 }
1442 #endif
1443 return LDB_SUCCESS;
1444 }
1445
1446 static int replmd_replicated_uptodate_search(struct replmd_replicated_request *ar)
1447 {
1448 int ret;
1449 static const char *attrs[] = {
1450 "replUpToDateVector",
1451 "repsFrom",
1452 NULL
1453 };
1454
1455 ret = ldb_build_search_req(&ar->sub.search_req,
1456 ar->module->ldb,
1457 ar->sub.mem_ctx,
1458 ar->objs->partition_dn,
1459 LDB_SCOPE_BASE,
1460 "(objectClass=*)",
1461 attrs,
1462 NULL,
1463 ar,
1464 replmd_replicated_uptodate_search_callback);
1465 if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1466
1467 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1468 return ldb_next_request(ar->module, ar->sub.search_req);
1469 #else
1470 ret = ldb_next_request(ar->module, ar->sub.search_req);
1471 if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1472
1473 ar->sub.search_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
1474 if (ar->sub.search_ret != LDB_SUCCESS) {
1475 return replmd_replicated_request_error(ar, ar->sub.search_ret);
1476 }
1477 if (!ar->sub.search_msg) {
1478 return replmd_replicated_request_werror(ar, WERR_DS_DRA_INTERNAL_ERROR);
1479 }
1480
1481 return replmd_replicated_uptodate_modify(ar);
1482 #endif
1483 }
1484
1485 static int replmd_replicated_uptodate_vector(struct replmd_replicated_request *ar)
1486 {
1487 ar->sub.mem_ctx = talloc_new(ar);
1488 if (!ar->sub.mem_ctx) return replmd_replicated_request_werror(ar, WERR_NOMEM);
1489
1490 return replmd_replicated_uptodate_search(ar);
1491 }
1492
1493 static int replmd_extended_replicated_objects(struct ldb_module *module, struct ldb_request *req)
1494 {
1495 struct dsdb_extended_replicated_objects *objs;
1496 struct replmd_replicated_request *ar;
1497 struct ldb_control **ctrls;
1498 int ret;
1499
1500 ldb_debug(module->ldb, LDB_DEBUG_TRACE, "replmd_extended_replicated_objects\n");
1501
1502 objs = talloc_get_type(req->op.extended.data, struct dsdb_extended_replicated_objects);
1503 if (!objs) {
1504 ldb_debug(module->ldb, LDB_DEBUG_FATAL, "replmd_extended_replicated_objects: invalid extended data\n");
1505 return LDB_ERR_PROTOCOL_ERROR;
1506 }
1507
1508 if (objs->version != DSDB_EXTENDED_REPLICATED_OBJECTS_VERSION) {
1509 ldb_debug(module->ldb, LDB_DEBUG_FATAL, "replmd_extended_replicated_objects: extended data invalid version [%u != %u]\n",
1510 objs->version, DSDB_EXTENDED_REPLICATED_OBJECTS_VERSION);
1511 return LDB_ERR_PROTOCOL_ERROR;
1512 }
1513
1514 ar = replmd_replicated_init_handle(module, req, objs);
1515 if (!ar) {
1516 return LDB_ERR_OPERATIONS_ERROR;
1517 }
1518
1519 ctrls = req->controls;
1520
1521 if (req->controls) {
1522 req->controls = talloc_memdup(ar, req->controls,
1523 talloc_get_size(req->controls));
1524 if (!req->controls) return replmd_replicated_request_werror(ar, WERR_NOMEM);
1525 }
1526
1527 ret = ldb_request_add_control(req, DSDB_CONTROL_REPLICATED_UPDATE_OID, false, NULL);
1528 if (ret != LDB_SUCCESS) {
1529 return ret;
1530 }
1531
1532 ar->controls = req->controls;
1533 req->controls = ctrls;
1534
1535 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1536 return replmd_replicated_apply_next(ar);
1537 #else
1538 while (ar->index_current < ar->objs->num_objects &&
1539 req->handle->state != LDB_ASYNC_DONE) {
1540 replmd_replicated_apply_next(ar);
1541 }
1542
1543 if (req->handle->state != LDB_ASYNC_DONE) {
1544 replmd_replicated_uptodate_vector(ar);
1545 }
1546
1547 return LDB_SUCCESS;
1548 #endif
1549 }
1550
1551 static int replmd_extended(struct ldb_module *module, struct ldb_request *req)
1552 {
1553 if (strcmp(req->op.extended.oid, DSDB_EXTENDED_REPLICATED_OBJECTS_OID) == 0) {
1554 return replmd_extended_replicated_objects(module, req);
1555 }
1556
1557 return ldb_next_request(module, req);
1558 }
1559
1560 static int replmd_wait_none(struct ldb_handle *handle) {
1561 struct replmd_replicated_request *ar;
1562
1563 if (!handle || !handle->private_data) {
1564 return LDB_ERR_OPERATIONS_ERROR;
1565 }
1566
1567 ar = talloc_get_type(handle->private_data, struct replmd_replicated_request);
1568 if (!ar) {
1569 return LDB_ERR_OPERATIONS_ERROR;
1570 }
1571
1572 /* we do only sync calls */
1573 if (handle->state != LDB_ASYNC_DONE) {
1574 return LDB_ERR_OPERATIONS_ERROR;
1575 }
1576
1577 return handle->status;
1578 }
1579
1580 static int replmd_wait_all(struct ldb_handle *handle) {
1581
1582 int ret;
1583
1584 while (handle->state != LDB_ASYNC_DONE) {
1585 ret = replmd_wait_none(handle);
1586 if (ret != LDB_SUCCESS) {
1587 return ret;
1588 }
1589 }
1590
1591 return handle->status;
1592 }
1593
1594 static int replmd_wait(struct ldb_handle *handle, enum ldb_wait_type type)
1595 {
1596 if (type == LDB_WAIT_ALL) {
1597 return replmd_wait_all(handle);
1598 } else {
1599 return replmd_wait_none(handle);
1600 }
1601 }
1602
1603 _PUBLIC_ const struct ldb_module_ops ldb_repl_meta_data_module_ops = {
1604 .name = "repl_meta_data",
1605 .add = replmd_add,
1606 .modify = replmd_modify,
1607 .extended = replmd_extended,
1608 .wait = replmd_wait
1609 };