c1ea5ad90f88e7e3c7012ba9d96ac772dc6a030e
[samba.git] / source4 / dsdb / samdb / ldb_modules / repl_meta_data.c
1 /*
2    ldb database library
3
4    Copyright (C) Simo Sorce  2004-2008
5    Copyright (C) Andrew Bartlett <abartlet@samba.org> 2005-2013
6    Copyright (C) Andrew Tridgell 2005-2009
7    Copyright (C) Stefan Metzmacher <metze@samba.org> 2007
8    Copyright (C) Matthieu Patou <mat@samba.org> 2010-2011
9
10    This program is free software; you can redistribute it and/or modify
11    it under the terms of the GNU General Public License as published by
12    the Free Software Foundation; either version 3 of the License, or
13    (at your option) any later version.
14
15    This program is distributed in the hope that it will be useful,
16    but WITHOUT ANY WARRANTY; without even the implied warranty of
17    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
18    GNU General Public License for more details.
19
20    You should have received a copy of the GNU General Public License
21    along with this program.  If not, see <http://www.gnu.org/licenses/>.
22 */
23
24 /*
25  *  Name: ldb
26  *
27  *  Component: ldb repl_meta_data module
28  *
29  *  Description: - add a unique objectGUID onto every new record,
30  *               - handle whenCreated, whenChanged timestamps
31  *               - handle uSNCreated, uSNChanged numbers
32  *               - handle replPropertyMetaData attribute
33  *
34  *  Author: Simo Sorce
35  *  Author: Stefan Metzmacher
36  */
37
38 #include "includes.h"
39 #include "ldb_module.h"
40 #include "dsdb/samdb/samdb.h"
41 #include "dsdb/common/proto.h"
42 #include "dsdb/common/util.h"
43 #include "../libds/common/flags.h"
44 #include "librpc/gen_ndr/irpc.h"
45 #include "librpc/gen_ndr/ndr_misc.h"
46 #include "librpc/gen_ndr/ndr_drsuapi.h"
47 #include "librpc/gen_ndr/ndr_drsblobs.h"
48 #include "param/param.h"
49 #include "libcli/security/security.h"
50 #include "lib/util/dlinklist.h"
51 #include "dsdb/samdb/ldb_modules/util.h"
52 #include "lib/util/tsort.h"
53 #include "lib/util/binsearch.h"
54
55 #undef strcasecmp
56
57 #undef DBGC_CLASS
58 #define DBGC_CLASS            DBGC_DRS_REPL
59
60 /* the RMD_VERSION for linked attributes starts from 1 */
61 #define RMD_VERSION_INITIAL   1
62
63 /*
64  * It's 29/12/9999 at 23:59:59 UTC as specified in MS-ADTS 7.1.1.4.2
65  * Deleted Objects Container
66  */
67 static const NTTIME DELETED_OBJECT_CONTAINER_CHANGE_TIME = 2650466015990000000ULL;
68
69 struct replmd_private {
70         TALLOC_CTX *la_ctx;
71         struct la_group *la_list;
72         struct nc_entry {
73                 struct nc_entry *prev, *next;
74                 struct ldb_dn *dn;
75                 uint64_t mod_usn;
76                 uint64_t mod_usn_urgent;
77         } *ncs;
78         struct ldb_dn *schema_dn;
79         bool originating_updates;
80         bool sorted_links;
81         uint32_t total_links;
82         uint32_t num_processed;
83         bool recyclebin_enabled;
84         bool recyclebin_state_known;
85 };
86
87 /*
88  * groups link attributes together by source-object and attribute-ID,
89  * to improve processing efficiency (i.e. for 'member' attribute, which
90  * could have 100s or 1000s of links).
91  * Note this grouping is best effort - the same source object could still
92  * correspond to several la_groups (a lot depends on the order DRS sends
93  * the links in). The groups currently don't span replication chunks (which
94  * caps the size to ~1500 links by default).
95  */
96 struct la_group {
97         struct la_group *next, *prev;
98         struct la_entry *la_entries;
99 };
100
101 struct la_entry {
102         struct la_entry *next, *prev;
103         struct drsuapi_DsReplicaLinkedAttribute *la;
104         uint32_t dsdb_repl_flags;
105 };
106
107 struct replmd_replicated_request {
108         struct ldb_module *module;
109         struct ldb_request *req;
110
111         const struct dsdb_schema *schema;
112         struct GUID our_invocation_id;
113
114         /* the controls we pass down */
115         struct ldb_control **controls;
116
117         /*
118          * Backlinks for the replmd_add() case (we want to create
119          * backlinks after creating the user, but before the end of
120          * the ADD request) 
121          */
122         struct la_backlink *la_backlinks;
123
124         /* details for the mode where we apply a bunch of inbound replication meessages */
125         bool apply_mode;
126         uint32_t index_current;
127         struct dsdb_extended_replicated_objects *objs;
128
129         struct ldb_message *search_msg;
130         struct GUID local_parent_guid;
131
132         uint64_t seq_num;
133         bool is_urgent;
134
135         bool isDeleted;
136
137         bool fix_link_sid;
138 };
139
140 /*
141  * the result of replmd_process_linked_attribute(): either there was no change
142  * (update was ignored), a new link was added (either inactive or active), or
143  * an existing link was modified (active/inactive status may have changed).
144  */
145 typedef enum {
146         LINK_CHANGE_NONE,
147         LINK_CHANGE_ADDED,
148         LINK_CHANGE_MODIFIED,
149 } replmd_link_changed;
150
151 static int replmd_replicated_apply_merge(struct replmd_replicated_request *ar);
152 static int replmd_delete_internals(struct ldb_module *module, struct ldb_request *req, bool re_delete);
153 static int replmd_check_upgrade_links(struct ldb_context *ldb,
154                                       struct parsed_dn *dns, uint32_t count,
155                                       struct ldb_message_element *el,
156                                       const char *ldap_oid);
157 static int replmd_verify_link_target(struct replmd_replicated_request *ar,
158                                      TALLOC_CTX *mem_ctx,
159                                      struct la_entry *la_entry,
160                                      struct ldb_dn *src_dn,
161                                      const struct dsdb_attribute *attr);
162 static int replmd_get_la_entry_source(struct ldb_module *module,
163                                       struct la_entry *la_entry,
164                                       TALLOC_CTX *mem_ctx,
165                                       const struct dsdb_attribute **ret_attr,
166                                       struct ldb_message **source_msg);
167 static int replmd_set_la_val(TALLOC_CTX *mem_ctx, struct ldb_val *v, struct dsdb_dn *dsdb_dn,
168                              struct dsdb_dn *old_dsdb_dn, const struct GUID *invocation_id,
169                              uint64_t usn, uint64_t local_usn, NTTIME nttime,
170                              uint32_t version, bool deleted);
171
172 static int replmd_make_deleted_child_dn(TALLOC_CTX *tmp_ctx,
173                                         struct ldb_context *ldb,
174                                         struct ldb_dn *dn,
175                                         const char *rdn_name,
176                                         const struct ldb_val *rdn_value,
177                                         struct GUID guid);
178
179 enum urgent_situation {
180         REPL_URGENT_ON_CREATE = 1,
181         REPL_URGENT_ON_UPDATE = 2,
182         REPL_URGENT_ON_DELETE = 4
183 };
184
185 enum deletion_state {
186         OBJECT_NOT_DELETED=1,
187         OBJECT_DELETED=2,
188         OBJECT_RECYCLED=3,
189         OBJECT_TOMBSTONE=4,
190         OBJECT_REMOVED=5
191 };
192
193 static bool replmd_recyclebin_enabled(struct ldb_module *module)
194 {
195         bool enabled = false;
196         struct replmd_private *replmd_private =
197                 talloc_get_type_abort(ldb_module_get_private(module),
198                                       struct replmd_private);
199
200         /*
201          * only lookup the recycle-bin state once per replication, then cache
202          * the result. This can save us 1000s of DB searches
203          */
204         if (!replmd_private->recyclebin_state_known) {
205                 int ret = dsdb_recyclebin_enabled(module, &enabled);
206                 if (ret != LDB_SUCCESS) {
207                         return false;
208                 }
209
210                 replmd_private->recyclebin_enabled = enabled;
211                 replmd_private->recyclebin_state_known = true;
212         }
213
214         return replmd_private->recyclebin_enabled;
215 }
216
217 static void replmd_deletion_state(struct ldb_module *module,
218                                   const struct ldb_message *msg,
219                                   enum deletion_state *current_state,
220                                   enum deletion_state *next_state)
221 {
222         bool enabled = false;
223
224         if (msg == NULL) {
225                 *current_state = OBJECT_REMOVED;
226                 if (next_state != NULL) {
227                         *next_state = OBJECT_REMOVED;
228                 }
229                 return;
230         }
231
232         enabled = replmd_recyclebin_enabled(module);
233
234         if (ldb_msg_check_string_attribute(msg, "isDeleted", "TRUE")) {
235                 if (!enabled) {
236                         *current_state = OBJECT_TOMBSTONE;
237                         if (next_state != NULL) {
238                                 *next_state = OBJECT_REMOVED;
239                         }
240                         return;
241                 }
242
243                 if (ldb_msg_check_string_attribute(msg, "isRecycled", "TRUE")) {
244                         *current_state = OBJECT_RECYCLED;
245                         if (next_state != NULL) {
246                                 *next_state = OBJECT_REMOVED;
247                         }
248                         return;
249                 }
250
251                 *current_state = OBJECT_DELETED;
252                 if (next_state != NULL) {
253                         *next_state = OBJECT_RECYCLED;
254                 }
255                 return;
256         }
257
258         *current_state = OBJECT_NOT_DELETED;
259         if (next_state == NULL) {
260                 return;
261         }
262
263         if (enabled) {
264                 *next_state = OBJECT_DELETED;
265         } else {
266                 *next_state = OBJECT_TOMBSTONE;
267         }
268 }
269
270 static const struct {
271         const char *update_name;
272         enum urgent_situation repl_situation;
273 } urgent_objects[] = {
274                 {"nTDSDSA", (REPL_URGENT_ON_CREATE | REPL_URGENT_ON_DELETE)},
275                 {"crossRef", (REPL_URGENT_ON_CREATE | REPL_URGENT_ON_DELETE)},
276                 {"attributeSchema", (REPL_URGENT_ON_CREATE | REPL_URGENT_ON_UPDATE)},
277                 {"classSchema", (REPL_URGENT_ON_CREATE | REPL_URGENT_ON_UPDATE)},
278                 {"secret", (REPL_URGENT_ON_CREATE | REPL_URGENT_ON_UPDATE)},
279                 {"rIDManager", (REPL_URGENT_ON_CREATE | REPL_URGENT_ON_UPDATE)},
280                 {NULL, 0}
281 };
282
283 /* Attributes looked for when updating or deleting, to check for a urgent replication needed */
284 static const char *urgent_attrs[] = {
285                 "lockoutTime",
286                 "pwdLastSet",
287                 "userAccountControl",
288                 NULL
289 };
290
291
292 static bool replmd_check_urgent_objectclass(const struct ldb_message_element *objectclass_el,
293                                         enum urgent_situation situation)
294 {
295         unsigned int i, j;
296         for (i=0; urgent_objects[i].update_name; i++) {
297
298                 if ((situation & urgent_objects[i].repl_situation) == 0) {
299                         continue;
300                 }
301
302                 for (j=0; j<objectclass_el->num_values; j++) {
303                         const struct ldb_val *v = &objectclass_el->values[j];
304                         if (ldb_attr_cmp((const char *)v->data, urgent_objects[i].update_name) == 0) {
305                                 return true;
306                         }
307                 }
308         }
309         return false;
310 }
311
312 static bool replmd_check_urgent_attribute(const struct ldb_message_element *el)
313 {
314         if (ldb_attr_in_list(urgent_attrs, el->name)) {
315                 return true;
316         }
317         return false;
318 }
319
320 static int replmd_replicated_apply_isDeleted(struct replmd_replicated_request *ar);
321
322 /*
323   initialise the module
324   allocate the private structure and build the list
325   of partition DNs for use by replmd_notify()
326  */
327 static int replmd_init(struct ldb_module *module)
328 {
329         struct replmd_private *replmd_private;
330         struct ldb_context *ldb = ldb_module_get_ctx(module);
331         int ret;
332
333         replmd_private = talloc_zero(module, struct replmd_private);
334         if (replmd_private == NULL) {
335                 ldb_oom(ldb);
336                 return LDB_ERR_OPERATIONS_ERROR;
337         }
338
339         ret = dsdb_check_samba_compatible_feature(module,
340                                                   SAMBA_SORTED_LINKS_FEATURE,
341                                                   &replmd_private->sorted_links);
342         if (ret != LDB_SUCCESS) {
343                 talloc_free(replmd_private);
344                 return ret;
345         }
346
347         replmd_private->schema_dn = ldb_get_schema_basedn(ldb);
348         ldb_module_set_private(module, replmd_private);
349         return ldb_next_init(module);
350 }
351
352 /*
353   cleanup our per-transaction contexts
354  */
355 static void replmd_txn_cleanup(struct replmd_private *replmd_private)
356 {
357         talloc_free(replmd_private->la_ctx);
358         replmd_private->la_list = NULL;
359         replmd_private->la_ctx = NULL;
360         replmd_private->recyclebin_state_known = false;
361 }
362
363
364 struct la_backlink {
365         struct la_backlink *next, *prev;
366         const char *attr_name;
367         struct ldb_dn *forward_dn;
368         struct GUID target_guid;
369         bool active;
370 };
371
372 /*
373   a ldb_modify request operating on modules below the
374   current module
375  */
376 static int linked_attr_modify(struct ldb_module *module,
377                               const struct ldb_message *message,
378                               struct ldb_request *parent)
379 {
380         struct ldb_request *mod_req;
381         int ret;
382         struct ldb_context *ldb = ldb_module_get_ctx(module);
383         TALLOC_CTX *tmp_ctx = talloc_new(module);
384         struct ldb_result *res;
385
386         res = talloc_zero(tmp_ctx, struct ldb_result);
387         if (!res) {
388                 talloc_free(tmp_ctx);
389                 return ldb_oom(ldb_module_get_ctx(module));
390         }
391
392         ret = ldb_build_mod_req(&mod_req, ldb, tmp_ctx,
393                                 message,
394                                 NULL,
395                                 res,
396                                 ldb_modify_default_callback,
397                                 parent);
398         LDB_REQ_SET_LOCATION(mod_req);
399         if (ret != LDB_SUCCESS) {
400                 talloc_free(tmp_ctx);
401                 return ret;
402         }
403
404         ret = ldb_request_add_control(mod_req, DSDB_CONTROL_REPLICATED_UPDATE_OID,
405                                       false, NULL);
406         if (ret != LDB_SUCCESS) {
407                 return ret;
408         }
409
410         /* Run the new request */
411         ret = ldb_next_request(module, mod_req);
412
413         if (ret == LDB_SUCCESS) {
414                 ret = ldb_wait(mod_req->handle, LDB_WAIT_ALL);
415         }
416
417         talloc_free(tmp_ctx);
418         return ret;
419 }
420
421 /*
422   process a backlinks we accumulated during a transaction, adding and
423   deleting the backlinks from the target objects
424  */
425 static int replmd_process_backlink(struct ldb_module *module, struct la_backlink *bl, struct ldb_request *parent)
426 {
427         struct ldb_dn *target_dn, *source_dn;
428         int ret;
429         struct ldb_context *ldb = ldb_module_get_ctx(module);
430         struct ldb_message *msg;
431         TALLOC_CTX *frame = talloc_stackframe();
432         char *dn_string;
433
434         /*
435           - find DN of target
436           - find DN of source
437           - construct ldb_message
438               - either an add or a delete
439          */
440         ret = dsdb_module_dn_by_guid(module, frame, &bl->target_guid, &target_dn, parent);
441         if (ret != LDB_SUCCESS) {
442                 struct GUID_txt_buf guid_str;
443                 DBG_WARNING("Failed to find target DN for linked attribute with GUID %s\n",
444                             GUID_buf_string(&bl->target_guid, &guid_str));
445                 DBG_WARNING("Please run 'samba-tool dbcheck' to resolve any missing backlinks.\n");
446                 talloc_free(frame);
447                 return LDB_SUCCESS;
448         }
449
450         msg = ldb_msg_new(frame);
451         if (msg == NULL) {
452                 ldb_module_oom(module);
453                 talloc_free(frame);
454                 return LDB_ERR_OPERATIONS_ERROR;
455         }
456
457         source_dn = ldb_dn_copy(frame, bl->forward_dn);
458         if (!source_dn) {
459                 ldb_module_oom(module);
460                 talloc_free(frame);
461                 return LDB_ERR_OPERATIONS_ERROR;
462         } else {
463                 /* Filter down to the attributes we want in the backlink */
464                 const char *accept[] = { "GUID", "SID", NULL };
465                 ldb_dn_extended_filter(source_dn, accept);
466         }
467
468         /* construct a ldb_message for adding/deleting the backlink */
469         msg->dn = target_dn;
470         dn_string = ldb_dn_get_extended_linearized(frame, bl->forward_dn, 1);
471         if (!dn_string) {
472                 ldb_module_oom(module);
473                 talloc_free(frame);
474                 return LDB_ERR_OPERATIONS_ERROR;
475         }
476         ret = ldb_msg_add_steal_string(msg, bl->attr_name, dn_string);
477         if (ret != LDB_SUCCESS) {
478                 talloc_free(frame);
479                 return ret;
480         }
481         msg->elements[0].flags = bl->active?LDB_FLAG_MOD_ADD:LDB_FLAG_MOD_DELETE;
482
483         /* a backlink should never be single valued. Unfortunately the
484            exchange schema has a attribute
485            msExchBridgeheadedLocalConnectorsDNBL which is single
486            valued and a backlink. We need to cope with that by
487            ignoring the single value flag */
488         msg->elements[0].flags |= LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK;
489
490         ret = dsdb_module_modify(module, msg, DSDB_FLAG_NEXT_MODULE, parent);
491         if (ret == LDB_ERR_NO_SUCH_ATTRIBUTE && !bl->active) {
492                 /* we allow LDB_ERR_NO_SUCH_ATTRIBUTE as success to
493                    cope with possible corruption where the backlink has
494                    already been removed */
495                 DEBUG(3,("WARNING: backlink from %s already removed from %s - %s\n",
496                          ldb_dn_get_linearized(target_dn),
497                          ldb_dn_get_linearized(source_dn),
498                          ldb_errstring(ldb)));
499                 ret = LDB_SUCCESS;
500         } else if (ret != LDB_SUCCESS) {
501                 ldb_asprintf_errstring(ldb, "Failed to %s backlink from %s to %s - %s",
502                                        bl->active?"add":"remove",
503                                        ldb_dn_get_linearized(source_dn),
504                                        ldb_dn_get_linearized(target_dn),
505                                        ldb_errstring(ldb));
506                 talloc_free(frame);
507                 return ret;
508         }
509         talloc_free(frame);
510         return ret;
511 }
512
513 /*
514   add a backlink to the list of backlinks to add/delete in the prepare
515   commit
516
517   forward_dn is stolen onto the defereed context
518  */
519 static int replmd_defer_add_backlink(struct ldb_module *module,
520                                      struct replmd_private *replmd_private,
521                                      const struct dsdb_schema *schema,
522                                      struct replmd_replicated_request *ac,
523                                      struct ldb_dn *forward_dn,
524                                      struct GUID *target_guid, bool active,
525                                      const struct dsdb_attribute *schema_attr,
526                                      struct ldb_request *parent)
527 {
528         const struct dsdb_attribute *target_attr;
529         struct la_backlink *bl;
530         
531         bl = talloc(ac, struct la_backlink);
532         if (bl == NULL) {
533                 ldb_module_oom(module);
534                 return LDB_ERR_OPERATIONS_ERROR;
535         }
536
537         target_attr = dsdb_attribute_by_linkID(schema, schema_attr->linkID ^ 1);
538         if (!target_attr) {
539                 /*
540                  * windows 2003 has a broken schema where the
541                  * definition of msDS-IsDomainFor is missing (which is
542                  * supposed to be the backlink of the
543                  * msDS-HasDomainNCs attribute
544                  */
545                 return LDB_SUCCESS;
546         }
547
548         bl->attr_name = target_attr->lDAPDisplayName;
549         bl->forward_dn = talloc_steal(bl, forward_dn);
550         bl->target_guid = *target_guid;
551         bl->active = active;
552
553         DLIST_ADD(ac->la_backlinks, bl);
554
555         return LDB_SUCCESS;
556 }
557
558 /*
559   add a backlink to the list of backlinks to add/delete in the prepare
560   commit
561  */
562 static int replmd_add_backlink(struct ldb_module *module,
563                                struct replmd_private *replmd_private,
564                                const struct dsdb_schema *schema,
565                                struct ldb_dn *forward_dn,
566                                struct GUID *target_guid, bool active,
567                                const struct dsdb_attribute *schema_attr,
568                                struct ldb_request *parent)
569 {
570         const struct dsdb_attribute *target_attr;
571         struct la_backlink bl;
572         int ret;
573         
574         target_attr = dsdb_attribute_by_linkID(schema, schema_attr->linkID ^ 1);
575         if (!target_attr) {
576                 /*
577                  * windows 2003 has a broken schema where the
578                  * definition of msDS-IsDomainFor is missing (which is
579                  * supposed to be the backlink of the
580                  * msDS-HasDomainNCs attribute
581                  */
582                 return LDB_SUCCESS;
583         }
584
585         bl.attr_name = target_attr->lDAPDisplayName;
586         bl.forward_dn = forward_dn;
587         bl.target_guid = *target_guid;
588         bl.active = active;
589
590         ret = replmd_process_backlink(module, &bl, parent);
591         return ret;
592 }
593
594
595 /*
596  * Callback for most write operations in this module:
597  *
598  * notify the repl task that a object has changed. The notifies are
599  * gathered up in the replmd_private structure then written to the
600  * @REPLCHANGED object in each partition during the prepare_commit
601  */
602 static int replmd_op_callback(struct ldb_request *req, struct ldb_reply *ares)
603 {
604         int ret;
605         struct replmd_replicated_request *ac =
606                 talloc_get_type_abort(req->context, struct replmd_replicated_request);
607         struct replmd_private *replmd_private =
608                 talloc_get_type_abort(ldb_module_get_private(ac->module), struct replmd_private);
609         struct nc_entry *modified_partition;
610         struct ldb_control *partition_ctrl;
611         const struct dsdb_control_current_partition *partition;
612
613         struct ldb_control **controls;
614
615         partition_ctrl = ldb_reply_get_control(ares, DSDB_CONTROL_CURRENT_PARTITION_OID);
616
617         controls = ares->controls;
618         if (ldb_request_get_control(ac->req,
619                                     DSDB_CONTROL_CURRENT_PARTITION_OID) == NULL) {
620                 /*
621                  * Remove the current partition control from what we pass up
622                  * the chain if it hasn't been requested manually.
623                  */
624                 controls = ldb_controls_except_specified(ares->controls, ares,
625                                                          partition_ctrl);
626         }
627
628         if (ares->error != LDB_SUCCESS) {
629                 struct GUID_txt_buf guid_txt;
630                 struct ldb_message *msg = NULL;
631                 char *s = NULL;
632
633                 if (ac->apply_mode == false) {
634                         DBG_NOTICE("Originating update failure. Error is: %s\n",
635                                    ldb_strerror(ares->error));
636                         return ldb_module_done(ac->req, controls,
637                                                ares->response, ares->error);
638                 }
639
640                 msg = ac->objs->objects[ac->index_current].msg;
641                 /*
642                  * Set at DBG_NOTICE as once these start to happe, they
643                  * will happen a lot until resolved, due to repeated
644                  * replication.  The caller will probably print the
645                  * ldb error string anyway.
646                  */
647                 DBG_NOTICE("DRS replication apply failure for %s. Error is: %s\n",
648                            ldb_dn_get_linearized(msg->dn),
649                            ldb_strerror(ares->error));
650
651                 s = ldb_ldif_message_redacted_string(ldb_module_get_ctx(ac->module),
652                                                      ac,
653                                                      LDB_CHANGETYPE_ADD,
654                                                      msg);
655
656                 DBG_INFO("Failing DRS %s replication message was %s:\n%s\n",
657                          ac->search_msg == NULL ? "ADD" : "MODIFY",
658                          GUID_buf_string(&ac->objs->objects[ac->index_current].object_guid,
659                                          &guid_txt),
660                          s);
661                 talloc_free(s);
662                 return ldb_module_done(ac->req, controls,
663                                        ares->response, ares->error);
664         }
665
666         if (ares->type != LDB_REPLY_DONE) {
667                 ldb_set_errstring(ldb_module_get_ctx(ac->module), "Invalid reply type for notify\n!");
668                 return ldb_module_done(ac->req, NULL,
669                                        NULL, LDB_ERR_OPERATIONS_ERROR);
670         }
671
672         if (ac->apply_mode == false) {
673                 struct la_backlink *bl;
674                 /*
675                  * process our backlink list after an replmd_add(),
676                  * creating and deleting backlinks as necessary (this
677                  * code is sync).  The other cases are handled inline
678                  * with the modify.
679                  */
680                 for (bl=ac->la_backlinks; bl; bl=bl->next) {
681                         ret = replmd_process_backlink(ac->module, bl, ac->req);
682                         if (ret != LDB_SUCCESS) {
683                                 return ldb_module_done(ac->req, NULL,
684                                                        NULL, ret);
685                         }
686                 }
687         }
688         
689         if (!partition_ctrl) {
690                 ldb_set_errstring(ldb_module_get_ctx(ac->module),"No partition control on reply");
691                 return ldb_module_done(ac->req, NULL,
692                                        NULL, LDB_ERR_OPERATIONS_ERROR);
693         }
694
695         partition = talloc_get_type_abort(partition_ctrl->data,
696                                     struct dsdb_control_current_partition);
697
698         if (ac->seq_num > 0) {
699                 for (modified_partition = replmd_private->ncs; modified_partition;
700                      modified_partition = modified_partition->next) {
701                         if (ldb_dn_compare(modified_partition->dn, partition->dn) == 0) {
702                                 break;
703                         }
704                 }
705
706                 if (modified_partition == NULL) {
707                         modified_partition = talloc_zero(replmd_private, struct nc_entry);
708                         if (!modified_partition) {
709                                 ldb_oom(ldb_module_get_ctx(ac->module));
710                                 return ldb_module_done(ac->req, NULL,
711                                                        NULL, LDB_ERR_OPERATIONS_ERROR);
712                         }
713                         modified_partition->dn = ldb_dn_copy(modified_partition, partition->dn);
714                         if (!modified_partition->dn) {
715                                 ldb_oom(ldb_module_get_ctx(ac->module));
716                                 return ldb_module_done(ac->req, NULL,
717                                                        NULL, LDB_ERR_OPERATIONS_ERROR);
718                         }
719                         DLIST_ADD(replmd_private->ncs, modified_partition);
720                 }
721
722                 if (ac->seq_num > modified_partition->mod_usn) {
723                         modified_partition->mod_usn = ac->seq_num;
724                         if (ac->is_urgent) {
725                                 modified_partition->mod_usn_urgent = ac->seq_num;
726                         }
727                 }
728                 if (!ac->apply_mode) {
729                         replmd_private->originating_updates = true;
730                 }
731         }
732
733         if (ac->apply_mode) {
734                 ret = replmd_replicated_apply_isDeleted(ac);
735                 if (ret != LDB_SUCCESS) {
736                         return ldb_module_done(ac->req, NULL, NULL, ret);
737                 }
738                 return ret;
739         } else {
740                 /* free the partition control container here, for the
741                  * common path.  Other cases will have it cleaned up
742                  * eventually with the ares */
743                 talloc_free(partition_ctrl);
744                 return ldb_module_done(ac->req, controls,
745                                        ares->response, LDB_SUCCESS);
746         }
747 }
748
749
750 /*
751  * update a @REPLCHANGED record in each partition if there have been
752  * any writes of replicated data in the partition
753  */
754 static int replmd_notify_store(struct ldb_module *module, struct ldb_request *parent)
755 {
756         struct replmd_private *replmd_private =
757                 talloc_get_type(ldb_module_get_private(module), struct replmd_private);
758
759         while (replmd_private->ncs) {
760                 int ret;
761                 struct nc_entry *modified_partition = replmd_private->ncs;
762
763                 ret = dsdb_module_save_partition_usn(module, modified_partition->dn,
764                                                      modified_partition->mod_usn,
765                                                      modified_partition->mod_usn_urgent, parent);
766                 if (ret != LDB_SUCCESS) {
767                         DEBUG(0,(__location__ ": Failed to save partition uSN for %s\n",
768                                  ldb_dn_get_linearized(modified_partition->dn)));
769                         return ret;
770                 }
771
772                 if (ldb_dn_compare(modified_partition->dn,
773                                    replmd_private->schema_dn) == 0) {
774                         struct ldb_result *ext_res;
775                         ret = dsdb_module_extended(module,
776                                                    replmd_private->schema_dn,
777                                                    &ext_res,
778                                                    DSDB_EXTENDED_SCHEMA_UPDATE_NOW_OID,
779                                                    ext_res,
780                                                    DSDB_FLAG_NEXT_MODULE,
781                                                    parent);
782                         if (ret != LDB_SUCCESS) {
783                                 return ret;
784                         }
785                         talloc_free(ext_res);
786                 }
787
788                 DLIST_REMOVE(replmd_private->ncs, modified_partition);
789                 talloc_free(modified_partition);
790         }
791
792         return LDB_SUCCESS;
793 }
794
795
796 /*
797   created a replmd_replicated_request context
798  */
799 static struct replmd_replicated_request *replmd_ctx_init(struct ldb_module *module,
800                                                          struct ldb_request *req)
801 {
802         struct ldb_context *ldb;
803         struct replmd_replicated_request *ac;
804         const struct GUID *our_invocation_id;
805
806         ldb = ldb_module_get_ctx(module);
807
808         ac = talloc_zero(req, struct replmd_replicated_request);
809         if (ac == NULL) {
810                 ldb_oom(ldb);
811                 return NULL;
812         }
813
814         ac->module = module;
815         ac->req = req;
816
817         ac->schema = dsdb_get_schema(ldb, ac);
818         if (!ac->schema) {
819                 ldb_debug_set(ldb, LDB_DEBUG_FATAL,
820                               "replmd_modify: no dsdb_schema loaded");
821                 DEBUG(0,(__location__ ": %s\n", ldb_errstring(ldb)));
822                 talloc_free(ac);
823                 return NULL;
824         }
825
826         /* get our invocationId */
827         our_invocation_id = samdb_ntds_invocation_id(ldb);
828         if (!our_invocation_id) {
829                 ldb_debug_set(ldb, LDB_DEBUG_FATAL,
830                               "replmd_add: unable to find invocationId\n");
831                 talloc_free(ac);
832                 return NULL;
833         }
834         ac->our_invocation_id = *our_invocation_id;
835
836         return ac;
837 }
838
839 /*
840   add a time element to a record
841 */
842 static int add_time_element(struct ldb_message *msg, const char *attr, time_t t)
843 {
844         struct ldb_message_element *el;
845         char *s;
846         int ret;
847
848         if (ldb_msg_find_element(msg, attr) != NULL) {
849                 return LDB_SUCCESS;
850         }
851
852         s = ldb_timestring(msg, t);
853         if (s == NULL) {
854                 return LDB_ERR_OPERATIONS_ERROR;
855         }
856
857         ret = ldb_msg_add_string(msg, attr, s);
858         if (ret != LDB_SUCCESS) {
859                 return ret;
860         }
861
862         el = ldb_msg_find_element(msg, attr);
863         /* always set as replace. This works because on add ops, the flag
864            is ignored */
865         el->flags = LDB_FLAG_MOD_REPLACE;
866
867         return LDB_SUCCESS;
868 }
869
870 /*
871   add a uint64_t element to a record
872 */
873 static int add_uint64_element(struct ldb_context *ldb, struct ldb_message *msg,
874                               const char *attr, uint64_t v)
875 {
876         struct ldb_message_element *el;
877         int ret;
878
879         if (ldb_msg_find_element(msg, attr) != NULL) {
880                 return LDB_SUCCESS;
881         }
882
883         ret = samdb_msg_add_uint64(ldb, msg, msg, attr, v);
884         if (ret != LDB_SUCCESS) {
885                 return ret;
886         }
887
888         el = ldb_msg_find_element(msg, attr);
889         /* always set as replace. This works because on add ops, the flag
890            is ignored */
891         el->flags = LDB_FLAG_MOD_REPLACE;
892
893         return LDB_SUCCESS;
894 }
895
896 static int replmd_replPropertyMetaData1_attid_sort(const struct replPropertyMetaData1 *m1,
897                                                    const struct replPropertyMetaData1 *m2)
898 {
899         /*
900          * This assignment seems inoccous, but it is critical for the
901          * system, as we need to do the comparisons as a unsigned
902          * quantity, not signed (enums are signed integers)
903          */
904         uint32_t attid_1 = m1->attid;
905         uint32_t attid_2 = m2->attid;
906
907         if (attid_1 == attid_2) {
908                 return 0;
909         }
910
911         /*
912          * See above regarding this being an unsigned comparison.
913          * Otherwise when the high bit is set on non-standard
914          * attributes, they would end up first, before objectClass
915          * (0).
916          */
917         return attid_1 > attid_2 ? 1 : -1;
918 }
919
920 static int replmd_replPropertyMetaDataCtr1_verify(struct ldb_context *ldb,
921                                                   struct replPropertyMetaDataCtr1 *ctr1,
922                                                   struct ldb_dn *dn)
923 {
924         if (ctr1->count == 0) {
925                 ldb_debug_set(ldb, LDB_DEBUG_FATAL,
926                               "No elements found in replPropertyMetaData for %s!\n",
927                               ldb_dn_get_linearized(dn));
928                 return LDB_ERR_CONSTRAINT_VIOLATION;
929         }
930
931         /* the objectClass attribute is value 0x00000000, so must be first */
932         if (ctr1->array[0].attid != DRSUAPI_ATTID_objectClass) {
933                 ldb_debug_set(ldb, LDB_DEBUG_FATAL,
934                               "No objectClass found in replPropertyMetaData for %s!\n",
935                               ldb_dn_get_linearized(dn));
936                 return LDB_ERR_OBJECT_CLASS_VIOLATION;
937         }
938
939         return LDB_SUCCESS;
940 }
941
942 static int replmd_replPropertyMetaDataCtr1_sort_and_verify(struct ldb_context *ldb,
943                                                            struct replPropertyMetaDataCtr1 *ctr1,
944                                                            struct ldb_dn *dn)
945 {
946         /* Note this is O(n^2) for the almost-sorted case, which this is */
947         TYPESAFE_QSORT(ctr1->array, ctr1->count,
948                        replmd_replPropertyMetaData1_attid_sort);
949         return replmd_replPropertyMetaDataCtr1_verify(ldb, ctr1, dn);
950 }
951
952 static int replmd_ldb_message_element_attid_sort(const struct ldb_message_element *e1,
953                                                  const struct ldb_message_element *e2,
954                                                  const struct dsdb_schema *schema)
955 {
956         const struct dsdb_attribute *a1;
957         const struct dsdb_attribute *a2;
958
959         /*
960          * TODO: make this faster by caching the dsdb_attribute pointer
961          *       on the ldb_messag_element
962          */
963
964         a1 = dsdb_attribute_by_lDAPDisplayName(schema, e1->name);
965         a2 = dsdb_attribute_by_lDAPDisplayName(schema, e2->name);
966
967         /*
968          * TODO: remove this check, we should rely on e1 and e2 having valid attribute names
969          *       in the schema
970          */
971         if (!a1 || !a2) {
972                 return strcasecmp(e1->name, e2->name);
973         }
974         if (a1->attributeID_id == a2->attributeID_id) {
975                 return 0;
976         }
977         return a1->attributeID_id > a2->attributeID_id ? 1 : -1;
978 }
979
980 static void replmd_ldb_message_sort(struct ldb_message *msg,
981                                     const struct dsdb_schema *schema)
982 {
983         LDB_TYPESAFE_QSORT(msg->elements, msg->num_elements, schema, replmd_ldb_message_element_attid_sort);
984 }
985
986 static int replmd_build_la_val(TALLOC_CTX *mem_ctx, struct ldb_val *v, struct dsdb_dn *dsdb_dn,
987                                const struct GUID *invocation_id,
988                                uint64_t local_usn, NTTIME nttime);
989
990 static int parsed_dn_compare(struct parsed_dn *pdn1, struct parsed_dn *pdn2);
991
992 static int get_parsed_dns(struct ldb_module *module, TALLOC_CTX *mem_ctx,
993                           struct ldb_message_element *el, struct parsed_dn **pdn,
994                           const char *ldap_oid, struct ldb_request *parent);
995
996 static int check_parsed_dn_duplicates(struct ldb_module *module,
997                                       struct ldb_message_element *el,
998                                       struct parsed_dn *pdn);
999
1000 /*
1001   fix up linked attributes in replmd_add.
1002   This involves setting up the right meta-data in extended DN
1003   components, and creating backlinks to the object
1004  */
1005 static int replmd_add_fix_la(struct ldb_module *module, TALLOC_CTX *mem_ctx,
1006                              struct replmd_private *replmd_private,
1007                              struct ldb_message_element *el,
1008                              struct replmd_replicated_request *ac,
1009                              NTTIME now,
1010                              struct ldb_dn *forward_dn,
1011                              const struct dsdb_attribute *sa,
1012                              struct ldb_request *parent)
1013 {
1014         unsigned int i;
1015         TALLOC_CTX *tmp_ctx = talloc_new(mem_ctx);
1016         struct ldb_context *ldb = ldb_module_get_ctx(module);
1017         struct parsed_dn *pdn;
1018         /* We will take a reference to the schema in replmd_add_backlink */
1019         const struct dsdb_schema *schema = dsdb_get_schema(ldb, NULL);
1020         struct ldb_val *new_values = NULL;
1021         int ret;
1022
1023         if (dsdb_check_single_valued_link(sa, el) == LDB_SUCCESS) {
1024                 el->flags |= LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK;
1025         } else {
1026                 ldb_asprintf_errstring(ldb,
1027                                        "Attribute %s is single valued but "
1028                                        "more than one value has been supplied",
1029                                        el->name);
1030                 talloc_free(tmp_ctx);
1031                 return LDB_ERR_CONSTRAINT_VIOLATION;
1032         }
1033
1034         /*
1035          * At the successful end of these functions el->values is
1036          * overwritten with new_values.  However get_parsed_dns()
1037          * points p->v at the supplied el and it effectively gets used
1038          * as a working area by replmd_build_la_val().  So we must
1039          * duplicate it because our caller only called
1040          * ldb_msg_copy_shallow().
1041          */
1042
1043         el->values = talloc_memdup(tmp_ctx,
1044                                    el->values,
1045                                    sizeof(el->values[0]) * el->num_values);
1046         if (el->values == NULL) {
1047                 ldb_module_oom(module);
1048                 talloc_free(tmp_ctx);
1049                 return LDB_ERR_OPERATIONS_ERROR;
1050         }
1051         
1052         ret = get_parsed_dns(module, tmp_ctx, el, &pdn,
1053                              sa->syntax->ldap_oid, parent);
1054         if (ret != LDB_SUCCESS) {
1055                 talloc_free(tmp_ctx);
1056                 return ret;
1057         }
1058
1059         ret = check_parsed_dn_duplicates(module, el, pdn);
1060         if (ret != LDB_SUCCESS) {
1061                 talloc_free(tmp_ctx);
1062                 return ret;
1063         }
1064
1065         new_values = talloc_array(tmp_ctx, struct ldb_val, el->num_values);
1066         if (new_values == NULL) {
1067                 ldb_module_oom(module);
1068                 talloc_free(tmp_ctx);
1069                 return LDB_ERR_OPERATIONS_ERROR;
1070         }
1071
1072         for (i = 0; i < el->num_values; i++) {
1073                 struct parsed_dn *p = &pdn[i];
1074                 ret = replmd_build_la_val(new_values, p->v, p->dsdb_dn,
1075                                           &ac->our_invocation_id,
1076                                           ac->seq_num, now);
1077                 if (ret != LDB_SUCCESS) {
1078                         talloc_free(tmp_ctx);
1079                         return ret;
1080                 }
1081
1082                 ret = replmd_defer_add_backlink(module, replmd_private,
1083                                                 schema, ac,
1084                                                 forward_dn, &p->guid, true, sa,
1085                                                 parent);
1086                 if (ret != LDB_SUCCESS) {
1087                         talloc_free(tmp_ctx);
1088                         return ret;
1089                 }
1090
1091                 new_values[i] = *p->v;
1092         }
1093         el->values = talloc_steal(mem_ctx, new_values);
1094
1095         talloc_free(tmp_ctx);
1096         return LDB_SUCCESS;
1097 }
1098
1099 static int replmd_add_make_extended_dn(struct ldb_request *req,
1100                                        const DATA_BLOB *guid_blob,
1101                                        struct ldb_dn **_extended_dn)
1102 {
1103         int ret;
1104         const DATA_BLOB *sid_blob;
1105         /* Calculate an extended DN for any linked attributes */
1106         struct ldb_dn *extended_dn = ldb_dn_copy(req, req->op.add.message->dn);
1107         if (!extended_dn) {
1108                 return LDB_ERR_OPERATIONS_ERROR;
1109         }
1110         ret = ldb_dn_set_extended_component(extended_dn, "GUID", guid_blob);
1111         if (ret != LDB_SUCCESS) {
1112                 return ret;
1113         }
1114
1115         sid_blob = ldb_msg_find_ldb_val(req->op.add.message, "objectSID");
1116         if (sid_blob != NULL) {
1117                 ret = ldb_dn_set_extended_component(extended_dn, "SID", sid_blob);
1118                 if (ret != LDB_SUCCESS) {
1119                         return ret;
1120                 }
1121         }
1122         *_extended_dn = extended_dn;
1123         return LDB_SUCCESS;
1124 }
1125
1126 /*
1127   intercept add requests
1128  */
1129 static int replmd_add(struct ldb_module *module, struct ldb_request *req)
1130 {
1131         struct ldb_context *ldb;
1132         struct ldb_control *control;
1133         struct replmd_replicated_request *ac;
1134         enum ndr_err_code ndr_err;
1135         struct ldb_request *down_req;
1136         struct ldb_message *msg;
1137         const DATA_BLOB *guid_blob;
1138         DATA_BLOB guid_blob_stack;
1139         struct GUID guid;
1140         uint8_t guid_data[16];
1141         struct replPropertyMetaDataBlob nmd;
1142         struct ldb_val nmd_value;
1143         struct ldb_dn *extended_dn = NULL;
1144         
1145         /*
1146          * The use of a time_t here seems odd, but as the NTTIME
1147          * elements are actually declared as NTTIME_1sec in the IDL,
1148          * getting a higher resolution timestamp is not required.
1149          */
1150         time_t t = time(NULL);
1151         NTTIME now;
1152         char *time_str;
1153         int ret;
1154         unsigned int i;
1155         unsigned int functional_level;
1156         uint32_t ni=0;
1157         bool allow_add_guid = false;
1158         bool remove_current_guid = false;
1159         bool is_urgent = false;
1160         bool is_schema_nc = false;
1161         struct ldb_message_element *objectclass_el;
1162         struct replmd_private *replmd_private =
1163                 talloc_get_type_abort(ldb_module_get_private(module), struct replmd_private);
1164
1165         /* check if there's a show relax control (used by provision to say 'I know what I'm doing') */
1166         control = ldb_request_get_control(req, LDB_CONTROL_RELAX_OID);
1167         if (control) {
1168                 allow_add_guid = true;
1169         }
1170
1171         /* do not manipulate our control entries */
1172         if (ldb_dn_is_special(req->op.add.message->dn)) {
1173                 return ldb_next_request(module, req);
1174         }
1175
1176         ldb = ldb_module_get_ctx(module);
1177
1178         ldb_debug(ldb, LDB_DEBUG_TRACE, "replmd_add\n");
1179
1180         guid_blob = ldb_msg_find_ldb_val(req->op.add.message, "objectGUID");
1181         if (guid_blob != NULL) {
1182                 if (!allow_add_guid) {
1183                         ldb_set_errstring(ldb,
1184                                           "replmd_add: it's not allowed to add an object with objectGUID!");
1185                         return LDB_ERR_UNWILLING_TO_PERFORM;
1186                 } else {
1187                         NTSTATUS status = GUID_from_data_blob(guid_blob,&guid);
1188                         if (!NT_STATUS_IS_OK(status)) {
1189                                 ldb_set_errstring(ldb,
1190                                                   "replmd_add: Unable to parse the 'objectGUID' as a GUID!");
1191                                 return LDB_ERR_UNWILLING_TO_PERFORM;
1192                         }
1193                         /* we remove this attribute as it can be a string and
1194                          * will not be treated correctly and then we will re-add
1195                          * it later on in the good format */
1196                         remove_current_guid = true;
1197                 }
1198         } else {
1199                 /* a new GUID */
1200                 guid = GUID_random();
1201                 
1202                 guid_blob_stack = data_blob_const(guid_data, sizeof(guid_data));
1203                 
1204                 /* This can't fail */
1205                 ndr_push_struct_into_fixed_blob(&guid_blob_stack, &guid,
1206                                                 (ndr_push_flags_fn_t)ndr_push_GUID);
1207                 guid_blob = &guid_blob_stack;
1208         }
1209
1210         ac = replmd_ctx_init(module, req);
1211         if (ac == NULL) {
1212                 return ldb_module_oom(module);
1213         }
1214
1215         functional_level = dsdb_functional_level(ldb);
1216
1217         /* Get a sequence number from the backend */
1218         ret = ldb_sequence_number(ldb, LDB_SEQ_NEXT, &ac->seq_num);
1219         if (ret != LDB_SUCCESS) {
1220                 talloc_free(ac);
1221                 return ret;
1222         }
1223
1224         /* we have to copy the message as the caller might have it as a const */
1225         msg = ldb_msg_copy_shallow(ac, req->op.add.message);
1226         if (msg == NULL) {
1227                 ldb_oom(ldb);
1228                 talloc_free(ac);
1229                 return LDB_ERR_OPERATIONS_ERROR;
1230         }
1231
1232         /* generated times */
1233         unix_to_nt_time(&now, t);
1234         time_str = ldb_timestring(msg, t);
1235         if (!time_str) {
1236                 ldb_oom(ldb);
1237                 talloc_free(ac);
1238                 return LDB_ERR_OPERATIONS_ERROR;
1239         }
1240         if (remove_current_guid) {
1241                 ldb_msg_remove_attr(msg,"objectGUID");
1242         }
1243
1244         /*
1245          * remove autogenerated attributes
1246          */
1247         ldb_msg_remove_attr(msg, "whenCreated");
1248         ldb_msg_remove_attr(msg, "whenChanged");
1249         ldb_msg_remove_attr(msg, "uSNCreated");
1250         ldb_msg_remove_attr(msg, "uSNChanged");
1251         ldb_msg_remove_attr(msg, "replPropertyMetaData");
1252
1253         /*
1254          * readd replicated attributes
1255          */
1256         ret = ldb_msg_add_string(msg, "whenCreated", time_str);
1257         if (ret != LDB_SUCCESS) {
1258                 ldb_oom(ldb);
1259                 talloc_free(ac);
1260                 return ret;
1261         }
1262
1263         /* build the replication meta_data */
1264         ZERO_STRUCT(nmd);
1265         nmd.version             = 1;
1266         nmd.ctr.ctr1.count      = msg->num_elements;
1267         nmd.ctr.ctr1.array      = talloc_array(msg,
1268                                                struct replPropertyMetaData1,
1269                                                nmd.ctr.ctr1.count);
1270         if (!nmd.ctr.ctr1.array) {
1271                 ldb_oom(ldb);
1272                 talloc_free(ac);
1273                 return LDB_ERR_OPERATIONS_ERROR;
1274         }
1275
1276         is_schema_nc = ldb_dn_compare_base(replmd_private->schema_dn, msg->dn) == 0;
1277
1278         for (i=0; i < msg->num_elements;) {
1279                 struct ldb_message_element *e = &msg->elements[i];
1280                 struct replPropertyMetaData1 *m = &nmd.ctr.ctr1.array[ni];
1281                 const struct dsdb_attribute *sa;
1282
1283                 if (e->name[0] == '@') {
1284                         i++;
1285                         continue;
1286                 }
1287
1288                 sa = dsdb_attribute_by_lDAPDisplayName(ac->schema, e->name);
1289                 if (!sa) {
1290                         ldb_debug_set(ldb, LDB_DEBUG_ERROR,
1291                                       "replmd_add: attribute '%s' not defined in schema\n",
1292                                       e->name);
1293                         talloc_free(ac);
1294                         return LDB_ERR_NO_SUCH_ATTRIBUTE;
1295                 }
1296
1297                 if ((sa->systemFlags & DS_FLAG_ATTR_NOT_REPLICATED) || (sa->systemFlags & DS_FLAG_ATTR_IS_CONSTRUCTED)) {
1298                         /* if the attribute is not replicated (0x00000001)
1299                          * or constructed (0x00000004) it has no metadata
1300                          */
1301                         i++;
1302                         continue;
1303                 }
1304
1305                 if (sa->linkID != 0 && functional_level > DS_DOMAIN_FUNCTION_2000) {
1306                         if (extended_dn == NULL) {
1307                                 ret = replmd_add_make_extended_dn(req,
1308                                                                   guid_blob,
1309                                                                   &extended_dn);
1310                                 if (ret != LDB_SUCCESS) {
1311                                         talloc_free(ac);
1312                                         return ret;
1313                                 }
1314                         }                       
1315
1316                         /*
1317                          * Prepare the context for the backlinks and
1318                          * create metadata for the forward links.  The
1319                          * backlinks are created in
1320                          * replmd_op_callback() after the successful
1321                          * ADD of the object.
1322                          */
1323                         ret = replmd_add_fix_la(module, msg->elements,
1324                                                 replmd_private, e,
1325                                                 ac, now,
1326                                                 extended_dn,
1327                                                 sa, req);
1328                         if (ret != LDB_SUCCESS) {
1329                                 talloc_free(ac);
1330                                 return ret;
1331                         }
1332                         /* linked attributes are not stored in
1333                            replPropertyMetaData in FL above w2k */
1334                         i++;
1335                         continue;
1336                 }
1337
1338                 m->attid   = dsdb_attribute_get_attid(sa, is_schema_nc);
1339                 m->version = 1;
1340                 if (m->attid == DRSUAPI_ATTID_isDeleted) {
1341                         const struct ldb_val *rdn_val = ldb_dn_get_rdn_val(msg->dn);
1342                         const char* rdn;
1343
1344                         if (rdn_val == NULL) {
1345                                 ldb_oom(ldb);
1346                                 talloc_free(ac);
1347                                 return LDB_ERR_OPERATIONS_ERROR;
1348                         }
1349
1350                         rdn = (const char*)rdn_val->data;
1351                         if (strcmp(rdn, "Deleted Objects") == 0) {
1352                                 /*
1353                                  * Set the originating_change_time to 29/12/9999 at 23:59:59
1354                                  * as specified in MS-ADTS 7.1.1.4.2 Deleted Objects Container
1355                                  */
1356                                 m->originating_change_time      = DELETED_OBJECT_CONTAINER_CHANGE_TIME;
1357                         } else {
1358                                 m->originating_change_time      = now;
1359                         }
1360                 } else {
1361                         m->originating_change_time      = now;
1362                 }
1363                 m->originating_invocation_id    = ac->our_invocation_id;
1364                 m->originating_usn              = ac->seq_num;
1365                 m->local_usn                    = ac->seq_num;
1366                 ni++;
1367
1368                 if (!(e->flags & DSDB_FLAG_INTERNAL_FORCE_META_DATA)) {
1369                         i++;
1370                         continue;
1371                 }
1372
1373                 e->flags &= ~DSDB_FLAG_INTERNAL_FORCE_META_DATA;
1374
1375                 if (e->num_values != 0) {
1376                         i++;
1377                         continue;
1378                 }
1379
1380                 ldb_msg_remove_element(msg, e);
1381         }
1382
1383         /* fix meta data count */
1384         nmd.ctr.ctr1.count = ni;
1385
1386         /*
1387          * sort meta data array
1388          */
1389         ret = replmd_replPropertyMetaDataCtr1_sort_and_verify(ldb, &nmd.ctr.ctr1, msg->dn);
1390         if (ret != LDB_SUCCESS) {
1391                 ldb_asprintf_errstring(ldb, "%s: error during direct ADD: %s", __func__, ldb_errstring(ldb));
1392                 talloc_free(ac);
1393                 return ret;
1394         }
1395
1396         /* generated NDR encoded values */
1397         ndr_err = ndr_push_struct_blob(&nmd_value, msg,
1398                                        &nmd,
1399                                        (ndr_push_flags_fn_t)ndr_push_replPropertyMetaDataBlob);
1400         if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
1401                 ldb_oom(ldb);
1402                 talloc_free(ac);
1403                 return LDB_ERR_OPERATIONS_ERROR;
1404         }
1405
1406         /*
1407          * add the autogenerated values
1408          */
1409         ret = dsdb_msg_add_guid(msg, &guid, "objectGUID");
1410         if (ret != LDB_SUCCESS) {
1411                 ldb_oom(ldb);
1412                 talloc_free(ac);
1413                 return ret;
1414         }
1415         ret = ldb_msg_add_string(msg, "whenChanged", time_str);
1416         if (ret != LDB_SUCCESS) {
1417                 ldb_oom(ldb);
1418                 talloc_free(ac);
1419                 return ret;
1420         }
1421         ret = samdb_msg_add_uint64(ldb, msg, msg, "uSNCreated", ac->seq_num);
1422         if (ret != LDB_SUCCESS) {
1423                 ldb_oom(ldb);
1424                 talloc_free(ac);
1425                 return ret;
1426         }
1427         ret = samdb_msg_add_uint64(ldb, msg, msg, "uSNChanged", ac->seq_num);
1428         if (ret != LDB_SUCCESS) {
1429                 ldb_oom(ldb);
1430                 talloc_free(ac);
1431                 return ret;
1432         }
1433         ret = ldb_msg_add_value(msg, "replPropertyMetaData", &nmd_value, NULL);
1434         if (ret != LDB_SUCCESS) {
1435                 ldb_oom(ldb);
1436                 talloc_free(ac);
1437                 return ret;
1438         }
1439
1440         /*
1441          * sort the attributes by attid before storing the object
1442          */
1443         replmd_ldb_message_sort(msg, ac->schema);
1444
1445         /*
1446          * Assert that we do have an objectClass
1447          */
1448         objectclass_el = ldb_msg_find_element(msg, "objectClass");
1449         if (objectclass_el == NULL) {
1450                 ldb_asprintf_errstring(ldb, __location__
1451                                        ": objectClass missing on %s\n",
1452                                        ldb_dn_get_linearized(msg->dn));
1453                 talloc_free(ac);
1454                 return LDB_ERR_OBJECT_CLASS_VIOLATION;
1455         }
1456         is_urgent = replmd_check_urgent_objectclass(objectclass_el,
1457                                                         REPL_URGENT_ON_CREATE);
1458
1459         ac->is_urgent = is_urgent;
1460         ret = ldb_build_add_req(&down_req, ldb, ac,
1461                                 msg,
1462                                 req->controls,
1463                                 ac, replmd_op_callback,
1464                                 req);
1465
1466         LDB_REQ_SET_LOCATION(down_req);
1467         if (ret != LDB_SUCCESS) {
1468                 talloc_free(ac);
1469                 return ret;
1470         }
1471
1472         /* current partition control is needed by "replmd_op_callback" */
1473         if (ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID) == NULL) {
1474                 ret = ldb_request_add_control(down_req,
1475                                               DSDB_CONTROL_CURRENT_PARTITION_OID,
1476                                               false, NULL);
1477                 if (ret != LDB_SUCCESS) {
1478                         talloc_free(ac);
1479                         return ret;
1480                 }
1481         }
1482
1483         if (functional_level == DS_DOMAIN_FUNCTION_2000) {
1484                 ret = ldb_request_add_control(down_req, DSDB_CONTROL_APPLY_LINKS, false, NULL);
1485                 if (ret != LDB_SUCCESS) {
1486                         talloc_free(ac);
1487                         return ret;
1488                 }
1489         }
1490
1491         /* mark the relax control done */
1492         if (control) {
1493                 control->critical = 0;
1494         }
1495         /* go on with the call chain */
1496         return ldb_next_request(module, down_req);
1497 }
1498
1499
1500 /*
1501  * update the replPropertyMetaData for one element
1502  */
1503 static int replmd_update_rpmd_element(struct ldb_context *ldb,
1504                                       struct ldb_message *msg,
1505                                       struct ldb_message_element *el,
1506                                       struct ldb_message_element *old_el,
1507                                       struct replPropertyMetaDataBlob *omd,
1508                                       const struct dsdb_schema *schema,
1509                                       uint64_t *seq_num,
1510                                       const struct GUID *our_invocation_id,
1511                                       NTTIME now,
1512                                       bool is_schema_nc,
1513                                       bool is_forced_rodc,
1514                                       struct ldb_request *req)
1515 {
1516         uint32_t i;
1517         const struct dsdb_attribute *a;
1518         struct replPropertyMetaData1 *md1;
1519         bool may_skip = false;
1520         uint32_t attid;
1521
1522         a = dsdb_attribute_by_lDAPDisplayName(schema, el->name);
1523         if (a == NULL) {
1524                 if (ldb_request_get_control(req, LDB_CONTROL_RELAX_OID)) {
1525                         /* allow this to make it possible for dbcheck
1526                            to remove bad attributes */
1527                         return LDB_SUCCESS;
1528                 }
1529
1530                 DEBUG(0,(__location__ ": Unable to find attribute %s in schema\n",
1531                          el->name));
1532                 return LDB_ERR_OPERATIONS_ERROR;
1533         }
1534
1535         attid = dsdb_attribute_get_attid(a, is_schema_nc);
1536
1537         if ((a->systemFlags & DS_FLAG_ATTR_NOT_REPLICATED) || (a->systemFlags & DS_FLAG_ATTR_IS_CONSTRUCTED)) {
1538                 return LDB_SUCCESS;
1539         }
1540
1541         /*
1542          * if the attribute's value haven't changed, and this isn't
1543          * just a delete of everything then return LDB_SUCCESS Unless
1544          * we have the provision control or if the attribute is
1545          * interSiteTopologyGenerator as this page explain:
1546          * http://support.microsoft.com/kb/224815 this attribute is
1547          * periodicaly written by the DC responsible for the intersite
1548          * generation in a given site
1549          *
1550          * Unchanged could be deleting or replacing an already-gone
1551          * thing with an unconstrained delete/empty replace or a
1552          * replace with the same value, but not an add with the same
1553          * value because that could be about adding a duplicate (which
1554          * is for someone else to error out on).
1555          */
1556         if (old_el != NULL && ldb_msg_element_equal_ordered(el, old_el)) {
1557                 if (LDB_FLAG_MOD_TYPE(el->flags) == LDB_FLAG_MOD_REPLACE) {
1558                         may_skip = true;
1559                 }
1560         } else if (old_el == NULL && el->num_values == 0) {
1561                 if (LDB_FLAG_MOD_TYPE(el->flags) == LDB_FLAG_MOD_REPLACE) {
1562                         may_skip = true;
1563                 } else if (LDB_FLAG_MOD_TYPE(el->flags) == LDB_FLAG_MOD_DELETE) {
1564                         may_skip = true;
1565                 }
1566         } else if (a->linkID != 0 && LDB_FLAG_MOD_TYPE(el->flags) == LDB_FLAG_MOD_DELETE &&
1567                    ldb_request_get_control(req, DSDB_CONTROL_REPLMD_VANISH_LINKS) != NULL) {
1568                 /*
1569                  * We intentionally skip the version bump when attempting to
1570                  * vanish links.
1571                  *
1572                  * The control is set by dbcheck and expunge-tombstones which
1573                  * both attempt to be non-replicating. Otherwise, making an
1574                  * alteration to the replication state would trigger a
1575                  * broadcast of all expunged objects.
1576                  */
1577                 may_skip = true;
1578         }
1579
1580         if (el->flags & DSDB_FLAG_INTERNAL_FORCE_META_DATA) {
1581                 may_skip = false;
1582                 el->flags &= ~DSDB_FLAG_INTERNAL_FORCE_META_DATA;
1583         }
1584
1585         if (may_skip) {
1586                 if (strcmp(el->name, "interSiteTopologyGenerator") != 0 &&
1587                     !ldb_request_get_control(req, LDB_CONTROL_PROVISION_OID)) {
1588                         /*
1589                          * allow this to make it possible for dbcheck
1590                          * to rebuild broken metadata
1591                          */
1592                         return LDB_SUCCESS;
1593                 }
1594         }
1595
1596         for (i=0; i<omd->ctr.ctr1.count; i++) {
1597                 /*
1598                  * First check if we find it under the msDS-IntID,
1599                  * then check if we find it under the OID and
1600                  * prefixMap ID.
1601                  *
1602                  * This allows the administrator to simply re-write
1603                  * the attributes and so restore replication, which is
1604                  * likely what they will try to do.
1605                  */
1606                 if (attid == omd->ctr.ctr1.array[i].attid) {
1607                         break;
1608                 }
1609
1610                 if (a->attributeID_id == omd->ctr.ctr1.array[i].attid) {
1611                         break;
1612                 }
1613         }
1614
1615         if (a->linkID != 0 && dsdb_functional_level(ldb) > DS_DOMAIN_FUNCTION_2000) {
1616                 /* linked attributes are not stored in
1617                    replPropertyMetaData in FL above w2k, but we do
1618                    raise the seqnum for the object  */
1619                 if (*seq_num == 0 &&
1620                     ldb_sequence_number(ldb, LDB_SEQ_NEXT, seq_num) != LDB_SUCCESS) {
1621                         return LDB_ERR_OPERATIONS_ERROR;
1622                 }
1623                 return LDB_SUCCESS;
1624         }
1625
1626         if (i == omd->ctr.ctr1.count) {
1627                 /* we need to add a new one */
1628                 omd->ctr.ctr1.array = talloc_realloc(msg, omd->ctr.ctr1.array,
1629                                                      struct replPropertyMetaData1, omd->ctr.ctr1.count+1);
1630                 if (omd->ctr.ctr1.array == NULL) {
1631                         ldb_oom(ldb);
1632                         return LDB_ERR_OPERATIONS_ERROR;
1633                 }
1634                 omd->ctr.ctr1.count++;
1635                 ZERO_STRUCT(omd->ctr.ctr1.array[i]);
1636         }
1637
1638         /* Get a new sequence number from the backend. We only do this
1639          * if we have a change that requires a new
1640          * replPropertyMetaData element
1641          */
1642         if (*seq_num == 0) {
1643                 int ret = ldb_sequence_number(ldb, LDB_SEQ_NEXT, seq_num);
1644                 if (ret != LDB_SUCCESS) {
1645                         return LDB_ERR_OPERATIONS_ERROR;
1646                 }
1647         }
1648
1649         md1 = &omd->ctr.ctr1.array[i];
1650         md1->version++;
1651         md1->attid = attid;
1652
1653         if (md1->attid == DRSUAPI_ATTID_isDeleted) {
1654                 const struct ldb_val *rdn_val = ldb_dn_get_rdn_val(msg->dn);
1655                 const char* rdn;
1656
1657                 if (rdn_val == NULL) {
1658                         ldb_oom(ldb);
1659                         return LDB_ERR_OPERATIONS_ERROR;
1660                 }
1661
1662                 rdn = (const char*)rdn_val->data;
1663                 if (strcmp(rdn, "Deleted Objects") == 0) {
1664                         /*
1665                          * Set the originating_change_time to 29/12/9999 at 23:59:59
1666                          * as specified in MS-ADTS 7.1.1.4.2 Deleted Objects Container
1667                          */
1668                         md1->originating_change_time    = DELETED_OBJECT_CONTAINER_CHANGE_TIME;
1669                 } else {
1670                         md1->originating_change_time    = now;
1671                 }
1672         } else {
1673                 md1->originating_change_time    = now;
1674         }
1675         md1->originating_invocation_id = *our_invocation_id;
1676         md1->originating_usn           = *seq_num;
1677         md1->local_usn                 = *seq_num;
1678
1679         if (is_forced_rodc) {
1680                 /* Force version to 0 to be overridden later via replication */
1681                 md1->version = 0;
1682         }
1683
1684         return LDB_SUCCESS;
1685 }
1686
1687 /*
1688  * Bump the replPropertyMetaData version on an attribute, and if it
1689  * has changed (or forced by leaving rdn_old NULL), update the value
1690  * in the entry.
1691  *
1692  * This is important, as calling a modify operation may not change the
1693  * version number if the values appear unchanged, but a rename between
1694  * parents bumps this value.
1695  *
1696  */
1697 static int replmd_update_rpmd_rdn_attr(struct ldb_context *ldb,
1698                                        struct ldb_message *msg,
1699                                        const struct ldb_val *rdn_new,
1700                                        const struct ldb_val *rdn_old,
1701                                        struct replPropertyMetaDataBlob *omd,
1702                                        struct replmd_replicated_request *ar,
1703                                        NTTIME now,
1704                                        bool is_schema_nc,
1705                                        bool is_forced_rodc)
1706 {
1707         const char *rdn_name = ldb_dn_get_rdn_name(msg->dn);
1708         const struct dsdb_attribute *rdn_attr =
1709                 dsdb_attribute_by_lDAPDisplayName(ar->schema, rdn_name);
1710         const char *attr_name = rdn_attr != NULL ?
1711                                 rdn_attr->lDAPDisplayName :
1712                                 rdn_name;
1713         struct ldb_message_element new_el = {
1714                 .flags = LDB_FLAG_MOD_REPLACE,
1715                 .name = attr_name,
1716                 .num_values = 1,
1717                 .values = discard_const_p(struct ldb_val, rdn_new)
1718         };
1719         struct ldb_message_element old_el = {
1720                 .flags = LDB_FLAG_MOD_REPLACE,
1721                 .name = attr_name,
1722                 .num_values = rdn_old ? 1 : 0,
1723                 .values = discard_const_p(struct ldb_val, rdn_old)
1724         };
1725
1726         if (ldb_msg_element_equal_ordered(&new_el, &old_el) == false) {
1727                 int ret = ldb_msg_add(msg, &new_el, LDB_FLAG_MOD_REPLACE);
1728                 if (ret != LDB_SUCCESS) {
1729                         return ldb_oom(ldb);
1730                 }
1731         }
1732
1733         return replmd_update_rpmd_element(ldb, msg, &new_el, NULL,
1734                                           omd, ar->schema, &ar->seq_num,
1735                                           &ar->our_invocation_id,
1736                                           now, is_schema_nc, is_forced_rodc,
1737                                           ar->req);
1738
1739 }
1740
1741 static uint64_t find_max_local_usn(struct replPropertyMetaDataBlob omd)
1742 {
1743         uint32_t count = omd.ctr.ctr1.count;
1744         uint64_t max = 0;
1745         uint32_t i;
1746         for (i=0; i < count; i++) {
1747                 struct replPropertyMetaData1 m = omd.ctr.ctr1.array[i];
1748                 if (max < m.local_usn) {
1749                         max = m.local_usn;
1750                 }
1751         }
1752         return max;
1753 }
1754
1755 /*
1756  * update the replPropertyMetaData object each time we modify an
1757  * object. This is needed for DRS replication, as the merge on the
1758  * client is based on this object
1759  */
1760 static int replmd_update_rpmd(struct ldb_module *module,
1761                               const struct dsdb_schema *schema,
1762                               struct ldb_request *req,
1763                               const char * const *rename_attrs,
1764                               struct ldb_message *msg, uint64_t *seq_num,
1765                               time_t t, bool is_schema_nc,
1766                               bool *is_urgent, bool *rodc)
1767 {
1768         const struct ldb_val *omd_value;
1769         enum ndr_err_code ndr_err;
1770         struct replPropertyMetaDataBlob omd;
1771         unsigned int i;
1772         NTTIME now;
1773         const struct GUID *our_invocation_id;
1774         int ret;
1775         const char * const *attrs = NULL;
1776         const char * const attrs2[] = { "uSNChanged", "objectClass", "instanceType", NULL };
1777         struct ldb_result *res;
1778         struct ldb_context *ldb;
1779         struct ldb_message_element *objectclass_el;
1780         enum urgent_situation situation;
1781         bool rmd_is_provided;
1782         bool rmd_is_just_resorted = false;
1783         const char *not_rename_attrs[4 + msg->num_elements];
1784         bool is_forced_rodc = false;
1785
1786         if (rename_attrs) {
1787                 attrs = rename_attrs;
1788         } else {
1789                 for (i = 0; i < msg->num_elements; i++) {
1790                         not_rename_attrs[i] = msg->elements[i].name;
1791                 }
1792                 not_rename_attrs[i] = "replPropertyMetaData";
1793                 not_rename_attrs[i+1] = "objectClass";
1794                 not_rename_attrs[i+2] = "instanceType";
1795                 not_rename_attrs[i+3] = NULL;
1796                 attrs = not_rename_attrs;
1797         }
1798
1799         ldb = ldb_module_get_ctx(module);
1800
1801         ret = samdb_rodc(ldb, rodc);
1802         if (ret != LDB_SUCCESS) {
1803                 DEBUG(4, (__location__ ": unable to tell if we are an RODC\n"));
1804                 *rodc = false;
1805         }
1806
1807         if (*rodc &&
1808             ldb_request_get_control(req, DSDB_CONTROL_FORCE_RODC_LOCAL_CHANGE)) {
1809                 is_forced_rodc = true;
1810         }
1811
1812         our_invocation_id = samdb_ntds_invocation_id(ldb);
1813         if (!our_invocation_id) {
1814                 /* this happens during an initial vampire while
1815                    updating the schema */
1816                 DEBUG(5,("No invocationID - skipping replPropertyMetaData update\n"));
1817                 return LDB_SUCCESS;
1818         }
1819
1820         unix_to_nt_time(&now, t);
1821
1822         if (ldb_request_get_control(req, DSDB_CONTROL_CHANGEREPLMETADATA_OID)) {
1823                 rmd_is_provided = true;
1824                 if (ldb_request_get_control(req, DSDB_CONTROL_CHANGEREPLMETADATA_RESORT_OID)) {
1825                         rmd_is_just_resorted = true;
1826                 }
1827         } else {
1828                 rmd_is_provided = false;
1829         }
1830
1831         /* if isDeleted is present and is TRUE, then we consider we are deleting,
1832          * otherwise we consider we are updating */
1833         if (ldb_msg_check_string_attribute(msg, "isDeleted", "TRUE")) {
1834                 situation = REPL_URGENT_ON_DELETE;
1835         } else if (rename_attrs) {
1836                 situation = REPL_URGENT_ON_CREATE | REPL_URGENT_ON_DELETE;
1837         } else {
1838                 situation = REPL_URGENT_ON_UPDATE;
1839         }
1840
1841         if (rmd_is_provided) {
1842                 /* In this case the change_replmetadata control was supplied */
1843                 /* We check that it's the only attribute that is provided
1844                  * (it's a rare case so it's better to keep the code simplier)
1845                  * We also check that the highest local_usn is bigger or the same as
1846                  * uSNChanged. */
1847                 uint64_t db_seq;
1848                 if( msg->num_elements != 1 ||
1849                         strncmp(msg->elements[0].name,
1850                                 "replPropertyMetaData", 20) ) {
1851                         DEBUG(0,(__location__ ": changereplmetada control called without "\
1852                                 "a specified replPropertyMetaData attribute or with others\n"));
1853                         return LDB_ERR_OPERATIONS_ERROR;
1854                 }
1855                 if (situation != REPL_URGENT_ON_UPDATE) {
1856                         DEBUG(0,(__location__ ": changereplmetada control can't be called when deleting an object\n"));
1857                         return LDB_ERR_OPERATIONS_ERROR;
1858                 }
1859                 omd_value = ldb_msg_find_ldb_val(msg, "replPropertyMetaData");
1860                 if (!omd_value) {
1861                         DEBUG(0,(__location__ ": replPropertyMetaData was not specified for Object %s\n",
1862                                  ldb_dn_get_linearized(msg->dn)));
1863                         return LDB_ERR_OPERATIONS_ERROR;
1864                 }
1865                 ndr_err = ndr_pull_struct_blob(omd_value, msg, &omd,
1866                                                (ndr_pull_flags_fn_t)ndr_pull_replPropertyMetaDataBlob);
1867                 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
1868                         DEBUG(0,(__location__ ": Failed to parse replPropertyMetaData for %s\n",
1869                                  ldb_dn_get_linearized(msg->dn)));
1870                         return LDB_ERR_OPERATIONS_ERROR;
1871                 }
1872
1873                 ret = dsdb_module_search_dn(module, msg, &res, msg->dn, attrs2,
1874                                             DSDB_FLAG_NEXT_MODULE |
1875                                             DSDB_SEARCH_SHOW_RECYCLED |
1876                                             DSDB_SEARCH_SHOW_EXTENDED_DN |
1877                                             DSDB_SEARCH_SHOW_DN_IN_STORAGE_FORMAT |
1878                                             DSDB_SEARCH_REVEAL_INTERNALS, req);
1879
1880                 if (ret != LDB_SUCCESS) {
1881                         return ret;
1882                 }
1883
1884                 if (rmd_is_just_resorted == false) {
1885                         *seq_num = find_max_local_usn(omd);
1886
1887                         db_seq = ldb_msg_find_attr_as_uint64(res->msgs[0], "uSNChanged", 0);
1888
1889                         /*
1890                          * The test here now allows for a new
1891                          * replPropertyMetaData with no change, if was
1892                          * just dbcheck re-sorting the values.
1893                          */
1894                         if (*seq_num <= db_seq) {
1895                                 DEBUG(0,(__location__ ": changereplmetada control provided but max(local_usn)" \
1896                                          " is less than uSNChanged (max = %lld uSNChanged = %lld)\n",
1897                                          (long long)*seq_num, (long long)db_seq));
1898                                 return LDB_ERR_OPERATIONS_ERROR;
1899                         }
1900                 }
1901
1902         } else {
1903                 /* search for the existing replPropertyMetaDataBlob. We need
1904                  * to use REVEAL and ask for DNs in storage format to support
1905                  * the check for values being the same in
1906                  * replmd_update_rpmd_element()
1907                  */
1908                 ret = dsdb_module_search_dn(module, msg, &res, msg->dn, attrs,
1909                                             DSDB_FLAG_NEXT_MODULE |
1910                                             DSDB_SEARCH_SHOW_RECYCLED |
1911                                             DSDB_SEARCH_SHOW_EXTENDED_DN |
1912                                             DSDB_SEARCH_SHOW_DN_IN_STORAGE_FORMAT |
1913                                             DSDB_SEARCH_REVEAL_INTERNALS, req);
1914                 if (ret != LDB_SUCCESS) {
1915                         return ret;
1916                 }
1917
1918                 omd_value = ldb_msg_find_ldb_val(res->msgs[0], "replPropertyMetaData");
1919                 if (!omd_value) {
1920                         DEBUG(0,(__location__ ": Object %s does not have a replPropertyMetaData attribute\n",
1921                                  ldb_dn_get_linearized(msg->dn)));
1922                         return LDB_ERR_OPERATIONS_ERROR;
1923                 }
1924
1925                 ndr_err = ndr_pull_struct_blob(omd_value, msg, &omd,
1926                                                (ndr_pull_flags_fn_t)ndr_pull_replPropertyMetaDataBlob);
1927                 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
1928                         DEBUG(0,(__location__ ": Failed to parse replPropertyMetaData for %s\n",
1929                                  ldb_dn_get_linearized(msg->dn)));
1930                         return LDB_ERR_OPERATIONS_ERROR;
1931                 }
1932
1933                 if (omd.version != 1) {
1934                         DEBUG(0,(__location__ ": bad version %u in replPropertyMetaData for %s\n",
1935                                  omd.version, ldb_dn_get_linearized(msg->dn)));
1936                         return LDB_ERR_OPERATIONS_ERROR;
1937                 }
1938
1939                 for (i=0; i<msg->num_elements;) {
1940                         struct ldb_message_element *el = &msg->elements[i];
1941                         struct ldb_message_element *old_el;
1942
1943                         old_el = ldb_msg_find_element(res->msgs[0], el->name);
1944                         ret = replmd_update_rpmd_element(ldb, msg, el, old_el,
1945                                                          &omd, schema, seq_num,
1946                                                          our_invocation_id,
1947                                                          now, is_schema_nc,
1948                                                          is_forced_rodc,
1949                                                          req);
1950                         if (ret != LDB_SUCCESS) {
1951                                 return ret;
1952                         }
1953
1954                         if (!*is_urgent && (situation == REPL_URGENT_ON_UPDATE)) {
1955                                 *is_urgent = replmd_check_urgent_attribute(el);
1956                         }
1957
1958                         if (!(el->flags & DSDB_FLAG_INTERNAL_FORCE_META_DATA)) {
1959                                 i++;
1960                                 continue;
1961                         }
1962
1963                         el->flags &= ~DSDB_FLAG_INTERNAL_FORCE_META_DATA;
1964
1965                         if (el->num_values != 0) {
1966                                 i++;
1967                                 continue;
1968                         }
1969
1970                         ldb_msg_remove_element(msg, el);
1971                 }
1972         }
1973
1974         /*
1975          * Assert that we have an objectClass attribute - this is major
1976          * corruption if we don't have this!
1977          */
1978         objectclass_el = ldb_msg_find_element(res->msgs[0], "objectClass");
1979         if (objectclass_el != NULL) {
1980                 /*
1981                  * Now check if this objectClass means we need to do urgent replication
1982                  */
1983                 if (!*is_urgent && replmd_check_urgent_objectclass(objectclass_el,
1984                                                                    situation)) {
1985                         *is_urgent = true;
1986                 }
1987         } else if (!ldb_request_get_control(req, DSDB_CONTROL_DBCHECK)) {
1988                 ldb_asprintf_errstring(ldb, __location__
1989                                        ": objectClass missing on %s\n",
1990                                        ldb_dn_get_linearized(msg->dn));
1991                 return LDB_ERR_OBJECT_CLASS_VIOLATION;
1992         }
1993
1994         /*
1995          * replmd_update_rpmd_element has done an update if the
1996          * seq_num is set
1997          */
1998         if (*seq_num != 0 || rmd_is_just_resorted == true) {
1999                 struct ldb_val *md_value;
2000                 struct ldb_message_element *el;
2001
2002                 /*if we are RODC and this is a DRSR update then its ok*/
2003                 if (!ldb_request_get_control(req, DSDB_CONTROL_REPLICATED_UPDATE_OID)
2004                     && !ldb_request_get_control(req, DSDB_CONTROL_DBCHECK_MODIFY_RO_REPLICA)
2005                     && !is_forced_rodc) {
2006                         unsigned instanceType;
2007
2008                         if (*rodc) {
2009                                 ldb_set_errstring(ldb, "RODC modify is forbidden!");
2010                                 return LDB_ERR_REFERRAL;
2011                         }
2012
2013                         instanceType = ldb_msg_find_attr_as_uint(res->msgs[0], "instanceType", INSTANCE_TYPE_WRITE);
2014                         if (!(instanceType & INSTANCE_TYPE_WRITE)) {
2015                                 return ldb_error(ldb, LDB_ERR_UNWILLING_TO_PERFORM,
2016                                                  "cannot change replicated attribute on partial replica");
2017                         }
2018                 }
2019
2020                 md_value = talloc(msg, struct ldb_val);
2021                 if (md_value == NULL) {
2022                         ldb_oom(ldb);
2023                         return LDB_ERR_OPERATIONS_ERROR;
2024                 }
2025
2026                 ret = replmd_replPropertyMetaDataCtr1_sort_and_verify(ldb, &omd.ctr.ctr1, msg->dn);
2027                 if (ret != LDB_SUCCESS) {
2028                         ldb_asprintf_errstring(ldb, "%s: %s", __func__, ldb_errstring(ldb));
2029                         return ret;
2030                 }
2031
2032                 ndr_err = ndr_push_struct_blob(md_value, msg, &omd,
2033                                                (ndr_push_flags_fn_t)ndr_push_replPropertyMetaDataBlob);
2034                 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
2035                         DEBUG(0,(__location__ ": Failed to marshall replPropertyMetaData for %s\n",
2036                                  ldb_dn_get_linearized(msg->dn)));
2037                         return LDB_ERR_OPERATIONS_ERROR;
2038                 }
2039
2040                 ret = ldb_msg_add_empty(msg, "replPropertyMetaData", LDB_FLAG_MOD_REPLACE, &el);
2041                 if (ret != LDB_SUCCESS) {
2042                         DEBUG(0,(__location__ ": Failed to add updated replPropertyMetaData %s\n",
2043                                  ldb_dn_get_linearized(msg->dn)));
2044                         return ret;
2045                 }
2046
2047                 el->num_values = 1;
2048                 el->values = md_value;
2049         }
2050
2051         return LDB_SUCCESS;
2052 }
2053
2054 static int parsed_dn_compare(struct parsed_dn *pdn1, struct parsed_dn *pdn2)
2055 {
2056         int ret = ndr_guid_compare(&pdn1->guid, &pdn2->guid);
2057         if (ret == 0) {
2058                 return data_blob_cmp(&pdn1->dsdb_dn->extra_part,
2059                                      &pdn2->dsdb_dn->extra_part);
2060         }
2061         return ret;
2062 }
2063
2064 /*
2065   get a series of message element values as an array of DNs and GUIDs
2066   the result is sorted by GUID
2067  */
2068 static int get_parsed_dns(struct ldb_module *module, TALLOC_CTX *mem_ctx,
2069                           struct ldb_message_element *el, struct parsed_dn **pdn,
2070                           const char *ldap_oid, struct ldb_request *parent)
2071 {
2072         unsigned int i;
2073         bool values_are_sorted = true;
2074         struct ldb_context *ldb = ldb_module_get_ctx(module);
2075
2076         if (el == NULL) {
2077                 *pdn = NULL;
2078                 return LDB_SUCCESS;
2079         }
2080
2081         (*pdn) = talloc_array(mem_ctx, struct parsed_dn, el->num_values);
2082         if (!*pdn) {
2083                 ldb_module_oom(module);
2084                 return LDB_ERR_OPERATIONS_ERROR;
2085         }
2086
2087         for (i=0; i<el->num_values; i++) {
2088                 struct ldb_val *v = &el->values[i];
2089                 NTSTATUS status;
2090                 struct ldb_dn *dn;
2091                 struct parsed_dn *p;
2092
2093                 p = &(*pdn)[i];
2094
2095                 p->dsdb_dn = dsdb_dn_parse(*pdn, ldb, v, ldap_oid);
2096                 if (p->dsdb_dn == NULL) {
2097                         return LDB_ERR_INVALID_DN_SYNTAX;
2098                 }
2099
2100                 dn = p->dsdb_dn->dn;
2101
2102                 status = dsdb_get_extended_dn_guid(dn, &p->guid, "GUID");
2103                 if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND) ||
2104                     unlikely(GUID_all_zero(&p->guid))) {
2105                         /* we got a DN without a GUID - go find the GUID */
2106                         int ret = dsdb_module_guid_by_dn(module, dn, &p->guid, parent);
2107                         if (ret != LDB_SUCCESS) {
2108                                 char *dn_str = NULL;
2109                                 dn_str = ldb_dn_get_extended_linearized(mem_ctx,
2110                                                                         (dn), 1);
2111                                 ldb_asprintf_errstring(ldb,
2112                                                 "Unable to find GUID for DN %s\n",
2113                                                 dn_str);
2114                                 if (ret == LDB_ERR_NO_SUCH_OBJECT &&
2115                                     LDB_FLAG_MOD_TYPE(el->flags) == LDB_FLAG_MOD_DELETE &&
2116                                     ldb_attr_cmp(el->name, "member") == 0) {
2117                                         return LDB_ERR_UNWILLING_TO_PERFORM;
2118                                 }
2119                                 return ret;
2120                         }
2121                         ret = dsdb_set_extended_dn_guid(dn, &p->guid, "GUID");
2122                         if (ret != LDB_SUCCESS) {
2123                                 return ret;
2124                         }
2125                 } else if (!NT_STATUS_IS_OK(status)) {
2126                         return LDB_ERR_OPERATIONS_ERROR;
2127                 }
2128                 if (i > 0 && values_are_sorted) {
2129                         int cmp = parsed_dn_compare(p, &(*pdn)[i - 1]);
2130                         if (cmp < 0) {
2131                                 values_are_sorted = false;
2132                         }
2133                 }
2134                 /* keep a pointer to the original ldb_val */
2135                 p->v = v;
2136         }
2137         if (! values_are_sorted) {
2138                 TYPESAFE_QSORT(*pdn, el->num_values, parsed_dn_compare);
2139         }
2140         return LDB_SUCCESS;
2141 }
2142
2143 /*
2144  * Get a series of trusted message element values. The result is sorted by
2145  * GUID, even though the GUIDs might not be known. That works because we trust
2146  * the database to give us the elements like that if the
2147  * replmd_private->sorted_links flag is set.
2148  *
2149  * We also ensure that the links are in the Functional Level 2003
2150  * linked attributes format.
2151  */
2152 static int get_parsed_dns_trusted_fallback(struct ldb_module *module,
2153                                            struct replmd_private *replmd_private,
2154                                            TALLOC_CTX *mem_ctx,
2155                                            struct ldb_message_element *el,
2156                                            struct parsed_dn **pdn,
2157                                            const char *ldap_oid,
2158                                            struct ldb_request *parent)
2159 {
2160         int ret;
2161         if (el == NULL) {
2162                 *pdn = NULL;
2163                 return LDB_SUCCESS;
2164         }
2165
2166         if (!replmd_private->sorted_links) {
2167                 /* We need to sort the list. This is the slow old path we want
2168                    to avoid.
2169                  */
2170                 ret = get_parsed_dns(module, mem_ctx, el, pdn, ldap_oid,
2171                                       parent);
2172                 if (ret != LDB_SUCCESS) {
2173                         return ret;
2174                 }
2175         } else {
2176                 ret = get_parsed_dns_trusted(mem_ctx, el, pdn);
2177                 if (ret != LDB_SUCCESS) {
2178                         ldb_module_oom(module);
2179                         return LDB_ERR_OPERATIONS_ERROR;
2180                 }
2181         }
2182
2183         /*
2184          * This upgrades links to FL2003 style, and sorts the result
2185          * if that was needed.
2186          *
2187          * TODO: Add a database feature that asserts we have no FL2000
2188          *       style links to avoid this check or add a feature that
2189          *       uses a similar check to find sorted/unsorted links
2190          *       for an on-the-fly upgrade.
2191          */
2192
2193         ret = replmd_check_upgrade_links(ldb_module_get_ctx(module),
2194                                          *pdn, el->num_values,
2195                                          el,
2196                                          ldap_oid);
2197         if (ret != LDB_SUCCESS) {
2198                 return ret;
2199         }
2200
2201         return LDB_SUCCESS;
2202 }
2203
2204 /*
2205    Return LDB_SUCCESS if a parsed_dn list contains no duplicate values,
2206    otherwise an error code. For compatibility the error code differs depending
2207    on whether or not the attribute is "member".
2208
2209    As always, the parsed_dn list is assumed to be sorted.
2210  */
2211 static int check_parsed_dn_duplicates(struct ldb_module *module,
2212                                       struct ldb_message_element *el,
2213                                       struct parsed_dn *pdn)
2214 {
2215         unsigned int i;
2216         struct ldb_context *ldb = ldb_module_get_ctx(module);
2217
2218         for (i = 1; i < el->num_values; i++) {
2219                 struct parsed_dn *p = &pdn[i];
2220                 if (parsed_dn_compare(p, &pdn[i - 1]) == 0) {
2221                         ldb_asprintf_errstring(ldb,
2222                                                "Linked attribute %s has "
2223                                                "multiple identical values",
2224                                                el->name);
2225                         if (ldb_attr_cmp(el->name, "member") == 0) {
2226                                 return LDB_ERR_ENTRY_ALREADY_EXISTS;
2227                         } else {
2228                                 return LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
2229                         }
2230                 }
2231         }
2232         return LDB_SUCCESS;
2233 }
2234
2235 /*
2236   build a new extended DN, including all meta data fields
2237
2238   RMD_FLAGS           = DSDB_RMD_FLAG_* bits
2239   RMD_ADDTIME         = originating_add_time
2240   RMD_INVOCID         = originating_invocation_id
2241   RMD_CHANGETIME      = originating_change_time
2242   RMD_ORIGINATING_USN = originating_usn
2243   RMD_LOCAL_USN       = local_usn
2244   RMD_VERSION         = version
2245  */
2246 static int replmd_build_la_val(TALLOC_CTX *mem_ctx, struct ldb_val *v,
2247                                struct dsdb_dn *dsdb_dn,
2248                                const struct GUID *invocation_id,
2249                                uint64_t local_usn, NTTIME nttime)
2250 {
2251         return replmd_set_la_val(mem_ctx, v, dsdb_dn, NULL, invocation_id,
2252                                  local_usn, local_usn, nttime,
2253                                  RMD_VERSION_INITIAL, false);
2254 }
2255
2256 static int replmd_update_la_val(TALLOC_CTX *mem_ctx, struct ldb_val *v, struct dsdb_dn *dsdb_dn,
2257                                 struct dsdb_dn *old_dsdb_dn, const struct GUID *invocation_id,
2258                                 uint64_t seq_num, uint64_t local_usn, NTTIME nttime,
2259                                 bool deleted);
2260
2261 /*
2262   check if any links need upgrading from w2k format
2263  */
2264 static int replmd_check_upgrade_links(struct ldb_context *ldb,
2265                                       struct parsed_dn *dns, uint32_t count,
2266                                       struct ldb_message_element *el,
2267                                       const char *ldap_oid)
2268 {
2269         uint32_t i;
2270         const struct GUID *invocation_id = NULL;
2271         for (i=0; i<count; i++) {
2272                 NTSTATUS status;
2273                 uint32_t version;
2274                 int ret;
2275                 if (dns[i].dsdb_dn == NULL) {
2276                         ret = really_parse_trusted_dn(dns, ldb, &dns[i],
2277                                                       ldap_oid);
2278                         if (ret != LDB_SUCCESS) {
2279                                 return LDB_ERR_INVALID_DN_SYNTAX;
2280                         }
2281                 }
2282
2283                 status = dsdb_get_extended_dn_uint32(dns[i].dsdb_dn->dn,
2284                                                      &version, "RMD_VERSION");
2285                 if (!NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
2286                         /*
2287                          *  We optimistically assume they are all the same; if
2288                          *  the first one is fixed, they are all fixed.
2289                          *
2290                          *  If the first one was *not* fixed and we find a
2291                          *  later one that is, that is an occasion to shout
2292                          *  with DEBUG(0).
2293                          */
2294                         if (i == 0) {
2295                                 return LDB_SUCCESS;
2296                         }
2297                         DEBUG(0, ("Mixed w2k and fixed format "
2298                                   "linked attributes\n"));
2299                         continue;
2300                 }
2301
2302                 if (invocation_id == NULL) {
2303                         invocation_id = samdb_ntds_invocation_id(ldb);
2304                         if (invocation_id == NULL) {
2305                                 return LDB_ERR_OPERATIONS_ERROR;
2306                         }
2307                 }
2308
2309
2310                 /* it's an old one that needs upgrading */
2311                 ret = replmd_update_la_val(el->values, dns[i].v,
2312                                            dns[i].dsdb_dn, dns[i].dsdb_dn,
2313                                            invocation_id, 1, 1, 0, false);
2314                 if (ret != LDB_SUCCESS) {
2315                         return ret;
2316                 }
2317         }
2318
2319         /*
2320          * This sort() is critical for the operation of
2321          * get_parsed_dns_trusted_fallback() because callers of this function
2322          * expect a sorted list, and FL2000 style links are not
2323          * sorted.  In particular, as well as the upgrade case,
2324          * get_parsed_dns_trusted_fallback() is called from
2325          * replmd_delete_remove_link() even in FL2000 mode
2326          *
2327          * We do not normally pay the cost of the qsort() due to the
2328          * early return in the RMD_VERSION found case.
2329          */
2330         TYPESAFE_QSORT(dns, count, parsed_dn_compare);
2331         return LDB_SUCCESS;
2332 }
2333
2334 /*
2335   Sets the value for a linked attribute, including all meta data fields
2336
2337   see replmd_build_la_val for value names
2338  */
2339 static int replmd_set_la_val(TALLOC_CTX *mem_ctx, struct ldb_val *v, struct dsdb_dn *dsdb_dn,
2340                              struct dsdb_dn *old_dsdb_dn, const struct GUID *invocation_id,
2341                              uint64_t usn, uint64_t local_usn, NTTIME nttime,
2342                              uint32_t version, bool deleted)
2343 {
2344         struct ldb_dn *dn = dsdb_dn->dn;
2345         const char *tstring, *usn_string, *flags_string;
2346         struct ldb_val tval;
2347         struct ldb_val iid;
2348         struct ldb_val usnv, local_usnv;
2349         struct ldb_val vers, flagsv;
2350         const struct ldb_val *old_addtime = NULL;
2351         NTSTATUS status;
2352         int ret;
2353         const char *dnstring;
2354         char *vstring;
2355         uint32_t rmd_flags = deleted?DSDB_RMD_FLAG_DELETED:0;
2356
2357         tstring = talloc_asprintf(mem_ctx, "%llu", (unsigned long long)nttime);
2358         if (!tstring) {
2359                 return LDB_ERR_OPERATIONS_ERROR;
2360         }
2361         tval = data_blob_string_const(tstring);
2362
2363         usn_string = talloc_asprintf(mem_ctx, "%llu", (unsigned long long)usn);
2364         if (!usn_string) {
2365                 return LDB_ERR_OPERATIONS_ERROR;
2366         }
2367         usnv = data_blob_string_const(usn_string);
2368
2369         usn_string = talloc_asprintf(mem_ctx, "%llu", (unsigned long long)local_usn);
2370         if (!usn_string) {
2371                 return LDB_ERR_OPERATIONS_ERROR;
2372         }
2373         local_usnv = data_blob_string_const(usn_string);
2374
2375         status = GUID_to_ndr_blob(invocation_id, dn, &iid);
2376         if (!NT_STATUS_IS_OK(status)) {
2377                 return LDB_ERR_OPERATIONS_ERROR;
2378         }
2379
2380         flags_string = talloc_asprintf(mem_ctx, "%u", rmd_flags);
2381         if (!flags_string) {
2382                 return LDB_ERR_OPERATIONS_ERROR;
2383         }
2384         flagsv = data_blob_string_const(flags_string);
2385
2386         ret = ldb_dn_set_extended_component(dn, "RMD_FLAGS", &flagsv);
2387         if (ret != LDB_SUCCESS) return ret;
2388
2389         /* get the ADDTIME from the original */
2390         if (old_dsdb_dn != NULL) {
2391                 old_addtime = ldb_dn_get_extended_component(old_dsdb_dn->dn,
2392                                                             "RMD_ADDTIME");
2393         }
2394         if (old_addtime == NULL) {
2395                 old_addtime = &tval;
2396         }
2397         if (dsdb_dn != old_dsdb_dn ||
2398             ldb_dn_get_extended_component(dn, "RMD_ADDTIME") == NULL) {
2399                 ret = ldb_dn_set_extended_component(dn, "RMD_ADDTIME", old_addtime);
2400                 if (ret != LDB_SUCCESS) return ret;
2401         }
2402
2403         /* use our invocation id */
2404         ret = ldb_dn_set_extended_component(dn, "RMD_INVOCID", &iid);
2405         if (ret != LDB_SUCCESS) return ret;
2406
2407         /* changetime is the current time */
2408         ret = ldb_dn_set_extended_component(dn, "RMD_CHANGETIME", &tval);
2409         if (ret != LDB_SUCCESS) return ret;
2410
2411         /* update the USN */
2412         ret = ldb_dn_set_extended_component(dn, "RMD_ORIGINATING_USN", &usnv);
2413         if (ret != LDB_SUCCESS) return ret;
2414
2415         ret = ldb_dn_set_extended_component(dn, "RMD_LOCAL_USN", &local_usnv);
2416         if (ret != LDB_SUCCESS) return ret;
2417
2418         vstring = talloc_asprintf(mem_ctx, "%lu", (unsigned long)version);
2419         vers = data_blob_string_const(vstring);
2420         ret = ldb_dn_set_extended_component(dn, "RMD_VERSION", &vers);
2421         if (ret != LDB_SUCCESS) return ret;
2422
2423         dnstring = dsdb_dn_get_extended_linearized(mem_ctx, dsdb_dn, 1);
2424         if (dnstring == NULL) {
2425                 return LDB_ERR_OPERATIONS_ERROR;
2426         }
2427         *v = data_blob_string_const(dnstring);
2428
2429         return LDB_SUCCESS;
2430 }
2431
2432 /**
2433  * Updates the value for a linked attribute, including all meta data fields
2434  */
2435 static int replmd_update_la_val(TALLOC_CTX *mem_ctx, struct ldb_val *v, struct dsdb_dn *dsdb_dn,
2436                                 struct dsdb_dn *old_dsdb_dn, const struct GUID *invocation_id,
2437                                 uint64_t usn, uint64_t local_usn, NTTIME nttime,
2438                                 bool deleted)
2439 {
2440         uint32_t old_version;
2441         uint32_t version = RMD_VERSION_INITIAL;
2442         NTSTATUS status;
2443
2444         /*
2445          * We're updating the linked attribute locally, so increase the version
2446          * by 1 so that other DCs will see the change when it gets replicated out
2447          */
2448         status = dsdb_get_extended_dn_uint32(old_dsdb_dn->dn, &old_version,
2449                                              "RMD_VERSION");
2450
2451         if (NT_STATUS_IS_OK(status)) {
2452                 version = old_version + 1;
2453         }
2454
2455         return replmd_set_la_val(mem_ctx, v, dsdb_dn, old_dsdb_dn, invocation_id,
2456                                  usn, local_usn, nttime, version, deleted);
2457 }
2458
2459 /*
2460   handle adding a linked attribute
2461  */
2462 static int replmd_modify_la_add(struct ldb_module *module,
2463                                 struct replmd_private *replmd_private,
2464                                 struct replmd_replicated_request *ac,
2465                                 struct ldb_message *msg,
2466                                 struct ldb_message_element *el,
2467                                 struct ldb_message_element *old_el,
2468                                 const struct dsdb_attribute *schema_attr,
2469                                 time_t t,
2470                                 struct ldb_dn *msg_dn,
2471                                 struct ldb_request *parent)
2472 {
2473         unsigned int i, j;
2474         struct parsed_dn *dns, *old_dns;
2475         TALLOC_CTX *tmp_ctx = talloc_new(msg);
2476         int ret;
2477         struct ldb_val *new_values = NULL;
2478         unsigned old_num_values = old_el ? old_el->num_values : 0;
2479         unsigned num_values = 0;
2480         unsigned max_num_values;
2481         struct ldb_context *ldb = ldb_module_get_ctx(module);
2482         NTTIME now;
2483         unix_to_nt_time(&now, t);
2484
2485         /* get the DNs to be added, fully parsed.
2486          *
2487          * We need full parsing because they came off the wire and we don't
2488          * trust them, besides which we need their details to know where to put
2489          * them.
2490          */
2491         ret = get_parsed_dns(module, tmp_ctx, el, &dns,
2492                              schema_attr->syntax->ldap_oid, parent);
2493         if (ret != LDB_SUCCESS) {
2494                 talloc_free(tmp_ctx);
2495                 return ret;
2496         }
2497
2498         /* get the existing DNs, lazily parsed */
2499         ret = get_parsed_dns_trusted_fallback(module, replmd_private,
2500                                               tmp_ctx, old_el, &old_dns,
2501                                               schema_attr->syntax->ldap_oid,
2502                                               parent);
2503
2504         if (ret != LDB_SUCCESS) {
2505                 talloc_free(tmp_ctx);
2506                 return ret;
2507         }
2508
2509         max_num_values = old_num_values + el->num_values;
2510         if (max_num_values < old_num_values) {
2511                 DEBUG(0, ("we seem to have overflow in replmd_modify_la_add. "
2512                           "old values: %u, new values: %u, sum: %u\n",
2513                           old_num_values, el->num_values, max_num_values));
2514                 talloc_free(tmp_ctx);
2515                 return LDB_ERR_OPERATIONS_ERROR;
2516         }
2517
2518         new_values = talloc_zero_array(tmp_ctx, struct ldb_val, max_num_values);
2519
2520         if (new_values == NULL) {
2521                 ldb_module_oom(module);
2522                 talloc_free(tmp_ctx);
2523                 return LDB_ERR_OPERATIONS_ERROR;
2524         }
2525
2526         /*
2527          * For each new value, find where it would go in the list. If there is
2528          * a matching GUID there, we update the existing value; otherwise we
2529          * put it in place.
2530          */
2531         j = 0;
2532         for (i = 0; i < el->num_values; i++) {
2533                 struct parsed_dn *exact;
2534                 struct parsed_dn *next;
2535                 unsigned offset;
2536                 int err = parsed_dn_find(ldb, old_dns, old_num_values,
2537                                          &dns[i].guid,
2538                                          dns[i].dsdb_dn->dn,
2539                                          dns[i].dsdb_dn->extra_part, 0,
2540                                          &exact, &next,
2541                                          schema_attr->syntax->ldap_oid,
2542                                          true);
2543                 if (err != LDB_SUCCESS) {
2544                         talloc_free(tmp_ctx);
2545                         return err;
2546                 }
2547
2548                 if (ac->fix_link_sid) {
2549                         char *fixed_dnstring = NULL;
2550                         struct dom_sid tmp_sid = { 0, };
2551                         DATA_BLOB sid_blob = data_blob_null;
2552                         enum ndr_err_code ndr_err;
2553                         NTSTATUS status;
2554                         int num;
2555
2556                         if (exact == NULL) {
2557                                 talloc_free(tmp_ctx);
2558                                 return ldb_operr(ldb);
2559                         }
2560
2561                         if (dns[i].dsdb_dn->dn_format != DSDB_NORMAL_DN) {
2562                                 talloc_free(tmp_ctx);
2563                                 return ldb_operr(ldb);
2564                         }
2565
2566                         /*
2567                          * Only "<GUID=...><SID=...>" is allowed.
2568                          *
2569                          * We get the GUID to just to find the old
2570                          * value and the SID in order to add it
2571                          * to the found value.
2572                          */
2573
2574                         num = ldb_dn_get_comp_num(dns[i].dsdb_dn->dn);
2575                         if (num != 0) {
2576                                 talloc_free(tmp_ctx);
2577                                 return ldb_operr(ldb);
2578                         }
2579
2580                         num = ldb_dn_get_extended_comp_num(dns[i].dsdb_dn->dn);
2581                         if (num != 2) {
2582                                 talloc_free(tmp_ctx);
2583                                 return ldb_operr(ldb);
2584                         }
2585
2586                         status = dsdb_get_extended_dn_sid(exact->dsdb_dn->dn,
2587                                                           &tmp_sid, "SID");
2588                         if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
2589                                 /* this is what we expect */
2590                         } else if (NT_STATUS_IS_OK(status)) {
2591                                 struct GUID_txt_buf guid_str;
2592                                 ldb_debug_set(ldb, LDB_DEBUG_FATAL,
2593                                                        "i[%u] SID NOT MISSING... Attribute %s already "
2594                                                        "exists for target GUID %s, SID %s, DN: %s",
2595                                                        i, el->name,
2596                                                        GUID_buf_string(&exact->guid,
2597                                                                        &guid_str),
2598                                                        dom_sid_string(tmp_ctx, &tmp_sid),
2599                                                        dsdb_dn_get_extended_linearized(tmp_ctx,
2600                                                                exact->dsdb_dn, 1));
2601                                 talloc_free(tmp_ctx);
2602                                 return LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
2603                         } else {
2604                                 talloc_free(tmp_ctx);
2605                                 return ldb_operr(ldb);
2606                         }
2607
2608                         status = dsdb_get_extended_dn_sid(dns[i].dsdb_dn->dn,
2609                                                           &tmp_sid, "SID");
2610                         if (!NT_STATUS_IS_OK(status)) {
2611                                 struct GUID_txt_buf guid_str;
2612                                 ldb_asprintf_errstring(ldb,
2613                                                        "NO SID PROVIDED... Attribute %s already "
2614                                                        "exists for target GUID %s",
2615                                                        el->name,
2616                                                        GUID_buf_string(&exact->guid,
2617                                                                        &guid_str));
2618                                 talloc_free(tmp_ctx);
2619                                 return LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
2620                         }
2621
2622                         ndr_err = ndr_push_struct_blob(&sid_blob, tmp_ctx, &tmp_sid,
2623                                                        (ndr_push_flags_fn_t)ndr_push_dom_sid);
2624                         if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
2625                                 talloc_free(tmp_ctx);
2626                                 return ldb_operr(ldb);
2627                         }
2628
2629                         ret = ldb_dn_set_extended_component(exact->dsdb_dn->dn, "SID", &sid_blob);
2630                         data_blob_free(&sid_blob);
2631                         if (ret != LDB_SUCCESS) {
2632                                 talloc_free(tmp_ctx);
2633                                 return ret;
2634                         }
2635
2636                         fixed_dnstring = dsdb_dn_get_extended_linearized(
2637                                         new_values, exact->dsdb_dn, 1);
2638                         if (fixed_dnstring == NULL) {
2639                                 talloc_free(tmp_ctx);
2640                                 return ldb_operr(ldb);
2641                         }
2642
2643                         /*
2644                          * We just replace the existing value...
2645                          */
2646                         *exact->v = data_blob_string_const(fixed_dnstring);
2647
2648                         continue;
2649                 }
2650
2651                 if (exact != NULL) {
2652                         /*
2653                          * We are trying to add one that exists, which is only
2654                          * allowed if it was previously deleted.
2655                          *
2656                          * When we do undelete a link we change it in place.
2657                          * It will be copied across into the right spot in due
2658                          * course.
2659                          */
2660                         uint32_t rmd_flags;
2661                         rmd_flags = dsdb_dn_rmd_flags(exact->dsdb_dn->dn);
2662
2663                         if (!(rmd_flags & DSDB_RMD_FLAG_DELETED)) {
2664                                 struct GUID_txt_buf guid_str;
2665                                 ldb_asprintf_errstring(ldb,
2666                                                        "Attribute %s already "
2667                                                        "exists for target GUID %s",
2668                                                        el->name,
2669                                                        GUID_buf_string(&exact->guid,
2670                                                                        &guid_str));
2671                                 talloc_free(tmp_ctx);
2672                                 /* error codes for 'member' need to be
2673                                    special cased */
2674                                 if (ldb_attr_cmp(el->name, "member") == 0) {
2675                                         return LDB_ERR_ENTRY_ALREADY_EXISTS;
2676                                 } else {
2677                                         return LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
2678                                 }
2679                         }
2680
2681                         ret = replmd_update_la_val(new_values, exact->v,
2682                                                    dns[i].dsdb_dn,
2683                                                    exact->dsdb_dn,
2684                                                    &ac->our_invocation_id,
2685                                                    ac->seq_num, ac->seq_num,
2686                                                    now, false);
2687                         if (ret != LDB_SUCCESS) {
2688                                 talloc_free(tmp_ctx);
2689                                 return ret;
2690                         }
2691
2692                         ret = replmd_add_backlink(module, replmd_private,
2693                                                   ac->schema,
2694                                                   msg_dn,
2695                                                   &dns[i].guid, 
2696                                                   true,
2697                                                   schema_attr,
2698                                                   parent);
2699                         if (ret != LDB_SUCCESS) {
2700                                 talloc_free(tmp_ctx);
2701                                 return ret;
2702                                 }
2703                         continue;
2704                 }
2705                 /*
2706                  * Here we don't have an exact match.
2707                  *
2708                  * If next is NULL, this one goes beyond the end of the
2709                  * existing list, so we need to add all of those ones first.
2710                  *
2711                  * If next is not NULL, we need to add all the ones before
2712                  * next.
2713                  */
2714                 if (next == NULL) {
2715                         offset = old_num_values;
2716                 } else {
2717                         /* next should have been parsed, but let's make sure */
2718                         if (next->dsdb_dn == NULL) {
2719                                 ret = really_parse_trusted_dn(tmp_ctx, ldb, next,
2720                                                               schema_attr->syntax->ldap_oid);
2721                                 if (ret != LDB_SUCCESS) {
2722                                         return ret;
2723                                 }
2724                         }
2725                         offset = MIN(next - old_dns, old_num_values);
2726                 }
2727
2728                 /* put all the old ones before next on the list */
2729                 for (; j < offset; j++) {
2730                         new_values[num_values] = *old_dns[j].v;
2731                         num_values++;
2732                 }
2733
2734                 ret = replmd_add_backlink(module, replmd_private,
2735                                           ac->schema, msg_dn,
2736                                           &dns[i].guid,
2737                                           true, schema_attr,
2738                                           parent);
2739                 /* Make the new linked attribute ldb_val. */
2740                 ret = replmd_build_la_val(new_values, &new_values[num_values],
2741                                           dns[i].dsdb_dn, &ac->our_invocation_id,
2742                                           ac->seq_num, now);
2743                 if (ret != LDB_SUCCESS) {
2744                         talloc_free(tmp_ctx);
2745                         return ret;
2746                 }
2747                 num_values++;
2748                 if (ret != LDB_SUCCESS) {
2749                         talloc_free(tmp_ctx);
2750                         return ret;
2751                 }
2752         }
2753         /* copy the rest of the old ones (if any) */
2754         for (; j < old_num_values; j++) {
2755                 new_values[num_values] = *old_dns[j].v;
2756                 num_values++;
2757         }
2758
2759         talloc_steal(msg->elements, new_values);
2760         if (old_el != NULL) {
2761                 talloc_steal(msg->elements, old_el->values);
2762         }
2763         el->values = new_values;
2764         el->num_values = num_values;
2765
2766         talloc_free(tmp_ctx);
2767
2768         /* we now tell the backend to replace all existing values
2769            with the one we have constructed */
2770         el->flags = LDB_FLAG_MOD_REPLACE;
2771
2772         return LDB_SUCCESS;
2773 }
2774
2775
2776 /*
2777   handle deleting all active linked attributes
2778  */
2779 static int replmd_modify_la_delete(struct ldb_module *module,
2780                                    struct replmd_private *replmd_private,
2781                                    struct replmd_replicated_request *ac,
2782                                    struct ldb_message *msg,
2783                                    struct ldb_message_element *el,
2784                                    struct ldb_message_element *old_el,
2785                                    const struct dsdb_attribute *schema_attr,
2786                                    time_t t,
2787                                    struct ldb_dn *msg_dn,
2788                                    struct ldb_request *parent)
2789 {
2790         unsigned int i;
2791         struct parsed_dn *dns, *old_dns;
2792         TALLOC_CTX *tmp_ctx = NULL;
2793         int ret;
2794         struct ldb_context *ldb = ldb_module_get_ctx(module);
2795         struct ldb_control *vanish_links_ctrl = NULL;
2796         bool vanish_links = false;
2797         unsigned int num_to_delete = el->num_values;
2798         uint32_t rmd_flags;
2799         NTTIME now;
2800
2801         unix_to_nt_time(&now, t);
2802
2803         if (old_el == NULL || old_el->num_values == 0) {
2804                 /* there is nothing to delete... */
2805                 if (num_to_delete == 0) {
2806                         /* and we're deleting nothing, so that's OK */
2807                         return LDB_SUCCESS;
2808                 }
2809                 return LDB_ERR_NO_SUCH_ATTRIBUTE;
2810         }
2811
2812         tmp_ctx = talloc_new(msg);
2813         if (tmp_ctx == NULL) {
2814                 return LDB_ERR_OPERATIONS_ERROR;
2815         }
2816
2817         ret = get_parsed_dns(module, tmp_ctx, el, &dns,
2818                              schema_attr->syntax->ldap_oid, parent);
2819         if (ret != LDB_SUCCESS) {
2820                 talloc_free(tmp_ctx);
2821                 return ret;
2822         }
2823
2824         ret = get_parsed_dns_trusted_fallback(module, replmd_private,
2825                                               tmp_ctx, old_el, &old_dns,
2826                                               schema_attr->syntax->ldap_oid,
2827                                               parent);
2828
2829         if (ret != LDB_SUCCESS) {
2830                 talloc_free(tmp_ctx);
2831                 return ret;
2832         }
2833
2834         vanish_links_ctrl = ldb_request_get_control(parent, DSDB_CONTROL_REPLMD_VANISH_LINKS);
2835         if (vanish_links_ctrl) {
2836                 vanish_links = true;
2837                 vanish_links_ctrl->critical = false;
2838         }
2839
2840         /* we empty out el->values here to avoid damage if we return early. */
2841         el->num_values = 0;
2842         el->values = NULL;
2843
2844         /*
2845          * If vanish links is set, we are actually removing members of
2846          *  old_el->values; otherwise we are just marking them deleted.
2847          *
2848          * There is a special case when no values are given: we remove them
2849          * all. When we have the vanish_links control we just have to remove
2850          * the backlinks and change our element to replace the existing values
2851          * with the empty list.
2852          */
2853
2854         if (num_to_delete == 0) {
2855                 for (i = 0; i < old_el->num_values; i++) {
2856                         struct parsed_dn *p = &old_dns[i];
2857                         if (p->dsdb_dn == NULL) {
2858                                 ret = really_parse_trusted_dn(tmp_ctx, ldb, p,
2859                                                               schema_attr->syntax->ldap_oid);
2860                                 if (ret != LDB_SUCCESS) {
2861                                         return ret;
2862                                 }
2863                         }
2864                         ret = replmd_add_backlink(module, replmd_private,
2865                                                   ac->schema, msg_dn, &p->guid,
2866                                                   false, schema_attr,
2867                                                   parent);
2868                         if (ret != LDB_SUCCESS) {
2869                                 talloc_free(tmp_ctx);
2870                                 return ret;
2871                         }
2872                         if (vanish_links) {
2873                                 continue;
2874                         }
2875
2876                         rmd_flags = dsdb_dn_rmd_flags(p->dsdb_dn->dn);
2877                         if (rmd_flags & DSDB_RMD_FLAG_DELETED) {
2878                                 continue;
2879                         }
2880
2881                         ret = replmd_update_la_val(old_el->values, p->v,
2882                                                    p->dsdb_dn, p->dsdb_dn,
2883                                                    &ac->our_invocation_id,
2884                                                    ac->seq_num, ac->seq_num,
2885                                                    now, true);
2886                         if (ret != LDB_SUCCESS) {
2887                                 talloc_free(tmp_ctx);
2888                                 return ret;
2889                         }
2890                 }
2891
2892                 if (vanish_links) {
2893                         el->flags = LDB_FLAG_MOD_REPLACE;
2894                         talloc_free(tmp_ctx);
2895                         return LDB_SUCCESS;
2896                 }
2897         }
2898
2899
2900         for (i = 0; i < num_to_delete; i++) {
2901                 struct parsed_dn *p = &dns[i];
2902                 struct parsed_dn *exact = NULL;
2903                 struct parsed_dn *next = NULL;
2904                 ret = parsed_dn_find(ldb, old_dns, old_el->num_values,
2905                                      &p->guid,
2906                                      NULL,
2907                                      p->dsdb_dn->extra_part, 0,
2908                                      &exact, &next,
2909                                      schema_attr->syntax->ldap_oid,
2910                                      true);
2911                 if (ret != LDB_SUCCESS) {
2912                         talloc_free(tmp_ctx);
2913                         return ret;
2914                 }
2915                 if (exact == NULL) {
2916                         struct GUID_txt_buf buf;
2917                         ldb_asprintf_errstring(ldb, "Attribute %s doesn't "
2918                                                "exist for target GUID %s",
2919                                                el->name,
2920                                                GUID_buf_string(&p->guid, &buf));
2921                         if (ldb_attr_cmp(el->name, "member") == 0) {
2922                                 talloc_free(tmp_ctx);
2923                                 return LDB_ERR_UNWILLING_TO_PERFORM;
2924                         } else {
2925                                 talloc_free(tmp_ctx);
2926                                 return LDB_ERR_NO_SUCH_ATTRIBUTE;
2927                         }
2928                 }
2929
2930                 if (vanish_links) {
2931                         if (CHECK_DEBUGLVL(5)) {
2932                                 rmd_flags = dsdb_dn_rmd_flags(exact->dsdb_dn->dn);
2933                                 if ((rmd_flags & DSDB_RMD_FLAG_DELETED)) {
2934                                         struct GUID_txt_buf buf;
2935                                         const char *guid_str = \
2936                                                 GUID_buf_string(&p->guid, &buf);
2937                                         DEBUG(5, ("Deleting deleted linked "
2938                                                   "attribute %s to %s, because "
2939                                                   "vanish_links control is set\n",
2940                                                   el->name, guid_str));
2941                                 }
2942                         }
2943
2944                         /* remove the backlink */
2945                         ret = replmd_add_backlink(module,
2946                                                   replmd_private,
2947                                                   ac->schema,
2948                                                   msg_dn,
2949                                                   &p->guid,
2950                                                   false, schema_attr,
2951                                                   parent);
2952                         if (ret != LDB_SUCCESS) {
2953                                 talloc_free(tmp_ctx);
2954                                 return ret;
2955                         }
2956
2957                         /* We flag the deletion and tidy it up later. */
2958                         exact->v = NULL;
2959                         continue;
2960                 }
2961
2962                 rmd_flags = dsdb_dn_rmd_flags(exact->dsdb_dn->dn);
2963
2964                 if (rmd_flags & DSDB_RMD_FLAG_DELETED) {
2965                         struct GUID_txt_buf buf;
2966                         const char *guid_str = GUID_buf_string(&p->guid, &buf);
2967                         ldb_asprintf_errstring(ldb, "Attribute %s already "
2968                                                "deleted for target GUID %s",
2969                                                el->name, guid_str);
2970                         if (ldb_attr_cmp(el->name, "member") == 0) {
2971                                 talloc_free(tmp_ctx);
2972                                 return LDB_ERR_UNWILLING_TO_PERFORM;
2973                         } else {
2974                                 talloc_free(tmp_ctx);
2975                                 return LDB_ERR_NO_SUCH_ATTRIBUTE;
2976                         }
2977                 }
2978
2979                 ret = replmd_update_la_val(old_el->values, exact->v,
2980                                            exact->dsdb_dn, exact->dsdb_dn,
2981                                            &ac->our_invocation_id,
2982                                            ac->seq_num, ac->seq_num,
2983                                            now, true);
2984                 if (ret != LDB_SUCCESS) {
2985                         talloc_free(tmp_ctx);
2986                         return ret;
2987                 }
2988                 ret = replmd_add_backlink(module, replmd_private,
2989                                           ac->schema, msg_dn,
2990                                           &p->guid,
2991                                           false, schema_attr,
2992                                           parent);
2993                 if (ret != LDB_SUCCESS) {
2994                         talloc_free(tmp_ctx);
2995                         return ret;
2996                 }
2997         }
2998
2999         if (vanish_links) {
3000                 unsigned j = 0;
3001                 struct ldb_val *tmp_vals = NULL;
3002
3003                 tmp_vals = talloc_array(tmp_ctx, struct ldb_val,
3004                                         old_el->num_values);
3005                 if (tmp_vals == NULL) {
3006                         talloc_free(tmp_ctx);
3007                         return ldb_module_oom(module);
3008                 }
3009                 for (i = 0; i < old_el->num_values; i++) {
3010                         if (old_dns[i].v == NULL) {
3011                                 continue;
3012                         }
3013                         tmp_vals[j] = *old_dns[i].v;
3014                         j++;
3015                 }
3016                 for (i = 0; i < j; i++) {
3017                         old_el->values[i] = tmp_vals[i];
3018                 }
3019                 old_el->num_values = j;
3020         }
3021
3022         el->values = talloc_steal(msg->elements, old_el->values);
3023         el->num_values = old_el->num_values;
3024
3025         talloc_free(tmp_ctx);
3026
3027         /* we now tell the backend to replace all existing values
3028            with the one we have constructed */
3029         el->flags = LDB_FLAG_MOD_REPLACE;
3030
3031         return LDB_SUCCESS;
3032 }
3033
3034 /*
3035   handle replacing a linked attribute
3036  */
3037 static int replmd_modify_la_replace(struct ldb_module *module,
3038                                     struct replmd_private *replmd_private,
3039                                     struct replmd_replicated_request *ac,
3040                                     struct ldb_message *msg,
3041                                     struct ldb_message_element *el,
3042                                     struct ldb_message_element *old_el,
3043                                     const struct dsdb_attribute *schema_attr,
3044                                     time_t t,
3045                                     struct ldb_dn *msg_dn,
3046                                     struct ldb_request *parent)
3047 {
3048         unsigned int i, old_i, new_i;
3049         struct parsed_dn *dns, *old_dns;
3050         TALLOC_CTX *tmp_ctx = talloc_new(msg);
3051         int ret;
3052         struct ldb_context *ldb = ldb_module_get_ctx(module);
3053         struct ldb_val *new_values = NULL;
3054         const char *ldap_oid = schema_attr->syntax->ldap_oid;
3055         unsigned int old_num_values;
3056         unsigned int repl_num_values;
3057         unsigned int max_num_values;
3058         NTTIME now;
3059
3060         unix_to_nt_time(&now, t);
3061
3062         /*
3063          * The replace operation is unlike the replace and delete cases in that
3064          * we need to look at every existing link to see whether it is being
3065          * retained or deleted. In other words, we can't avoid parsing the GUIDs.
3066          *
3067          * As we are trying to combine two sorted lists, the algorithm we use
3068          * is akin to the merge phase of a merge sort. We interleave the two
3069          * lists, doing different things depending on which side the current
3070          * item came from.
3071          *
3072          * There are three main cases, with some sub-cases.
3073          *
3074          *  - a DN is in the old list but not the new one. It needs to be
3075          *    marked as deleted (but left in the list).
3076          *     - maybe it is already deleted, and we have less to do.
3077          *
3078          *  - a DN is in both lists. The old data gets replaced by the new,
3079          *    and the list doesn't grow. The old link may have been marked as
3080          *    deleted, in which case we undelete it.
3081          *
3082          *  - a DN is in the new list only. We add it in the right place.
3083          */
3084
3085         old_num_values = old_el ? old_el->num_values : 0;
3086         repl_num_values = el->num_values;
3087         max_num_values = old_num_values + repl_num_values;
3088
3089         if (max_num_values == 0) {
3090                 /* There is nothing to do! */
3091                 return LDB_SUCCESS;
3092         }
3093
3094         /*
3095          * At the successful end of these functions el->values is
3096          * overwritten with new_values.  However get_parsed_dns()
3097          * points p->v at the supplied el and it effectively gets used
3098          * as a working area by replmd_build_la_val().  So we must
3099          * duplicate it because our caller only called
3100          * ldb_msg_copy_shallow().
3101          */
3102
3103         el->values = talloc_memdup(tmp_ctx,
3104                                    el->values,
3105                                    sizeof(el->values[0]) * el->num_values);
3106         if (el->values == NULL) {
3107                 ldb_module_oom(module);
3108                 talloc_free(tmp_ctx);
3109                 return LDB_ERR_OPERATIONS_ERROR;
3110         }
3111
3112         ret = get_parsed_dns(module, tmp_ctx, el, &dns, ldap_oid, parent);
3113         if (ret != LDB_SUCCESS) {
3114                 talloc_free(tmp_ctx);
3115                 return ret;
3116         }
3117
3118         ret = check_parsed_dn_duplicates(module, el, dns);
3119         if (ret != LDB_SUCCESS) {
3120                 talloc_free(tmp_ctx);
3121                 return ret;
3122         }
3123
3124         ret = get_parsed_dns(module, tmp_ctx, old_el, &old_dns,
3125                              ldap_oid, parent);
3126         if (ret != LDB_SUCCESS) {
3127                 talloc_free(tmp_ctx);
3128                 return ret;
3129         }
3130
3131         ret = replmd_check_upgrade_links(ldb, old_dns, old_num_values,
3132                                          old_el, ldap_oid);
3133         if (ret != LDB_SUCCESS) {
3134                 talloc_free(tmp_ctx);
3135                 return ret;
3136         }
3137
3138         new_values = talloc_array(tmp_ctx, struct ldb_val, max_num_values);
3139         if (new_values == NULL) {
3140                 ldb_module_oom(module);
3141                 talloc_free(tmp_ctx);
3142                 return LDB_ERR_OPERATIONS_ERROR;
3143         }
3144
3145         old_i = 0;
3146         new_i = 0;
3147         for (i = 0; i < max_num_values; i++) {
3148                 int cmp;
3149                 struct parsed_dn *old_p, *new_p;
3150                 if (old_i < old_num_values && new_i < repl_num_values) {
3151                         old_p = &old_dns[old_i];
3152                         new_p = &dns[new_i];
3153                         cmp = parsed_dn_compare(old_p, new_p);
3154                 } else if (old_i < old_num_values) {
3155                         /* the new list is empty, read the old list */
3156                         old_p = &old_dns[old_i];
3157                         new_p = NULL;
3158                         cmp = -1;
3159                 } else if (new_i < repl_num_values) {
3160                         /* the old list is empty, read new list */
3161                         old_p = NULL;
3162                         new_p = &dns[new_i];
3163                         cmp = 1;
3164                 } else {
3165                         break;
3166                 }
3167
3168                 if (cmp < 0) {
3169                         /*
3170                          * An old ones that come before the next replacement
3171                          * (if any). We mark it as deleted and add it to the
3172                          * final list.
3173                          */
3174                         uint32_t rmd_flags = dsdb_dn_rmd_flags(old_p->dsdb_dn->dn);
3175                         if ((rmd_flags & DSDB_RMD_FLAG_DELETED) == 0) {
3176                                 ret = replmd_update_la_val(new_values, old_p->v,
3177                                                            old_p->dsdb_dn,
3178                                                            old_p->dsdb_dn,
3179                                                            &ac->our_invocation_id,
3180                                                            ac->seq_num, ac->seq_num,
3181                                                            now, true);
3182                                 if (ret != LDB_SUCCESS) {
3183                                         talloc_free(tmp_ctx);
3184                                         return ret;
3185                                 }
3186
3187                                 ret = replmd_add_backlink(module, replmd_private,
3188                                                           ac->schema,
3189                                                           msg_dn,
3190                                                           &old_p->guid, false,
3191                                                           schema_attr,
3192                                                           parent);
3193                                 if (ret != LDB_SUCCESS) {
3194                                         talloc_free(tmp_ctx);
3195                                         return ret;
3196                                 }
3197                         }
3198                         new_values[i] = *old_p->v;
3199                         old_i++;
3200                 } else if (cmp == 0) {
3201                         /*
3202                          * We are overwriting one. If it was previously
3203                          * deleted, we need to add a backlink.
3204                          *
3205                          * Note that if any RMD_FLAGs in an extended new DN
3206                          * will be ignored.
3207                          */
3208                         uint32_t rmd_flags;
3209
3210                         ret = replmd_update_la_val(new_values, old_p->v,
3211                                                    new_p->dsdb_dn,
3212                                                    old_p->dsdb_dn,
3213                                                    &ac->our_invocation_id,
3214                                                    ac->seq_num, ac->seq_num,
3215                                                    now, false);
3216                         if (ret != LDB_SUCCESS) {
3217                                 talloc_free(tmp_ctx);
3218                                 return ret;
3219                         }
3220
3221                         rmd_flags = dsdb_dn_rmd_flags(old_p->dsdb_dn->dn);
3222                         if ((rmd_flags & DSDB_RMD_FLAG_DELETED) != 0) {
3223                                 ret = replmd_add_backlink(module, replmd_private,
3224                                                           ac->schema,
3225                                                           msg_dn,
3226                                                           &new_p->guid, true,
3227                                                           schema_attr,
3228                                                           parent);
3229                                 if (ret != LDB_SUCCESS) {
3230                                         talloc_free(tmp_ctx);
3231                                         return ret;
3232                                 }
3233                         }
3234
3235                         new_values[i] = *old_p->v;
3236                         old_i++;
3237                         new_i++;
3238                 } else {
3239                         /*
3240                          * Replacements that don't match an existing one. We
3241                          * just add them to the final list.
3242                          */
3243                         ret = replmd_build_la_val(new_values,
3244                                                   new_p->v,
3245                                                   new_p->dsdb_dn,
3246                                                   &ac->our_invocation_id,
3247                                                   ac->seq_num, now);
3248                         if (ret != LDB_SUCCESS) {
3249                                 talloc_free(tmp_ctx);
3250                                 return ret;
3251                         }
3252                         ret = replmd_add_backlink(module, replmd_private,
3253                                                   ac->schema,
3254                                                   msg_dn,
3255                                                   &new_p->guid, true,
3256                                                   schema_attr,
3257                                                   parent);
3258                         if (ret != LDB_SUCCESS) {
3259                                 talloc_free(tmp_ctx);
3260                                 return ret;
3261                         }
3262                         new_values[i] = *new_p->v;
3263                         new_i++;
3264                 }
3265         }
3266         if (old_el != NULL) {
3267                 talloc_steal(msg->elements, old_el->values);
3268         }
3269         el->values = talloc_steal(msg->elements, new_values);
3270         el->num_values = i;
3271         talloc_free(tmp_ctx);
3272
3273         el->flags = LDB_FLAG_MOD_REPLACE;
3274
3275         return LDB_SUCCESS;
3276 }
3277
3278
3279 /*
3280   handle linked attributes in modify requests
3281  */
3282 static int replmd_modify_handle_linked_attribs(struct ldb_module *module,
3283                                                struct replmd_private *replmd_private,
3284                                                struct replmd_replicated_request *ac,
3285                                                struct ldb_message *msg,
3286                                                time_t t,
3287                                                struct ldb_request *parent)
3288 {
3289         struct ldb_result *res;
3290         unsigned int i;
3291         int ret;
3292         struct ldb_context *ldb = ldb_module_get_ctx(module);
3293         struct ldb_message *old_msg;
3294
3295         if (dsdb_functional_level(ldb) == DS_DOMAIN_FUNCTION_2000) {
3296                 /*
3297                  * Nothing special is required for modifying or vanishing links
3298                  * in fl2000 since they are just strings in a multi-valued
3299                  * attribute.
3300                  */
3301                 struct ldb_control *ctrl = ldb_request_get_control(parent,
3302                                                                    DSDB_CONTROL_REPLMD_VANISH_LINKS);
3303                 if (ctrl) {
3304                         ctrl->critical = false;
3305                 }
3306                 return LDB_SUCCESS;
3307         }
3308
3309         /*
3310          * TODO:
3311          *
3312          * We should restrict this to the intersection of the list of
3313          * linked attributes in the schema and the list of attributes
3314          * being modified.
3315          *
3316          * This will help performance a little, as otherwise we have
3317          * to allocate the entire object value-by-value.
3318          */
3319         ret = dsdb_module_search_dn(module, msg, &res, msg->dn, NULL,
3320                                     DSDB_FLAG_NEXT_MODULE |
3321                                     DSDB_SEARCH_SHOW_RECYCLED |
3322                                     DSDB_SEARCH_REVEAL_INTERNALS |
3323                                     DSDB_SEARCH_SHOW_DN_IN_STORAGE_FORMAT,
3324                                     parent);
3325         if (ret != LDB_SUCCESS) {
3326                 return ret;
3327         }
3328
3329         old_msg = res->msgs[0];
3330
3331         for (i=0; i<msg->num_elements; i++) {
3332                 struct ldb_message_element *el = &msg->elements[i];
3333                 struct ldb_message_element *old_el, *new_el;
3334                 unsigned int mod_type = LDB_FLAG_MOD_TYPE(el->flags);
3335                 const struct dsdb_attribute *schema_attr
3336                         = dsdb_attribute_by_lDAPDisplayName(ac->schema, el->name);
3337                 if (!schema_attr) {
3338                         ldb_asprintf_errstring(ldb,
3339                                                "%s: attribute %s is not a valid attribute in schema",
3340                                                __FUNCTION__, el->name);
3341                         return LDB_ERR_OBJECT_CLASS_VIOLATION;
3342                 }
3343                 if (schema_attr->linkID == 0) {
3344                         continue;
3345                 }
3346                 if ((schema_attr->linkID & 1) == 1) {
3347                         struct ldb_control *ctrl;
3348
3349                         ctrl = ldb_request_get_control(parent,
3350                                                        DSDB_CONTROL_REPLMD_VANISH_LINKS);
3351                         if (ctrl != NULL) {
3352                                 ctrl->critical = false;
3353                                 continue;
3354                         }
3355                         ctrl = ldb_request_get_control(parent,
3356                                                        DSDB_CONTROL_DBCHECK);
3357                         if (ctrl != NULL) {
3358                                 continue;
3359                         }
3360
3361                         /* Odd is for the target.  Illegal to modify */
3362                         ldb_asprintf_errstring(ldb,
3363                                                "attribute %s must not be modified directly, it is a linked attribute", el->name);
3364                         return LDB_ERR_UNWILLING_TO_PERFORM;
3365                 }
3366                 old_el = ldb_msg_find_element(old_msg, el->name);
3367                 switch (mod_type) {
3368                 case LDB_FLAG_MOD_REPLACE:
3369                         ret = replmd_modify_la_replace(module, replmd_private,
3370                                                        ac, msg, el, old_el,
3371                                                        schema_attr, t,
3372                                                        old_msg->dn,
3373                                                        parent);
3374                         break;
3375                 case LDB_FLAG_MOD_DELETE:
3376                         ret = replmd_modify_la_delete(module, replmd_private,
3377                                                       ac, msg, el, old_el,
3378                                                       schema_attr, t,
3379                                                       old_msg->dn,
3380                                                       parent);
3381                         break;
3382                 case LDB_FLAG_MOD_ADD:
3383                         ret = replmd_modify_la_add(module, replmd_private,
3384                                                    ac, msg, el, old_el,
3385                                                    schema_attr, t,
3386                                                    old_msg->dn,
3387                                                    parent);
3388                         break;
3389                 default:
3390                         ldb_asprintf_errstring(ldb,
3391                                                "invalid flags 0x%x for %s linked attribute",
3392                                                el->flags, el->name);
3393                         return LDB_ERR_UNWILLING_TO_PERFORM;
3394                 }
3395                 if (ret != LDB_SUCCESS) {
3396                         return ret;
3397                 }
3398                 ret = dsdb_check_single_valued_link(schema_attr, el);
3399                 if (ret != LDB_SUCCESS) {
3400                         ldb_asprintf_errstring(ldb,
3401                                                "Attribute %s is single valued but more than one value has been supplied",
3402                                                el->name);
3403                         /* Return codes as found on Windows 2012r2 */
3404                         if (mod_type == LDB_FLAG_MOD_REPLACE) {
3405                                 return LDB_ERR_CONSTRAINT_VIOLATION;
3406                         } else {
3407                                 return LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
3408                         }
3409                 } else {
3410                         el->flags |= LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK;
3411                 }
3412
3413                 if (old_el) {
3414                         ldb_msg_remove_attr(old_msg, el->name);
3415                 }
3416                 ldb_msg_add_empty(old_msg, el->name, 0, &new_el);
3417                 new_el->num_values = el->num_values;
3418                 new_el->values = talloc_steal(msg->elements, el->values);
3419
3420                 /* TODO: this relises a bit too heavily on the exact
3421                    behaviour of ldb_msg_find_element and
3422                    ldb_msg_remove_element */
3423                 old_el = ldb_msg_find_element(msg, el->name);
3424                 if (old_el != el) {
3425                         ldb_msg_remove_element(msg, old_el);
3426                         i--;
3427                 }
3428         }
3429
3430         talloc_free(res);
3431         return ret;
3432 }
3433
3434
3435 static int send_rodc_referral(struct ldb_request *req,
3436                               struct ldb_context *ldb,
3437                               struct ldb_dn *dn)
3438 {
3439         char *referral = NULL;
3440         struct loadparm_context *lp_ctx = NULL;
3441         struct ldb_dn *fsmo_role_dn = NULL;
3442         struct ldb_dn *role_owner_dn = NULL;
3443         const char *domain = NULL;
3444         WERROR werr;
3445
3446         lp_ctx = talloc_get_type(ldb_get_opaque(ldb, "loadparm"),
3447                                  struct loadparm_context);
3448
3449         werr = dsdb_get_fsmo_role_info(req, ldb, DREPL_PDC_MASTER,
3450                                        &fsmo_role_dn, &role_owner_dn);
3451
3452         if (W_ERROR_IS_OK(werr)) {
3453                 struct ldb_dn *server_dn = ldb_dn_copy(req, role_owner_dn);
3454                 if (server_dn != NULL) {
3455                         ldb_dn_remove_child_components(server_dn, 1);
3456                         domain = samdb_dn_to_dnshostname(ldb, req,
3457                                                          server_dn);
3458                 }
3459         }
3460
3461         if (domain == NULL) {
3462                 domain = lpcfg_dnsdomain(lp_ctx);
3463         }
3464
3465         referral = talloc_asprintf(req, "ldap://%s/%s",
3466                                    domain,
3467                                    ldb_dn_get_linearized(dn));
3468         if (referral == NULL) {
3469                 ldb_oom(ldb);
3470                 return LDB_ERR_OPERATIONS_ERROR;
3471         }
3472
3473         return ldb_module_send_referral(req, referral);
3474 }
3475
3476
3477 static int replmd_modify(struct ldb_module *module, struct ldb_request *req)
3478 {
3479         struct ldb_context *ldb;
3480         struct replmd_replicated_request *ac;
3481         struct ldb_request *down_req;
3482         struct ldb_message *msg;
3483         time_t t = time(NULL);
3484         int ret;
3485         bool is_urgent = false, rodc = false;
3486         bool is_schema_nc = false;
3487         unsigned int functional_level;
3488         const struct ldb_message_element *guid_el = NULL;
3489         struct ldb_control *sd_propagation_control;
3490         struct ldb_control *fix_links_control = NULL;
3491         struct ldb_control *fix_dn_name_control = NULL;
3492         struct ldb_control *fix_dn_sid_control = NULL;
3493         struct replmd_private *replmd_private =
3494                 talloc_get_type(ldb_module_get_private(module), struct replmd_private);
3495
3496         /* do not manipulate our control entries */
3497         if (ldb_dn_is_special(req->op.mod.message->dn)) {
3498                 return ldb_next_request(module, req);
3499         }
3500
3501         sd_propagation_control = ldb_request_get_control(req,
3502                                         DSDB_CONTROL_SEC_DESC_PROPAGATION_OID);
3503         if (sd_propagation_control != NULL) {
3504                 if (req->op.mod.message->num_elements != 1) {
3505                         return ldb_module_operr(module);
3506                 }
3507                 ret = strcmp(req->op.mod.message->elements[0].name,
3508                              "nTSecurityDescriptor");
3509                 if (ret != 0) {
3510                         return ldb_module_operr(module);
3511                 }
3512
3513                 return ldb_next_request(module, req);
3514         }
3515
3516         ldb = ldb_module_get_ctx(module);
3517
3518         fix_links_control = ldb_request_get_control(req,
3519                                         DSDB_CONTROL_DBCHECK_FIX_DUPLICATE_LINKS);
3520         if (fix_links_control != NULL) {
3521                 struct dsdb_schema *schema = NULL;
3522                 const struct dsdb_attribute *sa = NULL;
3523
3524                 if (req->op.mod.message->num_elements != 1) {
3525                         return ldb_module_operr(module);
3526                 }
3527
3528                 if (LDB_FLAG_MOD_TYPE(req->op.mod.message->elements[0].flags) != LDB_FLAG_MOD_REPLACE) {
3529                         return ldb_module_operr(module);
3530                 }
3531
3532                 schema = dsdb_get_schema(ldb, req);
3533                 if (schema == NULL) {
3534                         return ldb_module_operr(module);
3535                 }
3536
3537                 sa = dsdb_attribute_by_lDAPDisplayName(schema,
3538                                 req->op.mod.message->elements[0].name);
3539                 if (sa == NULL) {
3540                         return ldb_module_operr(module);
3541                 }
3542
3543                 if (sa->linkID == 0) {
3544                         return ldb_module_operr(module);
3545                 }
3546
3547                 fix_links_control->critical = false;
3548                 return ldb_next_request(module, req);
3549         }
3550
3551         fix_dn_name_control = ldb_request_get_control(req,
3552                                         DSDB_CONTROL_DBCHECK_FIX_LINK_DN_NAME);
3553         if (fix_dn_name_control != NULL) {
3554                 struct dsdb_schema *schema = NULL;
3555                 const struct dsdb_attribute *sa = NULL;
3556
3557                 if (req->op.mod.message->num_elements != 2) {
3558                         return ldb_module_operr(module);
3559                 }
3560
3561                 if (LDB_FLAG_MOD_TYPE(req->op.mod.message->elements[0].flags) != LDB_FLAG_MOD_DELETE) {
3562                         return ldb_module_operr(module);
3563                 }
3564
3565                 if (LDB_FLAG_MOD_TYPE(req->op.mod.message->elements[1].flags) != LDB_FLAG_MOD_ADD) {
3566                         return ldb_module_operr(module);
3567                 }
3568
3569                 if (req->op.mod.message->elements[0].num_values != 1) {
3570                         return ldb_module_operr(module);
3571                 }
3572
3573                 if (req->op.mod.message->elements[1].num_values != 1) {
3574                         return ldb_module_operr(module);
3575                 }
3576
3577                 schema = dsdb_get_schema(ldb, req);
3578                 if (schema == NULL) {
3579                         return ldb_module_operr(module);
3580                 }
3581
3582                 if (ldb_attr_cmp(req->op.mod.message->elements[0].name,
3583                                  req->op.mod.message->elements[1].name) != 0) {
3584                         return ldb_module_operr(module);
3585                 }
3586
3587                 sa = dsdb_attribute_by_lDAPDisplayName(schema,
3588                                 req->op.mod.message->elements[0].name);
3589                 if (sa == NULL) {
3590                         return ldb_module_operr(module);
3591                 }
3592
3593                 if (sa->dn_format == DSDB_INVALID_DN) {
3594                         return ldb_module_operr(module);
3595                 }
3596
3597                 if (sa->linkID != 0) {
3598                         return ldb_module_operr(module);
3599                 }
3600
3601                 /*
3602                  * If we are run from dbcheck and we are not updating
3603                  * a link (as these would need to be sorted and so
3604                  * can't go via such a simple update, then do not
3605                  * trigger replicated updates and a new USN from this
3606                  * change, it wasn't a real change, just a new
3607                  * (correct) string DN
3608                  */
3609
3610                 fix_dn_name_control->critical = false;
3611                 return ldb_next_request(module, req);
3612         }
3613
3614         ldb_debug(ldb, LDB_DEBUG_TRACE, "replmd_modify\n");
3615
3616         guid_el = ldb_msg_find_element(req->op.mod.message, "objectGUID");
3617         if (guid_el != NULL) {
3618                 ldb_set_errstring(ldb,
3619                                   "replmd_modify: it's not allowed to change the objectGUID!");
3620                 return LDB_ERR_CONSTRAINT_VIOLATION;
3621         }
3622
3623         ac = replmd_ctx_init(module, req);
3624         if (ac == NULL) {
3625                 return ldb_module_oom(module);
3626         }
3627
3628         functional_level = dsdb_functional_level(ldb);
3629
3630         /* we have to copy the message as the caller might have it as a const */
3631         msg = ldb_msg_copy_shallow(ac, req->op.mod.message);
3632         if (msg == NULL) {
3633                 ldb_oom(ldb);
3634                 talloc_free(ac);
3635                 return LDB_ERR_OPERATIONS_ERROR;
3636         }
3637
3638         fix_dn_sid_control = ldb_request_get_control(req,
3639                                         DSDB_CONTROL_DBCHECK_FIX_LINK_DN_SID);
3640         if (fix_dn_sid_control != NULL) {
3641                 const struct dsdb_attribute *sa = NULL;
3642
3643                 if (msg->num_elements != 1) {
3644                         talloc_free(ac);
3645                         return ldb_module_operr(module);
3646                 }
3647
3648                 if (LDB_FLAG_MOD_TYPE(msg->elements[0].flags) != LDB_FLAG_MOD_ADD) {
3649                         talloc_free(ac);
3650                         return ldb_module_operr(module);
3651                 }
3652
3653                 if (msg->elements[0].num_values != 1) {
3654                         talloc_free(ac);
3655                         return ldb_module_operr(module);
3656                 }
3657
3658                 sa = dsdb_attribute_by_lDAPDisplayName(ac->schema,
3659                                 msg->elements[0].name);
3660                 if (sa == NULL) {
3661                         talloc_free(ac);
3662                         return ldb_module_operr(module);
3663                 }
3664
3665                 if (sa->dn_format != DSDB_NORMAL_DN) {
3666                         talloc_free(ac);
3667                         return ldb_module_operr(module);
3668                 }
3669
3670                 fix_dn_sid_control->critical = false;
3671                 ac->fix_link_sid = true;
3672
3673                 goto handle_linked_attribs;
3674         }
3675
3676         ldb_msg_remove_attr(msg, "whenChanged");
3677         ldb_msg_remove_attr(msg, "uSNChanged");
3678
3679         is_schema_nc = ldb_dn_compare_base(replmd_private->schema_dn, msg->dn) == 0;
3680
3681         ret = replmd_update_rpmd(module, ac->schema, req, NULL,
3682                                  msg, &ac->seq_num, t, is_schema_nc,
3683                                  &is_urgent, &rodc);
3684         if (rodc && (ret == LDB_ERR_REFERRAL)) {
3685                 ret = send_rodc_referral(req, ldb, msg->dn);
3686                 talloc_free(ac);
3687                 return ret;
3688
3689         }
3690
3691         if (ret != LDB_SUCCESS) {
3692                 talloc_free(ac);
3693                 return ret;
3694         }
3695
3696  handle_linked_attribs:
3697         ret = replmd_modify_handle_linked_attribs(module, replmd_private,
3698                                                   ac, msg, t, req);
3699         if (ret != LDB_SUCCESS) {
3700                 talloc_free(ac);
3701                 return ret;
3702         }
3703
3704         /* TODO:
3705          * - replace the old object with the newly constructed one
3706          */
3707
3708         ac->is_urgent = is_urgent;
3709
3710         ret = ldb_build_mod_req(&down_req, ldb, ac,
3711                                 msg,
3712                                 req->controls,
3713                                 ac, replmd_op_callback,
3714                                 req);
3715         LDB_REQ_SET_LOCATION(down_req);
3716         if (ret != LDB_SUCCESS) {
3717                 talloc_free(ac);
3718                 return ret;
3719         }
3720
3721         /* current partition control is needed by "replmd_op_callback" */
3722         if (ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID) == NULL) {
3723                 ret = ldb_request_add_control(down_req,
3724                                               DSDB_CONTROL_CURRENT_PARTITION_OID,
3725                                               false, NULL);
3726                 if (ret != LDB_SUCCESS) {
3727                         talloc_free(ac);
3728                         return ret;
3729                 }
3730         }
3731
3732         /* If we are in functional level 2000, then
3733          * replmd_modify_handle_linked_attribs will have done
3734          * nothing */
3735         if (functional_level == DS_DOMAIN_FUNCTION_2000) {
3736                 ret = ldb_request_add_control(down_req, DSDB_CONTROL_APPLY_LINKS, false, NULL);
3737                 if (ret != LDB_SUCCESS) {
3738                         talloc_free(ac);
3739                         return ret;
3740                 }
3741         }
3742
3743         talloc_steal(down_req, msg);
3744
3745         /* we only change whenChanged and uSNChanged if the seq_num
3746            has changed */
3747         if (ac->seq_num != 0) {
3748                 ret = add_time_element(msg, "whenChanged", t);
3749                 if (ret != LDB_SUCCESS) {
3750                         talloc_free(ac);
3751                         ldb_operr(ldb);
3752                         return ret;
3753                 }
3754
3755                 ret = add_uint64_element(ldb, msg, "uSNChanged", ac->seq_num);
3756                 if (ret != LDB_SUCCESS) {
3757                         talloc_free(ac);
3758                         ldb_operr(ldb);
3759                         return ret;
3760                 }
3761         }
3762
3763         /* go on with the call chain */
3764         return ldb_next_request(module, down_req);
3765 }
3766
3767 static int replmd_rename_callback(struct ldb_request *req, struct ldb_reply *ares);
3768
3769 /*
3770   handle a rename request
3771
3772   On a rename we need to do an extra ldb_modify which sets the
3773   whenChanged and uSNChanged attributes.  We do this in a callback after the success.
3774  */
3775 static int replmd_rename(struct ldb_module *module, struct ldb_request *req)
3776 {
3777         struct ldb_context *ldb;
3778         struct ldb_control *fix_dn_name_control = NULL;
3779         struct replmd_replicated_request *ac;
3780         int ret;
3781         struct ldb_request *down_req;
3782
3783         /* do not manipulate our control entries */
3784         if (ldb_dn_is_special(req->op.rename.olddn)) {
3785                 return ldb_next_request(module, req);
3786         }
3787
3788         fix_dn_name_control = ldb_request_get_control(req,
3789                                         DSDB_CONTROL_DBCHECK_FIX_LINK_DN_NAME);
3790         if (fix_dn_name_control != NULL) {
3791                 return ldb_next_request(module, req);
3792         }
3793
3794         ldb = ldb_module_get_ctx(module);
3795
3796         ldb_debug(ldb, LDB_DEBUG_TRACE, "replmd_rename\n");
3797
3798         ac = replmd_ctx_init(module, req);
3799         if (ac == NULL) {
3800                 return ldb_module_oom(module);
3801         }
3802
3803         ret = ldb_build_rename_req(&down_