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         NTSTATUS nt_status;
 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         nt_status = ndr_push_struct_blob(&guid_value, msg, &guid,
 522                                          (ndr_push_flags_fn_t)ndr_push_GUID);
 523         if (!NT_STATUS_IS_OK(nt_status)) {
 524                 talloc_free(down_req);
 525                 ldb_oom(module->ldb);
 526                 return LDB_ERR_OPERATIONS_ERROR;
 527         }
 528         nt_status = ndr_push_struct_blob(&nmd_value, msg, &nmd,
 529                                          (ndr_push_flags_fn_t)ndr_push_replPropertyMetaDataBlob);
 530         if (!NT_STATUS_IS_OK(nt_status)) {
 531                 talloc_free(down_req);
 532                 ldb_oom(module->ldb);
 533                 return LDB_ERR_OPERATIONS_ERROR;
 534         }
 535
 536         /*
 537          * add the autogenerated values
 538          */
 539         ret = ldb_msg_add_value(msg, "objectGUID", &guid_value, NULL);
 540         if (ret != LDB_SUCCESS) {
 541                 talloc_free(down_req);
 542                 ldb_oom(module->ldb);
 543                 return LDB_ERR_OPERATIONS_ERROR;
 544         }
 545         ret = ldb_msg_add_string(msg, "whenChanged", time_str);
 546         if (ret != LDB_SUCCESS) {
 547                 talloc_free(down_req);
 548                 ldb_oom(module->ldb);
 549                 return LDB_ERR_OPERATIONS_ERROR;
 550         }
 551         ret = samdb_msg_add_uint64(module->ldb, msg, msg, "uSNCreated", seq_num);
 552         if (ret != LDB_SUCCESS) {
 553                 talloc_free(down_req);
 554                 ldb_oom(module->ldb);
 555                 return LDB_ERR_OPERATIONS_ERROR;
 556         }
 557         ret = samdb_msg_add_uint64(module->ldb, msg, msg, "uSNChanged", seq_num);
 558         if (ret != LDB_SUCCESS) {
 559                 talloc_free(down_req);
 560                 ldb_oom(module->ldb);
 561                 return LDB_ERR_OPERATIONS_ERROR;
 562         }
 563         ret = ldb_msg_add_value(msg, "replPropertyMetaData", &nmd_value, NULL);
 564         if (ret != LDB_SUCCESS) {
 565                 talloc_free(down_req);
 566                 ldb_oom(module->ldb);
 567                 return LDB_ERR_OPERATIONS_ERROR;
 568         }
 569
 570         /*
 571          * sort the attributes by attid before storing the object
 572          */
 573         replmd_ldb_message_sort(msg, schema);
 574
 575         ldb_set_timeout_from_prev_req(module->ldb, req, down_req);
 576
 577         /* go on with the call chain */
 578         ret = ldb_next_request(module, down_req);
 579
 580         /* do not free down_req as the call results may be linked to it,
 581          * it will be freed when the upper level request get freed */
 582         if (ret == LDB_SUCCESS) {
 583                 req->handle = down_req->handle;
 584         }
 585
 586         return ret;
 587 }
 588
 589 static int replmd_add(struct ldb_module *module, struct ldb_request *req)
 590 {
 591         return replmd_prepare_originating(module, req, req->op.add.message->dn,
 592                                           "replmd_add", replmd_add_originating);
 593 }
 594
 595 static int replmd_modify_originating(struct ldb_module *module,
 596                                      struct ldb_request *req,
 597                                      const struct dsdb_schema *schema,
 598                                      const struct dsdb_control_current_partition *partition)
 599 {
 600         struct ldb_request *down_req;
 601         struct ldb_message *msg;
 602         int ret;
 603         time_t t = time(NULL);
 604         uint64_t seq_num;
 605
 606         ldb_debug(module->ldb, LDB_DEBUG_TRACE, "replmd_modify_originating\n");
 607
 608         down_req = talloc(req, struct ldb_request);
 609         if (down_req == NULL) {
 610                 return LDB_ERR_OPERATIONS_ERROR;
 611         }
 612
 613         *down_req = *req;
 614
 615         /* we have to copy the message as the caller might have it as a const */
 616         down_req->op.mod.message = msg = ldb_msg_copy_shallow(down_req, req->op.mod.message);
 617         if (msg == NULL) {
 618                 talloc_free(down_req);
 619                 return LDB_ERR_OPERATIONS_ERROR;
 620         }
 621
 622         if (add_time_element(msg, "whenChanged", t) != 0) {
 623                 talloc_free(down_req);
 624                 return LDB_ERR_OPERATIONS_ERROR;
 625         }
 626
 627         /* Get a sequence number from the backend */
 628         ret = ldb_sequence_number(module->ldb, LDB_SEQ_NEXT, &seq_num);
 629         if (ret == LDB_SUCCESS) {
 630                 if (add_uint64_element(msg, "uSNChanged", seq_num) != 0) {
 631                         talloc_free(down_req);
 632                         return LDB_ERR_OPERATIONS_ERROR;
 633                 }
 634         }
 635
 636         ldb_set_timeout_from_prev_req(module->ldb, req, down_req);
 637
 638         /* go on with the call chain */
 639         ret = ldb_next_request(module, down_req);
 640
 641         /* do not free down_req as the call results may be linked to it,
 642          * it will be freed when the upper level request get freed */
 643         if (ret == LDB_SUCCESS) {
 644                 req->handle = down_req->handle;
 645         }
 646
 647         return ret;
 648 }
 649
 650 static int replmd_modify(struct ldb_module *module, struct ldb_request *req)
 651 {
 652         return replmd_prepare_originating(module, req, req->op.mod.message->dn,
 653                                           "replmd_modify", replmd_modify_originating);
 654 }
 655
 656 static int replmd_replicated_request_reply_helper(struct replmd_replicated_request *ar, int ret)
 657 {
 658         struct ldb_reply *ares = NULL;
 659
 660         ar->handle->status = ret;
 661         ar->handle->state = LDB_ASYNC_DONE;
 662
 663         if (!ar->orig_req->callback) {
 664                 return LDB_SUCCESS;
 665         }
 666
 667         /* we're done and need to report the success to the caller */
 668         ares = talloc_zero(ar, struct ldb_reply);
 669         if (!ares) {
 670                 ar->handle->status = LDB_ERR_OPERATIONS_ERROR;
 671                 ar->handle->state = LDB_ASYNC_DONE;
 672                 return LDB_ERR_OPERATIONS_ERROR;
 673         }
 674
 675         ares->type      = LDB_REPLY_EXTENDED;
 676         ares->response  = NULL;
 677
 678         return ar->orig_req->callback(ar->module->ldb, ar->orig_req->context, ares);
 679 }
 680
 681 static int replmd_replicated_request_done(struct replmd_replicated_request *ar)
 682 {
 683         return replmd_replicated_request_reply_helper(ar, LDB_SUCCESS);
 684 }
 685
 686 static int replmd_replicated_request_error(struct replmd_replicated_request *ar, int ret)
 687 {
 688         return replmd_replicated_request_reply_helper(ar, ret);
 689 }
 690
 691 static int replmd_replicated_request_werror(struct replmd_replicated_request *ar, WERROR status)
 692 {
 693         int ret = LDB_ERR_OTHER;
 694         /* TODO: do some error mapping */
 695         return replmd_replicated_request_reply_helper(ar, ret);
 696 }
 697
 698 static int replmd_replicated_apply_next(struct replmd_replicated_request *ar);
 699
 700 static int replmd_replicated_apply_add_callback(struct ldb_context *ldb,
 701                                                 void *private_data,
 702                                                 struct ldb_reply *ares)
 703 {
 704 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
 705         struct replmd_replicated_request *ar = talloc_get_type(private_data,
 706                                                struct replmd_replicated_request);
 707
 708         ar->sub.change_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
 709         if (ar->sub.change_ret != LDB_SUCCESS) {
 710                 return replmd_replicated_request_error(ar, ar->sub.change_ret);
 711         }
 712
 713         talloc_free(ar->sub.mem_ctx);
 714         ZERO_STRUCT(ar->sub);
 715
 716         ar->index_current++;
 717
 718         return replmd_replicated_apply_next(ar);
 719 #else
 720         return LDB_SUCCESS;
 721 #endif
 722 }
 723
 724 static int replmd_replicated_apply_add(struct replmd_replicated_request *ar)
 725 {
 726         NTSTATUS nt_status;
 727         struct ldb_message *msg;
 728         struct replPropertyMetaDataBlob *md;
 729         struct ldb_val md_value;
 730         uint32_t i;
 731         uint64_t seq_num;
 732         int ret;
 733
 734         /*
 735          * TODO: check if the parent object exist
 736          */
 737
 738         /*
 739          * TODO: handle the conflict case where an object with the
 740          *       same name exist
 741          */
 742
 743         msg = ar->objs->objects[ar->index_current].msg;
 744         md = ar->objs->objects[ar->index_current].meta_data;
 745
 746         ret = ldb_sequence_number(ar->module->ldb, LDB_SEQ_NEXT, &seq_num);
 747         if (ret != LDB_SUCCESS) {
 748                 return replmd_replicated_request_error(ar, ret);
 749         }
 750
 751         ret = ldb_msg_add_value(msg, "objectGUID", &ar->objs->objects[ar->index_current].guid_value, NULL);
 752         if (ret != LDB_SUCCESS) {
 753                 return replmd_replicated_request_error(ar, ret);
 754         }
 755
 756         ret = ldb_msg_add_string(msg, "whenChanged", ar->objs->objects[ar->index_current].when_changed);
 757         if (ret != LDB_SUCCESS) {
 758                 return replmd_replicated_request_error(ar, ret);
 759         }
 760
 761         ret = samdb_msg_add_uint64(ar->module->ldb, msg, msg, "uSNCreated", seq_num);
 762         if (ret != LDB_SUCCESS) {
 763                 return replmd_replicated_request_error(ar, ret);
 764         }
 765
 766         ret = samdb_msg_add_uint64(ar->module->ldb, msg, msg, "uSNChanged", seq_num);
 767         if (ret != LDB_SUCCESS) {
 768                 return replmd_replicated_request_error(ar, ret);
 769         }
 770
 771         /*
 772          * the meta data array is already sorted by the caller
 773          */
 774         for (i=0; i < md->ctr.ctr1.count; i++) {
 775                 md->ctr.ctr1.array[i].local_usn = seq_num;
 776         }
 777         nt_status = ndr_push_struct_blob(&md_value, msg, md,
 778                                          (ndr_push_flags_fn_t)ndr_push_replPropertyMetaDataBlob);
 779         if (!NT_STATUS_IS_OK(nt_status)) {
 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         NTSTATUS nt_status;
 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                 nt_status = ndr_pull_struct_blob(omd_value, ar->sub.mem_ctx, &omd,
 906                                                  (ndr_pull_flags_fn_t)ndr_pull_replPropertyMetaDataBlob);
 907                 if (!NT_STATUS_IS_OK(nt_status)) {
 908                         return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
 909                 }
 910
 911                 if (omd.version != 1) {
 912                         return replmd_replicated_request_werror(ar, WERR_DS_DRA_INTERNAL_ERROR);
 913                 }
 914         }
 915
 916         ZERO_STRUCT(nmd);
 917         nmd.version = 1;
 918         nmd.ctr.ctr1.count = omd.ctr.ctr1.count + rmd->ctr.ctr1.count;
 919         nmd.ctr.ctr1.array = talloc_array(ar->sub.mem_ctx,
 920                                           struct replPropertyMetaData1,
 921                                           nmd.ctr.ctr1.count);
 922         if (!nmd.ctr.ctr1.array) return replmd_replicated_request_werror(ar, WERR_NOMEM);
 923
 924         /* first copy the old meta data */
 925         for (i=0; i < omd.ctr.ctr1.count; i++) {
 926                 nmd.ctr.ctr1.array[ni]  = omd.ctr.ctr1.array[i];
 927                 ni++;
 928         }
 929
 930         /* now merge in the new meta data */
 931         for (i=0; i < rmd->ctr.ctr1.count; i++) {
 932                 bool found = false;
 933
 934                 rmd->ctr.ctr1.array[i].local_usn = seq_num;
 935
 936                 for (j=0; j < ni; j++) {
 937                         int cmp;
 938
 939                         if (rmd->ctr.ctr1.array[i].attid != nmd.ctr.ctr1.array[j].attid) {
 940                                 continue;
 941                         }
 942
 943                         cmp = replmd_replPropertyMetaData1_conflict_compare(&rmd->ctr.ctr1.array[i],
 944                                                                             &nmd.ctr.ctr1.array[j]);
 945                         if (cmp > 0) {
 946                                 /* replace the entry */
 947                                 nmd.ctr.ctr1.array[j] = rmd->ctr.ctr1.array[i];
 948                                 found = true;
 949                                 break;
 950                         }
 951
 952                         /* we don't want to apply this change so remove the attribute */
 953                         ldb_msg_remove_element(msg, &msg->elements[i-removed_attrs]);
 954                         removed_attrs++;
 955
 956                         found = true;
 957                         break;
 958                 }
 959
 960                 if (found) continue;
 961
 962                 nmd.ctr.ctr1.array[ni] = rmd->ctr.ctr1.array[i];
 963                 ni++;
 964         }
 965
 966         /*
 967          * finally correct the size of the meta_data array
 968          */
 969         nmd.ctr.ctr1.count = ni;
 970
 971         /*
 972          * the rdn attribute (the alias for the name attribute),
 973          * 'cn' for most objects is the last entry in the meta data array
 974          * we have stored
 975          *
 976          * sort the new meta data array
 977          */
 978         {
 979                 struct replPropertyMetaData1 *rdn_p;
 980                 uint32_t rdn_idx = omd.ctr.ctr1.count - 1;
 981
 982                 rdn_p = &nmd.ctr.ctr1.array[rdn_idx];
 983                 replmd_replPropertyMetaDataCtr1_sort(&nmd.ctr.ctr1, &rdn_p->attid);
 984         }
 985
 986         /* create the meta data value */
 987         nt_status = ndr_push_struct_blob(&nmd_value, msg, &nmd,
 988                                          (ndr_push_flags_fn_t)ndr_push_replPropertyMetaDataBlob);
 989         if (!NT_STATUS_IS_OK(nt_status)) {
 990                 return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
 991         }
 992
 993         /*
 994          * check if some replicated attributes left, otherwise skip the ldb_modify() call
 995          */
 996         if (msg->num_elements == 0) {
 997                 ldb_debug(ar->module->ldb, LDB_DEBUG_TRACE, "replmd_replicated_apply_merge[%u]: skip replace\n",
 998                           ar->index_current);
 999                 goto next_object;
1000         }
1001
1002         ldb_debug(ar->module->ldb, LDB_DEBUG_TRACE, "replmd_replicated_apply_merge[%u]: replace %u attributes\n",
1003                   ar->index_current, msg->num_elements);
1004
1005         /*
1006          * when we now that we'll modify the record, add the whenChanged, uSNChanged
1007          * and replPopertyMetaData attributes
1008          */
1009         ret = ldb_msg_add_string(msg, "whenChanged", ar->objs->objects[ar->index_current].when_changed);
1010         if (ret != LDB_SUCCESS) {
1011                 return replmd_replicated_request_error(ar, ret);
1012         }
1013         ret = samdb_msg_add_uint64(ar->module->ldb, msg, msg, "uSNChanged", seq_num);
1014         if (ret != LDB_SUCCESS) {
1015                 return replmd_replicated_request_error(ar, ret);
1016         }
1017         ret = ldb_msg_add_value(msg, "replPropertyMetaData", &nmd_value, NULL);
1018         if (ret != LDB_SUCCESS) {
1019                 return replmd_replicated_request_error(ar, ret);
1020         }
1021
1022         replmd_ldb_message_sort(msg, ar->schema);
1023
1024         /* we want to replace the old values */
1025         for (i=0; i < msg->num_elements; i++) {
1026                 msg->elements[i].flags = LDB_FLAG_MOD_REPLACE;
1027         }
1028
1029         ret = ldb_build_mod_req(&ar->sub.change_req,
1030                                 ar->module->ldb,
1031                                 ar->sub.mem_ctx,
1032                                 msg,
1033                                 NULL,
1034                                 ar,
1035                                 replmd_replicated_apply_merge_callback);
1036         if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1037
1038 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1039         return ldb_next_request(ar->module, ar->sub.change_req);
1040 #else
1041         ret = ldb_next_request(ar->module, ar->sub.change_req);
1042         if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1043
1044         ar->sub.change_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
1045         if (ar->sub.change_ret != LDB_SUCCESS) {
1046                 return replmd_replicated_request_error(ar, ar->sub.change_ret);
1047         }
1048
1049 next_object:
1050         talloc_free(ar->sub.mem_ctx);
1051         ZERO_STRUCT(ar->sub);
1052
1053         ar->index_current++;
1054
1055         return LDB_SUCCESS;
1056 #endif
1057 }
1058
1059 static int replmd_replicated_apply_search_callback(struct ldb_context *ldb,
1060                                                    void *private_data,
1061                                                    struct ldb_reply *ares)
1062 {
1063         struct replmd_replicated_request *ar = talloc_get_type(private_data,
1064                                                struct replmd_replicated_request);
1065         bool is_done = false;
1066
1067         switch (ares->type) {
1068         case LDB_REPLY_ENTRY:
1069                 ar->sub.search_msg = talloc_steal(ar->sub.mem_ctx, ares->message);
1070                 break;
1071         case LDB_REPLY_REFERRAL:
1072                 /* we ignore referrals */
1073                 break;
1074         case LDB_REPLY_EXTENDED:
1075         case LDB_REPLY_DONE:
1076                 is_done = true;
1077         }
1078
1079         talloc_free(ares);
1080
1081 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1082         if (is_done) {
1083                 ar->sub.search_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
1084                 if (ar->sub.search_ret != LDB_SUCCESS) {
1085                         return replmd_replicated_request_error(ar, ar->sub.search_ret);
1086                 }
1087                 if (ar->sub.search_msg) {
1088                         return replmd_replicated_apply_merge(ar);
1089                 }
1090                 return replmd_replicated_apply_add(ar);
1091         }
1092 #endif
1093         return LDB_SUCCESS;
1094 }
1095
1096 static int replmd_replicated_apply_search(struct replmd_replicated_request *ar)
1097 {
1098         int ret;
1099         char *tmp_str;
1100         char *filter;
1101
1102         tmp_str = ldb_binary_encode(ar->sub.mem_ctx, ar->objs->objects[ar->index_current].guid_value);
1103         if (!tmp_str) return replmd_replicated_request_werror(ar, WERR_NOMEM);
1104
1105         filter = talloc_asprintf(ar->sub.mem_ctx, "(objectGUID=%s)", tmp_str);
1106         if (!filter) return replmd_replicated_request_werror(ar, WERR_NOMEM);
1107         talloc_free(tmp_str);
1108
1109         ret = ldb_build_search_req(&ar->sub.search_req,
1110                                    ar->module->ldb,
1111                                    ar->sub.mem_ctx,
1112                                    ar->objs->partition_dn,
1113                                    LDB_SCOPE_SUBTREE,
1114                                    filter,
1115                                    NULL,
1116                                    NULL,
1117                                    ar,
1118                                    replmd_replicated_apply_search_callback);
1119         if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1120
1121 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1122         return ldb_next_request(ar->module, ar->sub.search_req);
1123 #else
1124         ret = ldb_next_request(ar->module, ar->sub.search_req);
1125         if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1126
1127         ar->sub.search_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
1128         if (ar->sub.search_ret != LDB_SUCCESS) {
1129                 return replmd_replicated_request_error(ar, ar->sub.search_ret);
1130         }
1131         if (ar->sub.search_msg) {
1132                 return replmd_replicated_apply_merge(ar);
1133         }
1134
1135         return replmd_replicated_apply_add(ar);
1136 #endif
1137 }
1138
1139 static int replmd_replicated_apply_next(struct replmd_replicated_request *ar)
1140 {
1141 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1142         if (ar->index_current >= ar->objs->num_objects) {
1143                 return replmd_replicated_uptodate_vector(ar);
1144         }
1145 #endif
1146
1147         ar->sub.mem_ctx = talloc_new(ar);
1148         if (!ar->sub.mem_ctx) return replmd_replicated_request_werror(ar, WERR_NOMEM);
1149
1150         return replmd_replicated_apply_search(ar);
1151 }
1152
1153 static int replmd_replicated_uptodate_modify_callback(struct ldb_context *ldb,
1154                                                       void *private_data,
1155                                                       struct ldb_reply *ares)
1156 {
1157 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1158         struct replmd_replicated_request *ar = talloc_get_type(private_data,
1159                                                struct replmd_replicated_request);
1160
1161         ar->sub.change_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
1162         if (ar->sub.change_ret != LDB_SUCCESS) {
1163                 return replmd_replicated_request_error(ar, ar->sub.change_ret);
1164         }
1165
1166         talloc_free(ar->sub.mem_ctx);
1167         ZERO_STRUCT(ar->sub);
1168
1169         return replmd_replicated_request_done(ar);
1170 #else
1171         return LDB_SUCCESS;
1172 #endif
1173 }
1174
1175 static int replmd_drsuapi_DsReplicaCursor2_compare(const struct drsuapi_DsReplicaCursor2 *c1,
1176                                                    const struct drsuapi_DsReplicaCursor2 *c2)
1177 {
1178         return GUID_compare(&c1->source_dsa_invocation_id, &c2->source_dsa_invocation_id);
1179 }
1180
1181 static int replmd_replicated_uptodate_modify(struct replmd_replicated_request *ar)
1182 {
1183         NTSTATUS nt_status;
1184         struct ldb_message *msg;
1185         struct replUpToDateVectorBlob ouv;
1186         const struct ldb_val *ouv_value;
1187         const struct drsuapi_DsReplicaCursor2CtrEx *ruv;
1188         struct replUpToDateVectorBlob nuv;
1189         struct ldb_val nuv_value;
1190         struct ldb_message_element *nuv_el = NULL;
1191         const struct GUID *our_invocation_id;
1192         struct ldb_message_element *orf_el = NULL;
1193         struct repsFromToBlob nrf;
1194         struct ldb_val *nrf_value = NULL;
1195         struct ldb_message_element *nrf_el = NULL;
1196         uint32_t i,j,ni=0;
1197         uint64_t seq_num;
1198         bool found = false;
1199         time_t t = time(NULL);
1200         NTTIME now;
1201         int ret;
1202
1203         ruv = ar->objs->uptodateness_vector;
1204         ZERO_STRUCT(ouv);
1205         ouv.version = 2;
1206         ZERO_STRUCT(nuv);
1207         nuv.version = 2;
1208
1209         unix_to_nt_time(&now, t);
1210
1211         /*
1212          * we use the next sequence number for our own highest_usn
1213          * because we will do a modify request and this will increment
1214          * our highest_usn
1215          */
1216         ret = ldb_sequence_number(ar->module->ldb, LDB_SEQ_NEXT, &seq_num);
1217         if (ret != LDB_SUCCESS) {
1218                 return replmd_replicated_request_error(ar, ret);
1219         }
1220
1221         /*
1222          * first create the new replUpToDateVector
1223          */
1224         ouv_value = ldb_msg_find_ldb_val(ar->sub.search_msg, "replUpToDateVector");
1225         if (ouv_value) {
1226                 nt_status = ndr_pull_struct_blob(ouv_value, ar->sub.mem_ctx, &ouv,
1227                                                  (ndr_pull_flags_fn_t)ndr_pull_replUpToDateVectorBlob);
1228                 if (!NT_STATUS_IS_OK(nt_status)) {
1229                         return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
1230                 }
1231
1232                 if (ouv.version != 2) {
1233                         return replmd_replicated_request_werror(ar, WERR_DS_DRA_INTERNAL_ERROR);
1234                 }
1235         }
1236
1237         /*
1238          * the new uptodateness vector will at least
1239          * contain 1 entry, one for the source_dsa
1240          *
1241          * plus optional values from our old vector and the one from the source_dsa
1242          */
1243         nuv.ctr.ctr2.count = 1 + ouv.ctr.ctr2.count;
1244         if (ruv) nuv.ctr.ctr2.count += ruv->count;
1245         nuv.ctr.ctr2.cursors = talloc_array(ar->sub.mem_ctx,
1246                                             struct drsuapi_DsReplicaCursor2,
1247                                             nuv.ctr.ctr2.count);
1248         if (!nuv.ctr.ctr2.cursors) return replmd_replicated_request_werror(ar, WERR_NOMEM);
1249
1250         /* first copy the old vector */
1251         for (i=0; i < ouv.ctr.ctr2.count; i++) {
1252                 nuv.ctr.ctr2.cursors[ni] = ouv.ctr.ctr2.cursors[i];
1253                 ni++;
1254         }
1255
1256         /* get our invocation_id if we have one already attached to the ldb */
1257         our_invocation_id = samdb_ntds_invocation_id(ar->module->ldb);
1258
1259         /* merge in the source_dsa vector is available */
1260         for (i=0; (ruv && i < ruv->count); i++) {
1261                 found = false;
1262
1263                 if (our_invocation_id &&
1264                     GUID_equal(&ruv->cursors[i].source_dsa_invocation_id,
1265                                our_invocation_id)) {
1266                         continue;
1267                 }
1268
1269                 for (j=0; j < ni; j++) {
1270                         if (!GUID_equal(&ruv->cursors[i].source_dsa_invocation_id,
1271                                         &nuv.ctr.ctr2.cursors[j].source_dsa_invocation_id)) {
1272                                 continue;
1273                         }
1274
1275                         found = true;
1276
1277                         /*
1278                          * we update only the highest_usn and not the latest_sync_success time,
1279                          * because the last success stands for direct replication
1280                          */
1281                         if (ruv->cursors[i].highest_usn > nuv.ctr.ctr2.cursors[j].highest_usn) {
1282                                 nuv.ctr.ctr2.cursors[j].highest_usn = ruv->cursors[i].highest_usn;
1283                         }
1284                         break;
1285                 }
1286
1287                 if (found) continue;
1288
1289                 /* if it's not there yet, add it */
1290                 nuv.ctr.ctr2.cursors[ni] = ruv->cursors[i];
1291                 ni++;
1292         }
1293
1294         /*
1295          * merge in the current highwatermark for the source_dsa
1296          */
1297         found = false;
1298         for (j=0; j < ni; j++) {
1299                 if (!GUID_equal(&ar->objs->source_dsa->source_dsa_invocation_id,
1300                                 &nuv.ctr.ctr2.cursors[j].source_dsa_invocation_id)) {
1301                         continue;
1302                 }
1303
1304                 found = true;
1305
1306                 /*
1307                  * here we update the highest_usn and last_sync_success time
1308                  * because we're directly replicating from the source_dsa
1309                  *
1310                  * and use the tmp_highest_usn because this is what we have just applied
1311                  * to our ldb
1312                  */
1313                 nuv.ctr.ctr2.cursors[j].highest_usn             = ar->objs->source_dsa->highwatermark.tmp_highest_usn;
1314                 nuv.ctr.ctr2.cursors[j].last_sync_success       = now;
1315                 break;
1316         }
1317         if (!found) {
1318                 /*
1319                  * here we update the highest_usn and last_sync_success time
1320                  * because we're directly replicating from the source_dsa
1321                  *
1322                  * and use the tmp_highest_usn because this is what we have just applied
1323                  * to our ldb
1324                  */
1325                 nuv.ctr.ctr2.cursors[ni].source_dsa_invocation_id= ar->objs->source_dsa->source_dsa_invocation_id;
1326                 nuv.ctr.ctr2.cursors[ni].highest_usn            = ar->objs->source_dsa->highwatermark.tmp_highest_usn;
1327                 nuv.ctr.ctr2.cursors[ni].last_sync_success      = now;
1328                 ni++;
1329         }
1330
1331         /*
1332          * finally correct the size of the cursors array
1333          */
1334         nuv.ctr.ctr2.count = ni;
1335
1336         /*
1337          * sort the cursors
1338          */
1339         qsort(nuv.ctr.ctr2.cursors, nuv.ctr.ctr2.count,
1340               sizeof(struct drsuapi_DsReplicaCursor2),
1341               (comparison_fn_t)replmd_drsuapi_DsReplicaCursor2_compare);
1342
1343         /*
1344          * create the change ldb_message
1345          */
1346         msg = ldb_msg_new(ar->sub.mem_ctx);
1347         if (!msg) return replmd_replicated_request_werror(ar, WERR_NOMEM);
1348         msg->dn = ar->sub.search_msg->dn;
1349
1350         nt_status = ndr_push_struct_blob(&nuv_value, msg, &nuv,
1351                                          (ndr_push_flags_fn_t)ndr_push_replUpToDateVectorBlob);
1352         if (!NT_STATUS_IS_OK(nt_status)) {
1353                 return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
1354         }
1355         ret = ldb_msg_add_value(msg, "replUpToDateVector", &nuv_value, &nuv_el);
1356         if (ret != LDB_SUCCESS) {
1357                 return replmd_replicated_request_error(ar, ret);
1358         }
1359         nuv_el->flags = LDB_FLAG_MOD_REPLACE;
1360
1361         /*
1362          * now create the new repsFrom value from the given repsFromTo1 structure
1363          */
1364         ZERO_STRUCT(nrf);
1365         nrf.version                                     = 1;
1366         nrf.ctr.ctr1                                    = *ar->objs->source_dsa;
1367         /* and fix some values... */
1368         nrf.ctr.ctr1.consecutive_sync_failures          = 0;
1369         nrf.ctr.ctr1.last_success                       = now;
1370         nrf.ctr.ctr1.last_attempt                       = now;
1371         nrf.ctr.ctr1.result_last_attempt                = WERR_OK;
1372         nrf.ctr.ctr1.highwatermark.highest_usn          = nrf.ctr.ctr1.highwatermark.tmp_highest_usn;
1373
1374         /*
1375          * first see if we already have a repsFrom value for the current source dsa
1376          * if so we'll later replace this value
1377          */
1378         orf_el = ldb_msg_find_element(ar->sub.search_msg, "repsFrom");
1379         if (orf_el) {
1380                 for (i=0; i < orf_el->num_values; i++) {
1381                         struct repsFromToBlob *trf;
1382
1383                         trf = talloc(ar->sub.mem_ctx, struct repsFromToBlob);
1384                         if (!trf) return replmd_replicated_request_werror(ar, WERR_NOMEM);
1385
1386                         nt_status = ndr_pull_struct_blob(&orf_el->values[i], trf, trf,
1387                                                          (ndr_pull_flags_fn_t)ndr_pull_repsFromToBlob);
1388                         if (!NT_STATUS_IS_OK(nt_status)) {
1389                                 return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
1390                         }
1391
1392                         if (trf->version != 1) {
1393                                 return replmd_replicated_request_werror(ar, WERR_DS_DRA_INTERNAL_ERROR);
1394                         }
1395
1396                         /*
1397                          * we compare the source dsa objectGUID not the invocation_id
1398                          * because we want only one repsFrom value per source dsa
1399                          * and when the invocation_id of the source dsa has changed we don't need
1400                          * the old repsFrom with the old invocation_id
1401                          */
1402                         if (!GUID_equal(&trf->ctr.ctr1.source_dsa_obj_guid,
1403                                         &ar->objs->source_dsa->source_dsa_obj_guid)) {
1404                                 talloc_free(trf);
1405                                 continue;
1406                         }
1407
1408                         talloc_free(trf);
1409                         nrf_value = &orf_el->values[i];
1410                         break;
1411                 }
1412
1413                 /*
1414                  * copy over all old values to the new ldb_message
1415                  */
1416                 ret = ldb_msg_add_empty(msg, "repsFrom", 0, &nrf_el);
1417                 if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1418                 *nrf_el = *orf_el;
1419         }
1420
1421         /*
1422          * if we haven't found an old repsFrom value for the current source dsa
1423          * we'll add a new value
1424          */
1425         if (!nrf_value) {
1426                 struct ldb_val zero_value;
1427                 ZERO_STRUCT(zero_value);
1428                 ret = ldb_msg_add_value(msg, "repsFrom", &zero_value, &nrf_el);
1429                 if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1430
1431                 nrf_value = &nrf_el->values[nrf_el->num_values - 1];
1432         }
1433
1434         /* we now fill the value which is already attached to ldb_message */
1435         nt_status = ndr_push_struct_blob(nrf_value, msg, &nrf,
1436                                          (ndr_push_flags_fn_t)ndr_push_repsFromToBlob);
1437         if (!NT_STATUS_IS_OK(nt_status)) {
1438                 return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
1439         }
1440
1441         /*
1442          * the ldb_message_element for the attribute, has all the old values and the new one
1443          * so we'll replace the whole attribute with all values
1444          */
1445         nrf_el->flags = LDB_FLAG_MOD_REPLACE;
1446
1447         /* prepare the ldb_modify() request */
1448         ret = ldb_build_mod_req(&ar->sub.change_req,
1449                                 ar->module->ldb,
1450                                 ar->sub.mem_ctx,
1451                                 msg,
1452                                 NULL,
1453                                 ar,
1454                                 replmd_replicated_uptodate_modify_callback);
1455         if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1456
1457 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1458         return ldb_next_request(ar->module, ar->sub.change_req);
1459 #else
1460         ret = ldb_next_request(ar->module, ar->sub.change_req);
1461         if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1462
1463         ar->sub.change_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
1464         if (ar->sub.change_ret != LDB_SUCCESS) {
1465                 return replmd_replicated_request_error(ar, ar->sub.change_ret);
1466         }
1467
1468         talloc_free(ar->sub.mem_ctx);
1469         ZERO_STRUCT(ar->sub);
1470
1471         return replmd_replicated_request_done(ar);
1472 #endif
1473 }
1474
1475 static int replmd_replicated_uptodate_search_callback(struct ldb_context *ldb,
1476                                                       void *private_data,
1477                                                       struct ldb_reply *ares)
1478 {
1479         struct replmd_replicated_request *ar = talloc_get_type(private_data,
1480                                                struct replmd_replicated_request);
1481         bool is_done = false;
1482
1483         switch (ares->type) {
1484         case LDB_REPLY_ENTRY:
1485                 ar->sub.search_msg = talloc_steal(ar->sub.mem_ctx, ares->message);
1486                 break;
1487         case LDB_REPLY_REFERRAL:
1488                 /* we ignore referrals */
1489                 break;
1490         case LDB_REPLY_EXTENDED:
1491         case LDB_REPLY_DONE:
1492                 is_done = true;
1493         }
1494
1495         talloc_free(ares);
1496
1497 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1498         if (is_done) {
1499                 ar->sub.search_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
1500                 if (ar->sub.search_ret != LDB_SUCCESS) {
1501                         return replmd_replicated_request_error(ar, ar->sub.search_ret);
1502                 }
1503                 if (!ar->sub.search_msg) {
1504                         return replmd_replicated_request_werror(ar, WERR_DS_DRA_INTERNAL_ERROR);
1505                 }
1506
1507                 return replmd_replicated_uptodate_modify(ar);
1508         }
1509 #endif
1510         return LDB_SUCCESS;
1511 }
1512
1513 static int replmd_replicated_uptodate_search(struct replmd_replicated_request *ar)
1514 {
1515         int ret;
1516         static const char *attrs[] = {
1517                 "replUpToDateVector",
1518                 "repsFrom",
1519                 NULL
1520         };
1521
1522         ret = ldb_build_search_req(&ar->sub.search_req,
1523                                    ar->module->ldb,
1524                                    ar->sub.mem_ctx,
1525                                    ar->objs->partition_dn,
1526                                    LDB_SCOPE_BASE,
1527                                    "(objectClass=*)",
1528                                    attrs,
1529                                    NULL,
1530                                    ar,
1531                                    replmd_replicated_uptodate_search_callback);
1532         if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1533
1534 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1535         return ldb_next_request(ar->module, ar->sub.search_req);
1536 #else
1537         ret = ldb_next_request(ar->module, ar->sub.search_req);
1538         if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
1539
1540         ar->sub.search_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
1541         if (ar->sub.search_ret != LDB_SUCCESS) {
1542                 return replmd_replicated_request_error(ar, ar->sub.search_ret);
1543         }
1544         if (!ar->sub.search_msg) {
1545                 return replmd_replicated_request_werror(ar, WERR_DS_DRA_INTERNAL_ERROR);
1546         }
1547
1548         return replmd_replicated_uptodate_modify(ar);
1549 #endif
1550 }
1551
1552 static int replmd_replicated_uptodate_vector(struct replmd_replicated_request *ar)
1553 {
1554         ar->sub.mem_ctx = talloc_new(ar);
1555         if (!ar->sub.mem_ctx) return replmd_replicated_request_werror(ar, WERR_NOMEM);
1556
1557         return replmd_replicated_uptodate_search(ar);
1558 }
1559
1560 static int replmd_extended_replicated_objects(struct ldb_module *module, struct ldb_request *req)
1561 {
1562         struct dsdb_extended_replicated_objects *objs;
1563         struct replmd_replicated_request *ar;
1564
1565         ldb_debug(module->ldb, LDB_DEBUG_TRACE, "replmd_extended_replicated_objects\n");
1566
1567         objs = talloc_get_type(req->op.extended.data, struct dsdb_extended_replicated_objects);
1568         if (!objs) {
1569                 ldb_debug(module->ldb, LDB_DEBUG_FATAL, "replmd_extended_replicated_objects: invalid extended data\n");
1570                 return LDB_ERR_PROTOCOL_ERROR;
1571         }
1572
1573         if (objs->version != DSDB_EXTENDED_REPLICATED_OBJECTS_VERSION) {
1574                 ldb_debug(module->ldb, LDB_DEBUG_FATAL, "replmd_extended_replicated_objects: extended data invalid version [%u != %u]\n",
1575                           objs->version, DSDB_EXTENDED_REPLICATED_OBJECTS_VERSION);
1576                 return LDB_ERR_PROTOCOL_ERROR;
1577         }
1578
1579         ar = replmd_replicated_init_handle(module, req, objs);
1580         if (!ar) {
1581                 return LDB_ERR_OPERATIONS_ERROR;
1582         }
1583
1584 #ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */
1585         return replmd_replicated_apply_next(ar);
1586 #else
1587         while (ar->index_current < ar->objs->num_objects &&
1588                req->handle->state != LDB_ASYNC_DONE) {
1589                 replmd_replicated_apply_next(ar);
1590         }
1591
1592         if (req->handle->state != LDB_ASYNC_DONE) {
1593                 replmd_replicated_uptodate_vector(ar);
1594         }
1595
1596         return LDB_SUCCESS;
1597 #endif
1598 }
1599
1600 static int replmd_extended(struct ldb_module *module, struct ldb_request *req)
1601 {
1602         if (strcmp(req->op.extended.oid, DSDB_EXTENDED_REPLICATED_OBJECTS_OID) == 0) {
1603                 return replmd_extended_replicated_objects(module, req);
1604         }
1605
1606         return ldb_next_request(module, req);
1607 }
1608
1609 static int replmd_wait_none(struct ldb_handle *handle) {
1610         struct replmd_replicated_request *ar;
1611
1612         if (!handle || !handle->private_data) {
1613                 return LDB_ERR_OPERATIONS_ERROR;
1614         }
1615
1616         ar = talloc_get_type(handle->private_data, struct replmd_replicated_request);
1617         if (!ar) {
1618                 return LDB_ERR_OPERATIONS_ERROR;
1619         }
1620
1621         /* we do only sync calls */
1622         if (handle->state != LDB_ASYNC_DONE) {
1623                 return LDB_ERR_OPERATIONS_ERROR;
1624         }
1625
1626         return handle->status;
1627 }
1628
1629 static int replmd_wait_all(struct ldb_handle *handle) {
1630
1631         int ret;
1632
1633         while (handle->state != LDB_ASYNC_DONE) {
1634                 ret = replmd_wait_none(handle);
1635                 if (ret != LDB_SUCCESS) {
1636                         return ret;
1637                 }
1638         }
1639
1640         return handle->status;
1641 }
1642
1643 static int replmd_wait(struct ldb_handle *handle, enum ldb_wait_type type)
1644 {
1645         if (type == LDB_WAIT_ALL) {
1646                 return replmd_wait_all(handle);
1647         } else {
1648                 return replmd_wait_none(handle);
1649         }
1650 }
1651
1652 static const struct ldb_module_ops replmd_ops = {
1653         .name          = "repl_meta_data",
1654         .add           = replmd_add,
1655         .modify        = replmd_modify,
1656         .extended      = replmd_extended,
1657         .wait          = replmd_wait
1658 };
1659
1660 int repl_meta_data_module_init(void)
1661 {
1662         return ldb_register_module(&replmd_ops);
1663 }