s4-repl: Allow dsdb_replicated_objects_commit() to use different schema while committ...
[kai/samba.git] / source4 / dsdb / repl / replicated_objects.c
index 552d83f411ce30de257d7c58ef3c45501867bd34..f3b6356649b8fca0eee336f54ba8a62079f01dc7 100644 (file)
 #include "includes.h"
 #include "dsdb/samdb/samdb.h"
 #include "lib/ldb/include/ldb_errors.h"
-#include "lib/util/dlinklist.h"
+#include "../lib/util/dlinklist.h"
 #include "librpc/gen_ndr/ndr_misc.h"
 #include "librpc/gen_ndr/ndr_drsuapi.h"
 #include "librpc/gen_ndr/ndr_drsblobs.h"
-#include "lib/crypto/crypto.h"
+#include "../lib/crypto/crypto.h"
+#include "../libcli/drsuapi/drsuapi.h"
 #include "libcli/auth/libcli_auth.h"
 #include "param/param.h"
 
-static WERROR dsdb_decrypt_attribute_value(TALLOC_CTX *mem_ctx,
-                                          const DATA_BLOB *gensec_skey,
-                                          bool rid_crypt,
-                                          uint32_t rid,
-                                          DATA_BLOB *in,
-                                          DATA_BLOB *out)
+/**
+ * Multi-pass working schema creation
+ * Function will:
+ *  - shallow copy initial schema supplied
+ *  - create a working schema in multiple passes
+ *    until all objects are resolved
+ * Working schema is a schema with Attributes, Classes
+ * and indexes, but w/o subClassOf, possibleSupperiors etc.
+ * It is to be used just us cache for converting attribute values.
+ */
+WERROR dsdb_repl_make_working_schema(struct ldb_context *ldb,
+                                    const struct dsdb_schema *initial_schema,
+                                    const struct drsuapi_DsReplicaOIDMapping_Ctr *mapping_ctr,
+                                    uint32_t object_count,
+                                    const struct drsuapi_DsReplicaObjectListItemEx *first_object,
+                                    const DATA_BLOB *gensec_skey,
+                                    TALLOC_CTX *mem_ctx,
+                                    struct dsdb_schema **_schema_out)
 {
-       DATA_BLOB confounder;
-       DATA_BLOB enc_buffer;
-
-       struct MD5Context md5;
-       uint8_t _enc_key[16];
-       DATA_BLOB enc_key;
-
-       DATA_BLOB dec_buffer;
-
-       uint32_t crc32_given;
-       uint32_t crc32_calc;
-       DATA_BLOB checked_buffer;
-
-       DATA_BLOB plain_buffer;
+       struct schema_list {
+               struct schema_list *next, *prev;
+               const struct drsuapi_DsReplicaObjectListItemEx *obj;
+       };
+
+       WERROR werr;
+       struct dsdb_schema_prefixmap *pfm_remote;
+       struct schema_list *schema_list = NULL, *schema_list_item, *schema_list_next_item;
+       struct dsdb_schema *working_schema;
+       const struct drsuapi_DsReplicaObjectListItemEx *cur;
+       int ret, pass_no;
 
-       /*
-        * users with rid == 0 should not exist
-        */
-       if (rid_crypt && rid == 0) {
-               return WERR_DS_DRA_INVALID_PARAMETER;
+       /* make a copy of the iniatial_scheam so we don't mess with it */
+       working_schema = dsdb_schema_copy_shallow(mem_ctx, ldb, initial_schema);
+       if (!working_schema) {
+               DEBUG(0,(__location__ ": schema copy failed!\n"));
+               return WERR_NOMEM;
        }
 
-       /* 
-        * the first 16 bytes at the beginning are the confounder
-        * followed by the 4 byte crc32 checksum
-        */
-       if (in->length < 20) {
-               return WERR_DS_DRA_INVALID_PARAMETER;
-       }
-       confounder = data_blob_const(in->data, 16);
-       enc_buffer = data_blob_const(in->data + 16, in->length - 16);
-
-       /* 
-        * build the encryption key md5 over the session key followed
-        * by the confounder
-        * 
-        * here the gensec session key is used and
-        * not the dcerpc ncacn_ip_tcp "SystemLibraryDTC" key!
-        */
-       enc_key = data_blob_const(_enc_key, sizeof(_enc_key));
-       MD5Init(&md5);
-       MD5Update(&md5, gensec_skey->data, gensec_skey->length);
-       MD5Update(&md5, confounder.data, confounder.length);
-       MD5Final(enc_key.data, &md5);
-
-       /*
-        * copy the encrypted buffer part and 
-        * decrypt it using the created encryption key using arcfour
-        */
-       dec_buffer = data_blob_const(enc_buffer.data, enc_buffer.length);
-       arcfour_crypt_blob(dec_buffer.data, dec_buffer.length, &enc_key);
+       /* we are going to need remote prefixMap for decoding */
+       werr = dsdb_schema_pfm_from_drsuapi_pfm(mapping_ctr, true,
+                                               mem_ctx, &pfm_remote, NULL);
+       if (!W_ERROR_IS_OK(werr)) {
+               DEBUG(0,(__location__ ": Failed to decode remote prefixMap: %s",
+                        win_errstr(werr)));
+               return werr;
+       }
 
-       /* 
-        * the first 4 byte are the crc32 checksum
-        * of the remaining bytes
-        */
-       crc32_given = IVAL(dec_buffer.data, 0);
-       crc32_calc = crc32_calc_buffer(dec_buffer.data + 4 , dec_buffer.length - 4);
-       if (crc32_given != crc32_calc) {
-               return WERR_SEC_E_DECRYPT_FAILURE;
+       /* create a list of objects yet to be converted */
+       for (cur = first_object; cur; cur = cur->next_object) {
+               schema_list_item = talloc(mem_ctx, struct schema_list);
+               schema_list_item->obj = cur;
+               DLIST_ADD_END(schema_list, schema_list_item, struct schema_list);
        }
-       checked_buffer = data_blob_const(dec_buffer.data + 4, dec_buffer.length - 4);
 
-       plain_buffer = data_blob_talloc(mem_ctx, checked_buffer.data, checked_buffer.length);
-       W_ERROR_HAVE_NO_MEMORY(plain_buffer.data);
+       /* resolve objects until all are resolved and in local schema */
+       pass_no = 1;
+
+       while (schema_list) {
+               uint32_t converted_obj_count = 0;
+               uint32_t failed_obj_count = 0;
+               TALLOC_CTX *tmp_ctx = talloc_new(mem_ctx);
+               W_ERROR_HAVE_NO_MEMORY(tmp_ctx);
+
+               for (schema_list_item = schema_list; schema_list_item; schema_list_item=schema_list_next_item) {
+                       struct dsdb_extended_replicated_object object;
+
+                       cur = schema_list_item->obj;
+
+                       /* Save the next item, now we have saved out
+                        * the current one, so we can DLIST_REMOVE it
+                        * safely */
+                       schema_list_next_item = schema_list_item->next;
+
+                       /*
+                        * Convert the objects into LDB messages using the
+                        * schema we have so far. It's ok if we fail to convert
+                        * an object. We should convert more objects on next pass.
+                        */
+                       werr = dsdb_convert_object_ex(ldb, working_schema, pfm_remote,
+                                                     cur, gensec_skey,
+                                                     tmp_ctx, &object);
+                       if (!W_ERROR_IS_OK(werr)) {
+                               DEBUG(1,("Warning: Failed to convert schema object %s into ldb msg\n",
+                                        cur->object.identifier->dn));
+
+                               failed_obj_count++;
+                       } else {
+                               /*
+                                * Convert the schema from ldb_message format
+                                * (OIDs as OID strings) into schema, using
+                                * the remote prefixMap
+                                */
+                               werr = dsdb_schema_set_el_from_ldb_msg(ldb,
+                                                                      working_schema,
+                                                                      object.msg);
+                               if (!W_ERROR_IS_OK(werr)) {
+                                       DEBUG(1,("Warning: failed to convert object %s into a schema element: %s\n",
+                                                ldb_dn_get_linearized(object.msg->dn),
+                                                win_errstr(werr)));
+                                       failed_obj_count++;
+                               } else {
+                                       DLIST_REMOVE(schema_list, schema_list_item);
+                                       talloc_free(schema_list_item);
+                                       converted_obj_count++;
+                               }
+                       }
+               }
+               talloc_free(tmp_ctx);
 
-       /*
-        * The following rid_crypt obfuscation isn't session specific
-        * and not really needed here, because we allways know the rid of the
-        * user account.
-        *
-        * But for the rest of samba it's easier when we remove this static
-        * obfuscation here
-        */
-       if (rid_crypt) {
-               uint32_t i, num_hashes;
+               DEBUG(4,("Schema load pass %d: %d/%d of %d objects left to be converted.\n",
+                        pass_no, failed_obj_count, converted_obj_count, object_count));
+               pass_no++;
 
-               if ((checked_buffer.length % 16) != 0) {
-                       return WERR_DS_DRA_INVALID_PARAMETER;
+               /* check if we converted any objects in this pass */
+               if (converted_obj_count == 0) {
+                       DEBUG(0,("Can't continue Schema load: didn't manage to convert any objects: all %d remaining of %d objects failed to convert\n", failed_obj_count, object_count));
+                       return WERR_INTERNAL_ERROR;
                }
 
-               num_hashes = plain_buffer.length / 16;
-               for (i = 0; i < num_hashes; i++) {
-                       uint32_t offset = i * 16;
-                       sam_rid_crypt(rid, checked_buffer.data + offset, plain_buffer.data + offset, 0);
+               /* rebuild indexes */
+               ret = dsdb_setup_sorted_accessors(ldb, working_schema);
+               if (LDB_SUCCESS != ret) {
+                       DEBUG(0,("Failed to create schema-cache indexes!\n"));
+                       return WERR_INTERNAL_ERROR;
                }
-       }
-
-       *out = plain_buffer;
-       return WERR_OK;
-}
+       };
 
-static WERROR dsdb_decrypt_attribute(const DATA_BLOB *gensec_skey,
-                                    uint32_t rid,
-                                    struct drsuapi_DsReplicaAttribute *attr)
-{
-       WERROR status;
-       TALLOC_CTX *mem_ctx;
-       DATA_BLOB *enc_data;
-       DATA_BLOB plain_data;
-       bool rid_crypt = false;
-
-       if (attr->value_ctr.num_values == 0) {
-               return WERR_OK;
-       }
-
-       switch (attr->attid) {
-       case DRSUAPI_ATTRIBUTE_dBCSPwd:
-       case DRSUAPI_ATTRIBUTE_unicodePwd:
-       case DRSUAPI_ATTRIBUTE_ntPwdHistory:
-       case DRSUAPI_ATTRIBUTE_lmPwdHistory:
-               rid_crypt = true;
-               break;
-       case DRSUAPI_ATTRIBUTE_supplementalCredentials:
-       case DRSUAPI_ATTRIBUTE_priorValue:
-       case DRSUAPI_ATTRIBUTE_currentValue:
-       case DRSUAPI_ATTRIBUTE_trustAuthOutgoing:
-       case DRSUAPI_ATTRIBUTE_trustAuthIncoming:
-       case DRSUAPI_ATTRIBUTE_initialAuthOutgoing:
-       case DRSUAPI_ATTRIBUTE_initialAuthIncoming:
-               break;
-       default:
-               return WERR_OK;
-       }
-
-       if (attr->value_ctr.num_values > 1) {
-               return WERR_DS_DRA_INVALID_PARAMETER;
-       }
-
-       if (!attr->value_ctr.values[0].blob) {
-               return WERR_DS_DRA_INVALID_PARAMETER;
-       }
-
-       mem_ctx         = attr->value_ctr.values[0].blob;
-       enc_data        = attr->value_ctr.values[0].blob;
-
-       status = dsdb_decrypt_attribute_value(mem_ctx,
-                                             gensec_skey,
-                                             rid_crypt,
-                                             rid,
-                                             enc_data,
-                                             &plain_data);
-       W_ERROR_NOT_OK_RETURN(status);
-
-       talloc_free(attr->value_ctr.values[0].blob->data);
-       *attr->value_ctr.values[0].blob = plain_data;
+       *_schema_out = working_schema;
 
        return WERR_OK;
 }
 
-static WERROR dsdb_convert_object(struct ldb_context *ldb,
-                                 const struct dsdb_schema *schema,
-                                 struct dsdb_extended_replicated_objects *ctr,
-                                 const struct drsuapi_DsReplicaObjectListItemEx *in,
-                                 const DATA_BLOB *gensec_skey,
-                                 TALLOC_CTX *mem_ctx,
-                                 struct dsdb_extended_replicated_object *out)
+WERROR dsdb_convert_object_ex(struct ldb_context *ldb,
+                             const struct dsdb_schema *schema,
+                             const struct dsdb_schema_prefixmap *pfm_remote,
+                             const struct drsuapi_DsReplicaObjectListItemEx *in,
+                             const DATA_BLOB *gensec_skey,
+                             TALLOC_CTX *mem_ctx,
+                             struct dsdb_extended_replicated_object *out)
 {
        NTSTATUS nt_status;
-       enum ndr_err_code ndr_err;
        WERROR status;
        uint32_t i;
        struct ldb_message *msg;
@@ -273,16 +248,20 @@ static WERROR dsdb_convert_object(struct ldb_context *ldb,
                struct drsuapi_DsReplicaMetaData *d;
                struct replPropertyMetaData1 *m;
                struct ldb_message_element *e;
+               uint32_t j;
 
                a = &in->object.attribute_ctr.attributes[i];
                d = &in->meta_data_ctr->meta_data[i];
                m = &md->ctr.ctr1.array[i];
                e = &msg->elements[i];
 
-               status = dsdb_decrypt_attribute(gensec_skey, rid, a);
-               W_ERROR_NOT_OK_RETURN(status);
+               for (j=0; j<a->value_ctr.num_values; j++) {
+                       status = drsuapi_decrypt_attribute(a->value_ctr.values[j].blob, gensec_skey, rid, a);
+                       W_ERROR_NOT_OK_RETURN(status);
+               }
 
-               status = dsdb_attribute_drsuapi_to_ldb(schema, a, msg->elements, e);
+               status = dsdb_attribute_drsuapi_to_ldb(ldb, schema, pfm_remote,
+                                                      a, msg->elements, e);
                W_ERROR_NOT_OK_RETURN(status);
 
                m->attid                        = a->attid;
@@ -296,7 +275,7 @@ static WERROR dsdb_convert_object(struct ldb_context *ldb,
                        whenChanged = d->originating_change_time;
                }
 
-               if (a->attid == DRSUAPI_ATTRIBUTE_name) {
+               if (a->attid == DRSUAPI_ATTID_name) {
                        name_a = a;
                        name_d = d;
                        rdn_m = &md->ctr.ctr1.array[md->ctr.ctr1.count];
@@ -304,9 +283,25 @@ static WERROR dsdb_convert_object(struct ldb_context *ldb,
        }
 
        if (rdn_m) {
-               ret = ldb_msg_add_value(msg, rdn_attr->lDAPDisplayName, rdn_value, NULL);
-               if (ret != LDB_SUCCESS) {
-                       return WERR_FOOBAR;
+               struct ldb_message_element *el;
+               el = ldb_msg_find_element(msg, rdn_attr->lDAPDisplayName);
+               if (!el) {
+                       ret = ldb_msg_add_value(msg, rdn_attr->lDAPDisplayName, rdn_value, NULL);
+                       if (ret != LDB_SUCCESS) {
+                               return WERR_FOOBAR;
+                       }
+               } else {
+                       if (el->num_values != 1) {
+                               DEBUG(0,(__location__ ": Unexpected num_values=%u\n",
+                                        el->num_values));
+                               return WERR_FOOBAR;                             
+                       }
+                       if (!ldb_val_equal_exact(&el->values[0], rdn_value)) {
+                               DEBUG(0,(__location__ ": RDN value changed? '%*.*s' '%*.*s'\n",
+                                        (int)el->values[0].length, (int)el->values[0].length, el->values[0].data,
+                                        (int)rdn_value->length, (int)rdn_value->length, rdn_value->data));
+                               return WERR_FOOBAR;                             
+                       }
                }
 
                rdn_m->attid                            = rdn_attid;
@@ -323,12 +318,8 @@ static WERROR dsdb_convert_object(struct ldb_context *ldb,
        whenChanged_s = ldb_timestring(msg, whenChanged_t);
        W_ERROR_HAVE_NO_MEMORY(whenChanged_s);
 
-       ndr_err = ndr_push_struct_blob(&guid_value, msg, 
-                                      lp_iconv_convenience(ldb_get_opaque(ldb, "loadparm")),
-                                      &in->object.identifier->guid,
-                                        (ndr_push_flags_fn_t)ndr_push_GUID);
-       if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
-               nt_status = ndr_map_error2ntstatus(ndr_err);
+       nt_status = GUID_to_ndr_blob(&in->object.identifier->guid, msg, &guid_value);
+       if (!NT_STATUS_IS_OK(nt_status)) {
                return ntstatus_to_werror(nt_status);
        }
 
@@ -339,41 +330,65 @@ static WERROR dsdb_convert_object(struct ldb_context *ldb,
        return WERR_OK;
 }
 
-WERROR dsdb_extended_replicated_objects_commit(struct ldb_context *ldb,
-                                              const char *partition_dn,
-                                              const struct drsuapi_DsReplicaOIDMapping_Ctr *mapping_ctr,
-                                              uint32_t object_count,
-                                              const struct drsuapi_DsReplicaObjectListItemEx *first_object,
-                                              uint32_t linked_attributes_count,
-                                              const struct drsuapi_DsReplicaLinkedAttribute *linked_attributes,
-                                              const struct repsFromTo1 *source_dsa,
-                                              const struct drsuapi_DsReplicaCursor2CtrEx *uptodateness_vector,
-                                              const DATA_BLOB *gensec_skey,
-                                              TALLOC_CTX *mem_ctx,
-                                              struct dsdb_extended_replicated_objects **_out)
+WERROR dsdb_replicated_objects_convert(struct ldb_context *ldb,
+                                      const struct dsdb_schema *schema,
+                                      const char *partition_dn_str,
+                                      const struct drsuapi_DsReplicaOIDMapping_Ctr *mapping_ctr,
+                                      uint32_t object_count,
+                                      const struct drsuapi_DsReplicaObjectListItemEx *first_object,
+                                      uint32_t linked_attributes_count,
+                                      const struct drsuapi_DsReplicaLinkedAttribute *linked_attributes,
+                                      const struct repsFromTo1 *source_dsa,
+                                      const struct drsuapi_DsReplicaCursor2CtrEx *uptodateness_vector,
+                                      const DATA_BLOB *gensec_skey,
+                                      TALLOC_CTX *mem_ctx,
+                                      struct dsdb_extended_replicated_objects **objects)
 {
        WERROR status;
-       const struct dsdb_schema *schema;
+       struct ldb_dn *partition_dn;
+       struct dsdb_schema_prefixmap *pfm_remote;
        struct dsdb_extended_replicated_objects *out;
-       struct ldb_result *ext_res;
        const struct drsuapi_DsReplicaObjectListItemEx *cur;
        uint32_t i;
-       int ret;
-
-       schema = dsdb_get_schema(ldb);
-       if (!schema) {
-               return WERR_DS_SCHEMA_NOT_LOADED;
-       }
-
-       status = dsdb_verify_oid_mappings_drsuapi(schema, mapping_ctr);
-       W_ERROR_NOT_OK_RETURN(status);
 
        out = talloc_zero(mem_ctx, struct dsdb_extended_replicated_objects);
        W_ERROR_HAVE_NO_MEMORY(out);
        out->version            = DSDB_EXTENDED_REPLICATED_OBJECTS_VERSION;
 
-       out->partition_dn       = ldb_dn_new(out, ldb, partition_dn);
-       W_ERROR_HAVE_NO_MEMORY(out->partition_dn);
+       /*
+        * Ensure schema is kept valid for as long as 'out'
+        * which may contain pointers to it
+        */
+       schema = talloc_reference(out, schema);
+       W_ERROR_HAVE_NO_MEMORY(schema);
+
+       partition_dn = ldb_dn_new(out, ldb, partition_dn_str);
+       W_ERROR_HAVE_NO_MEMORY_AND_FREE(partition_dn, out);
+
+       status = dsdb_schema_pfm_from_drsuapi_pfm(mapping_ctr, true,
+                                                 out, &pfm_remote, NULL);
+       if (!W_ERROR_IS_OK(status)) {
+               DEBUG(0,(__location__ ": Failed to decode remote prefixMap: %s",
+                        win_errstr(status)));
+               talloc_free(out);
+               return status;
+       }
+
+       if (ldb_dn_compare(partition_dn, ldb_get_schema_basedn(ldb)) != 0) {
+               /*
+                * check for schema changes in case
+                * we are not replicating Schema NC
+                */
+               status = dsdb_schema_info_cmp(schema, mapping_ctr);
+               if (!W_ERROR_IS_OK(status)) {
+                       DEBUG(1,("Remote schema has changed while replicating %s\n",
+                                partition_dn_str));
+                       talloc_free(out);
+                       return status;
+               }
+       }
+
+       out->partition_dn       = partition_dn;
 
        out->source_dsa         = source_dsa;
        out->uptodateness_vector= uptodateness_vector;
@@ -382,36 +397,344 @@ WERROR dsdb_extended_replicated_objects_commit(struct ldb_context *ldb,
        out->objects            = talloc_array(out,
                                               struct dsdb_extended_replicated_object,
                                               out->num_objects);
-       W_ERROR_HAVE_NO_MEMORY(out->objects);
+       W_ERROR_HAVE_NO_MEMORY_AND_FREE(out->objects, out);
+
+       /* pass the linked attributes down to the repl_meta_data
+          module */
+       out->linked_attributes_count = linked_attributes_count;
+       out->linked_attributes       = linked_attributes;
 
        for (i=0, cur = first_object; cur; cur = cur->next_object, i++) {
                if (i == out->num_objects) {
+                       talloc_free(out);
                        return WERR_FOOBAR;
                }
 
-               status = dsdb_convert_object(ldb, schema, out, cur, gensec_skey, out->objects, &out->objects[i]);
-               W_ERROR_NOT_OK_RETURN(status);
+               status = dsdb_convert_object_ex(ldb, schema, pfm_remote,
+                                               cur, gensec_skey,
+                                               out->objects, &out->objects[i]);
+               if (!W_ERROR_IS_OK(status)) {
+                       talloc_free(out);
+                       DEBUG(0,("Failed to convert object %s: %s\n",
+                                cur->object.identifier->dn,
+                                win_errstr(status)));
+                       return status;
+               }
        }
        if (i != out->num_objects) {
+               talloc_free(out);
                return WERR_FOOBAR;
        }
 
+       /* free pfm_remote, we won't need it anymore */
+       talloc_free(pfm_remote);
+
+       *objects = out;
+       return WERR_OK;
+}
+
+/**
+ * Commits a list of replicated objects.
+ *
+ * @param working_schema dsdb_schema to be used for resolving
+ *                      Classes/Attributes during Schema replication. If not NULL,
+ *                      it will be set on ldb and used while committing replicated objects
+ */
+WERROR dsdb_replicated_objects_commit(struct ldb_context *ldb,
+                                     struct dsdb_schema *working_schema,
+                                     struct dsdb_extended_replicated_objects *objects,
+                                     uint64_t *notify_uSN)
+{
+       struct ldb_result *ext_res;
+       struct dsdb_schema *cur_schema = NULL;
+       int ret;
+       uint64_t seq_num1, seq_num2;
+
        /* TODO: handle linked attributes */
 
-       ret = ldb_extended(ldb, DSDB_EXTENDED_REPLICATED_OBJECTS_OID, out, &ext_res);
+       /* wrap the extended operation in a transaction 
+          See [MS-DRSR] 3.3.2 Transactions
+        */
+       ret = ldb_transaction_start(ldb);
+       if (ret != LDB_SUCCESS) {
+               DEBUG(0,(__location__ " Failed to start transaction\n"));
+               return WERR_FOOBAR;
+       }
+
+       ret = dsdb_load_partition_usn(ldb, objects->partition_dn, &seq_num1, NULL);
+       if (ret != LDB_SUCCESS) {
+               DEBUG(0,(__location__ " Failed to load partition uSN\n"));
+               ldb_transaction_cancel(ldb);
+               return WERR_FOOBAR;             
+       }
+
+       /*
+        * Set working_schema for ldb in case we are replicating from Schema NC.
+        * Schema won't be reloaded during Replicated Objects commit, as it is
+        * done in a transaction. So we need some way to search for newly
+        * added Classes and Attributes
+        */
+       if (working_schema) {
+               /* store current schema so we can fall back in case of failure */
+               cur_schema = dsdb_get_schema(ldb, objects);
+
+               ret = dsdb_reference_schema(ldb, working_schema, false);
+               if (ret != LDB_SUCCESS) {
+                       DEBUG(0,(__location__ "Failed to reference working schema - %s\n",
+                                ldb_strerror(ret)));
+                       /* TODO: Map LDB Error to NTSTATUS? */
+                       ldb_transaction_cancel(ldb);
+                       return WERR_INTERNAL_ERROR;
+               }
+       }
+
+       ret = ldb_extended(ldb, DSDB_EXTENDED_REPLICATED_OBJECTS_OID, objects, &ext_res);
        if (ret != LDB_SUCCESS) {
+               /* restore previous schema */
+               if (cur_schema ) {
+                       dsdb_reference_schema(ldb, cur_schema, false);
+               }
+
                DEBUG(0,("Failed to apply records: %s: %s\n",
                         ldb_errstring(ldb), ldb_strerror(ret)));
-               talloc_free(out);
+               ldb_transaction_cancel(ldb);
                return WERR_FOOBAR;
        }
        talloc_free(ext_res);
 
-       if (_out) {
-               *_out = out;
-       } else {
-               talloc_free(out);
+       ret = ldb_transaction_prepare_commit(ldb);
+       if (ret != LDB_SUCCESS) {
+               /* restore previous schema */
+               if (cur_schema ) {
+                       dsdb_reference_schema(ldb, cur_schema, false);
+               }
+               DEBUG(0,(__location__ " Failed to prepare commit of transaction: %s\n",
+                        ldb_errstring(ldb)));
+               return WERR_FOOBAR;
+       }
+
+       ret = dsdb_load_partition_usn(ldb, objects->partition_dn, &seq_num2, NULL);
+       if (ret != LDB_SUCCESS) {
+               /* restore previous schema */
+               if (cur_schema ) {
+                       dsdb_reference_schema(ldb, cur_schema, false);
+               }
+               DEBUG(0,(__location__ " Failed to load partition uSN\n"));
+               ldb_transaction_cancel(ldb);
+               return WERR_FOOBAR;             
        }
 
+       /* if this replication partner didn't need to be notified
+          before this transaction then it still doesn't need to be
+          notified, as the changes came from this server */    
+       if (seq_num2 > seq_num1 && seq_num1 <= *notify_uSN) {
+               *notify_uSN = seq_num2;
+       }
+
+       ret = ldb_transaction_commit(ldb);
+       if (ret != LDB_SUCCESS) {
+               /* restore previous schema */
+               if (cur_schema ) {
+                       dsdb_reference_schema(ldb, cur_schema, false);
+               }
+               DEBUG(0,(__location__ " Failed to commit transaction\n"));
+               return WERR_FOOBAR;
+       }
+
+       /*
+        * Reset the Schema used by ldb. This will lead to
+        * a schema cache being refreshed from database.
+        */
+       if (working_schema) {
+               cur_schema = dsdb_get_schema(ldb, NULL);
+               /* TODO: What we do in case dsdb_get_schema() fail?
+                *       We can't fallback at this point anymore */
+       }
+
+       DEBUG(2,("Replicated %u objects (%u linked attributes) for %s\n",
+                objects->num_objects, objects->linked_attributes_count,
+                ldb_dn_get_linearized(objects->partition_dn)));
+                
        return WERR_OK;
 }
+
+static WERROR dsdb_origin_object_convert(struct ldb_context *ldb,
+                                        const struct dsdb_schema *schema,
+                                        const struct drsuapi_DsReplicaObjectListItem *in,
+                                        TALLOC_CTX *mem_ctx,
+                                        struct ldb_message **_msg)
+{
+       WERROR status;
+       unsigned int i;
+       struct ldb_message *msg;
+
+       if (!in->object.identifier) {
+               return WERR_FOOBAR;
+       }
+
+       if (!in->object.identifier->dn || !in->object.identifier->dn[0]) {
+               return WERR_FOOBAR;
+       }
+
+       msg = ldb_msg_new(mem_ctx);
+       W_ERROR_HAVE_NO_MEMORY(msg);
+
+       msg->dn = ldb_dn_new(msg, ldb, in->object.identifier->dn);
+       W_ERROR_HAVE_NO_MEMORY(msg->dn);
+
+       msg->num_elements       = in->object.attribute_ctr.num_attributes;
+       msg->elements           = talloc_array(msg, struct ldb_message_element,
+                                              msg->num_elements);
+       W_ERROR_HAVE_NO_MEMORY(msg->elements);
+
+       for (i=0; i < msg->num_elements; i++) {
+               struct drsuapi_DsReplicaAttribute *a;
+               struct ldb_message_element *e;
+
+               a = &in->object.attribute_ctr.attributes[i];
+               e = &msg->elements[i];
+
+               status = dsdb_attribute_drsuapi_to_ldb(ldb, schema, schema->prefixmap,
+                                                      a, msg->elements, e);
+               W_ERROR_NOT_OK_RETURN(status);
+       }
+
+
+       *_msg = msg;
+
+       return WERR_OK;
+}
+
+WERROR dsdb_origin_objects_commit(struct ldb_context *ldb,
+                                 TALLOC_CTX *mem_ctx,
+                                 const struct drsuapi_DsReplicaObjectListItem *first_object,
+                                 uint32_t *_num,
+                                 struct drsuapi_DsReplicaObjectIdentifier2 **_ids)
+{
+       WERROR status;
+       const struct dsdb_schema *schema;
+       const struct drsuapi_DsReplicaObjectListItem *cur;
+       struct ldb_message **objects;
+       struct drsuapi_DsReplicaObjectIdentifier2 *ids;
+       uint32_t i;
+       uint32_t num_objects = 0;
+       const char * const attrs[] = {
+               "objectGUID",
+               "objectSid",
+               NULL
+       };
+       struct ldb_result *res;
+       int ret;
+
+       for (cur = first_object; cur; cur = cur->next_object) {
+               num_objects++;
+       }
+
+       if (num_objects == 0) {
+               return WERR_OK;
+       }
+
+       ret = ldb_transaction_start(ldb);
+       if (ret != LDB_SUCCESS) {
+               return WERR_DS_INTERNAL_FAILURE;
+       }
+
+       objects = talloc_array(mem_ctx, struct ldb_message *,
+                              num_objects);
+       if (objects == NULL) {
+               status = WERR_NOMEM;
+               goto cancel;
+       }
+
+       schema = dsdb_get_schema(ldb, objects);
+       if (!schema) {
+               return WERR_DS_SCHEMA_NOT_LOADED;
+       }
+
+       for (i=0, cur = first_object; cur; cur = cur->next_object, i++) {
+               status = dsdb_origin_object_convert(ldb, schema, cur,
+                                                   objects, &objects[i]);
+               if (!W_ERROR_IS_OK(status)) {
+                       goto cancel;
+               }
+       }
+
+       ids = talloc_array(mem_ctx,
+                          struct drsuapi_DsReplicaObjectIdentifier2,
+                          num_objects);
+       if (ids == NULL) {
+               status = WERR_NOMEM;
+               goto cancel;
+       }
+
+       for (i=0; i < num_objects; i++) {
+               struct dom_sid *sid = NULL;
+               struct ldb_request *add_req;
+
+               DEBUG(6,(__location__ ": adding %s\n", 
+                        ldb_dn_get_linearized(objects[i]->dn)));
+
+               ret = ldb_build_add_req(&add_req,
+                                       ldb,
+                                       objects,
+                                       objects[i],
+                                       NULL,
+                                       NULL,
+                                       ldb_op_default_callback,
+                                       NULL);
+               if (ret != LDB_SUCCESS) {
+                       status = WERR_DS_INTERNAL_FAILURE;
+                       goto cancel;
+               }
+
+               ret = ldb_request_add_control(add_req, LDB_CONTROL_RELAX_OID, true, NULL);
+               if (ret != LDB_SUCCESS) {
+                       status = WERR_DS_INTERNAL_FAILURE;
+                       goto cancel;
+               }
+               
+               ret = ldb_request(ldb, add_req);
+               if (ret == LDB_SUCCESS) {
+                       ret = ldb_wait(add_req->handle, LDB_WAIT_ALL);
+               }
+               if (ret != LDB_SUCCESS) {
+                       DEBUG(0,(__location__ ": Failed add of %s - %s\n",
+                                ldb_dn_get_linearized(objects[i]->dn), ldb_errstring(ldb)));
+                       status = WERR_DS_INTERNAL_FAILURE;
+                       goto cancel;
+               }
+
+               talloc_free(add_req);
+
+               ret = ldb_search(ldb, objects, &res, objects[i]->dn,
+                                LDB_SCOPE_BASE, attrs,
+                                "(objectClass=*)");
+               if (ret != LDB_SUCCESS) {
+                       status = WERR_DS_INTERNAL_FAILURE;
+                       goto cancel;
+               }
+               ids[i].guid = samdb_result_guid(res->msgs[0], "objectGUID");
+               sid = samdb_result_dom_sid(objects, res->msgs[0], "objectSid");
+               if (sid) {
+                       ids[i].sid = *sid;
+               } else {
+                       ZERO_STRUCT(ids[i].sid);
+               }
+       }
+
+       ret = ldb_transaction_commit(ldb);
+       if (ret != LDB_SUCCESS) {
+               return WERR_DS_INTERNAL_FAILURE;
+       }
+
+       talloc_free(objects);
+
+       *_num = num_objects;
+       *_ids = ids;
+       return WERR_OK;
+
+cancel:
+       talloc_free(objects);
+       ldb_transaction_cancel(ldb);
+       return status;
+}