source/dsdb/samdb/ldb_modules/repl_meta_data.c

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