s4:dsdb rewrite the linked_atrributes code to commit in the end_transaction hook

linked attribute changes can come in any order. This means it is
possible for a forward link to come over the wire in DRS before the
target even exists. To make this work this patch changed the linked
attributes module to gather up all the changes it needs to make in a
linked list, then execute the changes in the end_transaction hook for
the module.

During that commit phase we also fix up all the DNs that we got by
searching for their GUID, as the objects may have moved after the
linked attribute was sent, but before the end of the transaction

1 files changed, 281 insertions, 107 deletions

diff --git a/source4/dsdb/samdb/ldb_modules/linked_attributes.c b/source4/dsdb/samdb/ldb_modules/linked_attributes.c
index d83f6b9269..561fc66941 100644
--- a/source4/dsdb/samdb/ldb_modules/linked_attributes.c
+++ b/source4/dsdb/samdb/ldb_modules/linked_attributes.c
@@ -33,6 +33,10 @@
 #include "dlinklist.h"
 #include "dsdb/samdb/samdb.h"
 
+struct la_private {
+	struct la_context *la_list;
+};
+
 struct la_op_store {
 	struct la_op_store *next;
 	struct la_op_store *prev;
@@ -49,9 +53,11 @@ struct replace_context {
 };
 
 struct la_context {
+	struct la_context *next, *prev;
 	const struct dsdb_schema *schema;
 	struct ldb_module *module;
 	struct ldb_request *req;
+	struct ldb_dn *partition_dn;
 	struct ldb_dn *add_dn;
 	struct ldb_dn *del_dn;
 	struct replace_context *rc;
@@ -65,6 +71,7 @@ static struct la_context *linked_attributes_init(struct ldb_module *module,
 {
 	struct ldb_context *ldb;
 	struct la_context *ac;
+	const struct ldb_control *partition_ctrl;
 
 	ldb = ldb_module_get_ctx(module);
 
@@ -78,6 +85,17 @@ static struct la_context *linked_attributes_init(struct ldb_module *module,
 	ac->module = module;
 	ac->req = req;
 
+	/* remember the partition DN that came in, if given */
+	partition_ctrl = ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID);
+	if (partition_ctrl) {
+		const struct dsdb_control_current_partition *partition;
+		partition = talloc_get_type(partition_ctrl->data,
+					    struct dsdb_control_current_partition);
+		SMB_ASSERT(partition && partition->version == DSDB_CONTROL_CURRENT_PARTITION_VERSION);
+	
+		ac->partition_dn = ldb_dn_copy(ac, partition->dn);
+	}
+
 	return ac;
 }
 
@@ -130,9 +148,7 @@ static int la_store_op(struct la_context *ac,
 
 static int la_op_search_callback(struct ldb_request *req,
 				 struct ldb_reply *ares);
-static int la_do_mod_request(struct la_context *ac);
-static int la_mod_callback(struct ldb_request *req,
-			   struct ldb_reply *ares);
+static int la_queue_mod_request(struct la_context *ac);
 static int la_down_req(struct la_context *ac);
 
 
@@ -328,7 +344,7 @@ static int la_mod_search_callback(struct ldb_request *req, struct ldb_reply *are
 
 		if (ac->req->operation == LDB_ADD) {
 			/* Start the modifies to the backlinks */
-			ret = la_do_mod_request(ac);
+			ret = la_queue_mod_request(ac);
 
 			if (ret != LDB_SUCCESS) {
 				return ldb_module_done(ac->req, NULL, NULL,
@@ -745,7 +761,7 @@ static int la_op_search_callback(struct ldb_request *req,
 
 		case LDB_RENAME:	
 			/* start the mod requests chain */
-			ret = la_do_mod_request(ac);
+			ret = la_queue_mod_request(ac);
 			if (ret != LDB_SUCCESS) {
 				return ldb_module_done(ac->req, NULL, NULL,
 						       ret);
@@ -765,112 +781,23 @@ static int la_op_search_callback(struct ldb_request *req,
 	return LDB_SUCCESS;
 }
 
-/* do a linked attributes modify request */
-static int la_do_mod_request(struct la_context *ac)
+/* queue a linked attributes modify request in the la_private
+   structure */
+static int la_queue_mod_request(struct la_context *ac)
 {
-	struct ldb_message_element *ret_el;
-	struct ldb_request *mod_req;
-	struct ldb_message *new_msg;
-	struct ldb_context *ldb;
-	int ret;
-
-	/* If we have no modifies in the queue, we are done! */
-	if (!ac->ops) {
-		return ldb_module_done(ac->req, ac->op_controls,
-				       ac->op_response, LDB_SUCCESS);
-	}
+	struct la_private *la_private = 
+		talloc_get_type(ldb_module_get_private(ac->module), struct la_private);
 
-	ldb = ldb_module_get_ctx(ac->module);
-
-	/* Create the modify request */
-	new_msg = ldb_msg_new(ac);
-	if (!new_msg) {
-		ldb_oom(ldb);
-		return LDB_ERR_OPERATIONS_ERROR;
-	}
-	new_msg->dn = ac->ops->dn;
-
-	if (ac->ops->op == LA_OP_ADD) {
-		ret = ldb_msg_add_empty(new_msg, ac->ops->name,
-					LDB_FLAG_MOD_ADD, &ret_el);
-	} else {
-		ret = ldb_msg_add_empty(new_msg, ac->ops->name,
-					LDB_FLAG_MOD_DELETE, &ret_el);
-	}
-	if (ret != LDB_SUCCESS) {
-		return ret;
-	}
-	ret_el->values = talloc_array(new_msg, struct ldb_val, 1);
-	if (!ret_el->values) {
-		ldb_oom(ldb);
+	if (la_private == NULL) {
+		ldb_debug(ldb_module_get_ctx(ac->module), LDB_DEBUG_ERROR, __location__ ": No la_private transaction setup\n");
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
-	ret_el->num_values = 1;
-	if (ac->ops->op == LA_OP_ADD) {
-		ret_el->values[0] = data_blob_string_const(ldb_dn_get_extended_linearized(new_msg, ac->add_dn, 1));
-	} else {
-		ret_el->values[0] = data_blob_string_const(ldb_dn_get_extended_linearized(new_msg, ac->del_dn, 1));
-	}
 
-#if 0
-	ldb_debug(ldb, LDB_DEBUG_WARNING,
-		  "link on %s %s: %s %s\n", 
-		  ldb_dn_get_linearized(new_msg->dn), ret_el->name, 
-		  ret_el->values[0].data, ac->ops->op == LA_OP_ADD ? "added" : "deleted");
-#endif	
+	talloc_steal(la_private, ac);
+	DLIST_ADD(la_private->la_list, ac);
 
-	/* use ac->ops as the mem_ctx so that the request will be freed
-	 * in the callback as soon as completed */
-	ret = ldb_build_mod_req(&mod_req, ldb, ac->ops,
-				new_msg,
-				NULL,
-				ac, la_mod_callback,
-				ac->req);
-	if (ret != LDB_SUCCESS) {
-		return ret;
-	}
-	talloc_steal(mod_req, new_msg);
-
-	/* Run the new request */
-	return ldb_next_request(ac->module, mod_req);
-}
-
-static int la_mod_callback(struct ldb_request *req, struct ldb_reply *ares)
-{
-	struct la_context *ac;
-	struct ldb_context *ldb;
-	struct la_op_store *os;
-
-	ac = talloc_get_type(req->context, struct la_context);
-	ldb = ldb_module_get_ctx(ac->module);
-
-	if (!ares) {
-		return ldb_module_done(ac->req, NULL, NULL,
-					LDB_ERR_OPERATIONS_ERROR);
-	}
-	if (ares->error != LDB_SUCCESS) {
-		return ldb_module_done(ac->req, ares->controls,
-					ares->response, ares->error);
-	}
-
-	if (ares->type != LDB_REPLY_DONE) {
-		ldb_set_errstring(ldb,
-				  "invalid ldb_reply_type in callback");
-		talloc_free(ares);
-		return ldb_module_done(ac->req, NULL, NULL,
-					LDB_ERR_OPERATIONS_ERROR);
-	}
-
-	talloc_free(ares);
-
-	os = ac->ops;
-	DLIST_REMOVE(ac->ops, os);
-
-	/* this frees the request too
-	 * DO NOT access 'req' after this point */
-	talloc_free(os);
-
-	return la_do_mod_request(ac);
+	return ldb_module_done(ac->req, ac->op_controls,
+			       ac->op_response, LDB_SUCCESS);
 }
 
 /* Having done the original operation, then try to fix up all the linked attributes for modify and delete */
@@ -904,14 +831,14 @@ static int la_mod_del_callback(struct ldb_request *req, struct ldb_reply *ares)
 	ac->op_response = talloc_steal(ac, ares->response);
 
 	/* If we have modfies to make, this is the time to do them for modify and delete */
-	ret = la_do_mod_request(ac);
+	ret = la_queue_mod_request(ac);
 	
 	if (ret != LDB_SUCCESS) {
 		return ldb_module_done(ac->req, NULL, NULL, ret);
 	}
 	talloc_free(ares);
 
-	/* la_do_mod_request has already sent the callbacks */
+	/* la_queue_mod_request has already sent the callbacks */
 	return LDB_SUCCESS;
 
 }
@@ -1094,6 +1021,250 @@ static int la_down_req(struct la_context *ac)
 	return ldb_next_request(ac->module, down_req);
 }
 
+/*
+  use the GUID part of an extended DN to find the target DN, in case
+  it has moved
+ */
+static int la_find_dn_target(struct ldb_module *module, struct la_context *ac, struct ldb_dn **dn)
+{
+	const struct ldb_val *guid;
+	struct ldb_context *ldb;
+	int ret;
+	struct ldb_result *res;
+	const char *attrs[] = { NULL };
+	struct ldb_request *search_req;
+	char *expression;
+	struct ldb_search_options_control *options;
+
+	ldb = ldb_module_get_ctx(ac->module);
+
+	guid = ldb_dn_get_extended_component(*dn, "GUID");
+	if (guid == NULL) {
+		return LDB_SUCCESS;
+	}
+
+	expression = talloc_asprintf(ac, "objectGUID=%s", ldb_binary_encode(ac, *guid));
+	if (!expression) {
+		ldb_oom(ldb);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	res = talloc_zero(ac, struct ldb_result);
+	if (!res) {
+		ldb_oom(ldb);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ret = ldb_build_search_req(&search_req, ldb, ac,
+				   ldb_get_default_basedn(ldb),
+				   LDB_SCOPE_SUBTREE,
+				   expression, attrs,
+				   NULL,
+				   res, ldb_search_default_callback,
+				   NULL);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+
+	/* we need to cope with cross-partition links, so search for
+	   the GUID over all partitions */
+	options = talloc(search_req, struct ldb_search_options_control);
+	if (options == NULL) {
+		ldb_oom(ldb);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	options->search_options = LDB_SEARCH_OPTION_PHANTOM_ROOT;
+
+	ret = ldb_request_add_control(search_req,
+				      LDB_CONTROL_SEARCH_OPTIONS_OID,
+				      true, options);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+
+	ret = ldb_next_request(module, search_req);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+
+	ret = ldb_wait(search_req->handle, LDB_WAIT_ALL);
+	if (ret != LDB_SUCCESS) {
+		ldb_debug(ldb, LDB_DEBUG_ERROR, "GUID search failed (%s) for %s\n", 
+			  ldb_errstring(ldb), ldb_dn_get_extended_linearized(ac, *dn, 1));
+		return ret;
+	}
+
+	/* this really should be exactly 1, but there is a bug in the
+	   partitions module that can return two here with the
+	   search_options control set */
+	if (res->count < 1) {
+		ldb_debug(ldb, LDB_DEBUG_ERROR, "GUID search gave count=%d for %s\n", 
+			  res->count, ldb_dn_get_extended_linearized(ac, *dn, 1));
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	*dn = res->msgs[0]->dn;
+
+	return LDB_SUCCESS;
+}
+
+/* apply one la_context op change */
+static int la_do_op_request(struct ldb_module *module, struct la_context *ac, struct la_op_store *op)
+{
+	struct ldb_message_element *ret_el;
+	struct ldb_request *mod_req;
+	struct ldb_message *new_msg;
+	struct ldb_context *ldb;
+	int ret;
+
+	ldb = ldb_module_get_ctx(ac->module);
+
+	/* Create the modify request */
+	new_msg = ldb_msg_new(ac);
+	if (!new_msg) {
+		ldb_oom(ldb);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	new_msg->dn = op->dn;
+	ret = la_find_dn_target(module, ac, &new_msg->dn);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+
+	if (op->op == LA_OP_ADD) {
+		ret = ldb_msg_add_empty(new_msg, op->name,
+					LDB_FLAG_MOD_ADD, &ret_el);
+	} else {
+		ret = ldb_msg_add_empty(new_msg, op->name,
+					LDB_FLAG_MOD_DELETE, &ret_el);
+	}
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+	ret_el->values = talloc_array(new_msg, struct ldb_val, 1);
+	if (!ret_el->values) {
+		ldb_oom(ldb);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	ret_el->num_values = 1;
+	if (op->op == LA_OP_ADD) {
+		ret_el->values[0] = data_blob_string_const(ldb_dn_get_extended_linearized(new_msg, ac->add_dn, 1));
+	} else {
+		ret_el->values[0] = data_blob_string_const(ldb_dn_get_extended_linearized(new_msg, ac->del_dn, 1));
+	}
+
+#if 0
+	ldb_debug(ldb, LDB_DEBUG_WARNING,
+		  "link on %s %s: %s %s\n", 
+		  ldb_dn_get_linearized(new_msg->dn), ret_el->name, 
+		  ret_el->values[0].data, ac->ops->op == LA_OP_ADD ? "added" : "deleted");
+#endif	
+
+	ret = ldb_build_mod_req(&mod_req, ldb, op,
+				new_msg,
+				NULL,
+				NULL, 
+				ldb_op_default_callback,
+				NULL);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+	talloc_steal(mod_req, new_msg);
+
+	if (DEBUGLVL(4)) {
+		DEBUG(4,("Applying linked attribute change:\n%s\n",
+			 ldb_ldif_message_string(ldb, op, LDB_CHANGETYPE_MODIFY, new_msg)));
+	}
+
+	/* Run the new request */
+	ret = ldb_next_request(module, mod_req);
+
+	/* we need to wait for this to finish, as we are being called
+	   from the synchronous end_transaction hook of this module */
+	if (ret == LDB_SUCCESS) {
+		ret = ldb_wait(mod_req->handle, LDB_WAIT_ALL);
+	}
+
+	if (ret != LDB_SUCCESS) {
+		ldb_debug(ldb, LDB_DEBUG_WARNING, "Failed to apply linked attribute change '%s' %s\n",
+			  ldb_errstring(ldb),
+			  ldb_ldif_message_string(ldb, op, LDB_CHANGETYPE_MODIFY, new_msg));
+	}
+
+	return ret;
+}
+
+/* apply one set of la_context changes */
+static int la_do_mod_request(struct ldb_module *module, struct la_context *ac)
+{
+	struct la_op_store *op;
+
+	for (op = ac->ops; op; op=op->next) {
+		int ret = la_do_op_request(module, ac, op);
+		if (ret != LDB_SUCCESS) {
+			return ret;
+		}
+	}
+
+	return LDB_SUCCESS;
+}
+
+
+/*
+  we hook into the transaction operations to allow us to 
+  perform the linked attribute updates at the end of the whole
+  transaction. This allows a forward linked attribute to be created
+  before the target is created, as long as the target is created
+  in the same transaction
+ */
+static int linked_attributes_start_transaction(struct ldb_module *module)
+{
+	/* create our private structure for this transaction */
+	struct la_private *la_private = talloc_get_type(ldb_module_get_private(module),
+							struct la_private);
+	talloc_free(la_private);
+	la_private = talloc(module, struct la_private);
+	if (la_private == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	la_private->la_list = NULL;
+	ldb_module_set_private(module, la_private);
+	return LDB_SUCCESS;
+}
+
+/*
+  on end transaction we loop over our queued la_context structures and
+  apply each of them  
+ */
+static int linked_attributes_end_transaction(struct ldb_module *module)
+{
+	struct la_private *la_private = 
+		talloc_get_type(ldb_module_get_private(module), struct la_private);
+	struct la_context *ac;
+
+	for (ac=la_private->la_list; ac; ac=ac->next) {
+		int ret;
+		ac->req = NULL;
+		ret = la_do_mod_request(module, ac);
+		if (ret != LDB_SUCCESS) {
+			ret = la_do_mod_request(module, ac);
+			return ret;
+		}
+	}
+	
+	return LDB_SUCCESS;
+}
+
+static int linked_attributes_del_transaction(struct ldb_module *module)
+{
+	struct la_private *la_private = 
+		talloc_get_type(ldb_module_get_private(module), struct la_private);
+	talloc_free(la_private);
+	ldb_module_set_private(module, NULL);
+	return LDB_SUCCESS;
+}
+
 
 _PUBLIC_ const struct ldb_module_ops ldb_linked_attributes_module_ops = {
 	.name		   = "linked_attributes",
@@ -1101,4 +1272,7 @@ _PUBLIC_ const struct ldb_module_ops ldb_linked_attributes_module_ops = {
 	.modify            = linked_attributes_modify,
 	.del               = linked_attributes_del,
 	.rename            = linked_attributes_rename,
+	.start_transaction = linked_attributes_start_transaction,
+	.end_transaction   = linked_attributes_end_transaction,
+	.del_transaction   = linked_attributes_del_transaction,
 };