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