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