repl_meta_data.c

samba最新软件

	rmd = ar->objs->objects[ar->index_current].meta_data;
	ZERO_STRUCT(omd);
	omd.version = 1;

	/*
	 * TODO: add rename conflict handling
	 */
	if (ldb_dn_compare(msg->dn, ar->sub.search_msg->dn) != 0) {
		ldb_debug_set(ar->module->ldb, LDB_DEBUG_FATAL, "replmd_replicated_apply_merge[%u]: rename not supported",
			      ar->index_current);
		ldb_debug(ar->module->ldb, LDB_DEBUG_FATAL, "%s => %s\n",
			  ldb_dn_get_linearized(ar->sub.search_msg->dn),
			  ldb_dn_get_linearized(msg->dn));
		return replmd_replicated_request_werror(ar, WERR_NOT_SUPPORTED);
	}

	ret = ldb_sequence_number(ar->module->ldb, LDB_SEQ_NEXT, &seq_num);
	if (ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ret);
	}

	/* find existing meta data */
	omd_value = ldb_msg_find_ldb_val(ar->sub.search_msg, "replPropertyMetaData");
	if (omd_value) {
		ndr_err = ndr_pull_struct_blob(omd_value, ar->sub.mem_ctx, 
					       lp_iconv_convenience(ldb_get_opaque(ar->module->ldb, "loadparm")), &omd,
					       (ndr_pull_flags_fn_t)ndr_pull_replPropertyMetaDataBlob);
		if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
			NTSTATUS nt_status = ndr_map_error2ntstatus(ndr_err);
			return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
		}

		if (omd.version != 1) {
			return replmd_replicated_request_werror(ar, WERR_DS_DRA_INTERNAL_ERROR);
		}
	}

	ZERO_STRUCT(nmd);
	nmd.version = 1;
	nmd.ctr.ctr1.count = omd.ctr.ctr1.count + rmd->ctr.ctr1.count;
	nmd.ctr.ctr1.array = talloc_array(ar->sub.mem_ctx,
					  struct replPropertyMetaData1,
					  nmd.ctr.ctr1.count);
	if (!nmd.ctr.ctr1.array) return replmd_replicated_request_werror(ar, WERR_NOMEM);

	/* first copy the old meta data */
	for (i=0; i < omd.ctr.ctr1.count; i++) {
		nmd.ctr.ctr1.array[ni]	= omd.ctr.ctr1.array[i];
		ni++;
	}

	/* now merge in the new meta data */
	for (i=0; i < rmd->ctr.ctr1.count; i++) {
		bool found = false;

		rmd->ctr.ctr1.array[i].local_usn = seq_num;

		for (j=0; j < ni; j++) {
			int cmp;

			if (rmd->ctr.ctr1.array[i].attid != nmd.ctr.ctr1.array[j].attid) {
				continue;
			}

			cmp = replmd_replPropertyMetaData1_conflict_compare(&rmd->ctr.ctr1.array[i],
									    &nmd.ctr.ctr1.array[j]);
			if (cmp > 0) {
				/* replace the entry */
				nmd.ctr.ctr1.array[j] = rmd->ctr.ctr1.array[i];
				found = true;
				break;
			}

			/* we don't want to apply this change so remove the attribute */
			ldb_msg_remove_element(msg, &msg->elements[i-removed_attrs]);
			removed_attrs++;

			found = true;
			break;
		}

		if (found) continue;

		nmd.ctr.ctr1.array[ni] = rmd->ctr.ctr1.array[i];
		ni++;
	}

	/*
	 * finally correct the size of the meta_data array
	 */
	nmd.ctr.ctr1.count = ni;

	/*
	 * the rdn attribute (the alias for the name attribute),
	 * 'cn' for most objects is the last entry in the meta data array
	 * we have stored
	 *
	 * sort the new meta data array
	 */
	{
		struct replPropertyMetaData1 *rdn_p;
		uint32_t rdn_idx = omd.ctr.ctr1.count - 1;

		rdn_p = &nmd.ctr.ctr1.array[rdn_idx];
		replmd_replPropertyMetaDataCtr1_sort(&nmd.ctr.ctr1, &rdn_p->attid);
	}

	/* create the meta data value */
	ndr_err = ndr_push_struct_blob(&nmd_value, msg, 
				       lp_iconv_convenience(ldb_get_opaque(ar->module->ldb, "loadparm")),
				       &nmd,
				       (ndr_push_flags_fn_t)ndr_push_replPropertyMetaDataBlob);
	if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
		NTSTATUS nt_status = ndr_map_error2ntstatus(ndr_err);
		return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
	}

	/*
	 * check if some replicated attributes left, otherwise skip the ldb_modify() call
	 */
	if (msg->num_elements == 0) {
		ldb_debug(ar->module->ldb, LDB_DEBUG_TRACE, "replmd_replicated_apply_merge[%u]: skip replace\n",
			  ar->index_current);
		goto next_object;
	}

	ldb_debug(ar->module->ldb, LDB_DEBUG_TRACE, "replmd_replicated_apply_merge[%u]: replace %u attributes\n",
		  ar->index_current, msg->num_elements);

	/*
	 * when we now that we'll modify the record, add the whenChanged, uSNChanged
	 * and replPopertyMetaData attributes
	 */
	ret = ldb_msg_add_string(msg, "whenChanged", ar->objs->objects[ar->index_current].when_changed);
	if (ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ret);
	}
	ret = samdb_msg_add_uint64(ar->module->ldb, msg, msg, "uSNChanged", seq_num);
	if (ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ret);
	}
	ret = ldb_msg_add_value(msg, "replPropertyMetaData", &nmd_value, NULL);
	if (ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ret);
	}

	replmd_ldb_message_sort(msg, ar->schema);

	/* we want to replace the old values */
	for (i=0; i < msg->num_elements; i++) {
		msg->elements[i].flags = LDB_FLAG_MOD_REPLACE;
	}

	ret = ldb_build_mod_req(&ar->sub.change_req,
				ar->module->ldb,
				ar->sub.mem_ctx,
				msg,
				NULL,
				ar,
				replmd_replicated_apply_merge_callback);
	if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);

#ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */ 
	return ldb_next_request(ar->module, ar->sub.change_req);
#else
	ret = ldb_next_request(ar->module, ar->sub.change_req);
	if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);

	ar->sub.change_ret = ldb_wait(ar->sub.change_req->handle, LDB_WAIT_ALL);
	if (ar->sub.change_ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ar->sub.change_ret);
	}

next_object:
	talloc_free(ar->sub.mem_ctx);
	ZERO_STRUCT(ar->sub);

	ar->index_current++;

	return LDB_SUCCESS;
#endif
}

static int replmd_replicated_apply_search_callback(struct ldb_context *ldb,
						   void *private_data,
						   struct ldb_reply *ares)
{
	struct replmd_replicated_request *ar = talloc_get_type(private_data,
					       struct replmd_replicated_request);
	bool is_done = false;

	switch (ares->type) {
	case LDB_REPLY_ENTRY:
		ar->sub.search_msg = talloc_steal(ar->sub.mem_ctx, ares->message);
		break;
	case LDB_REPLY_REFERRAL:
		/* we ignore referrals */
		break;
	case LDB_REPLY_EXTENDED:
	case LDB_REPLY_DONE:
		is_done = true;
	}

	talloc_free(ares);

#ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */ 
	if (is_done) {
		ar->sub.search_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
		if (ar->sub.search_ret != LDB_SUCCESS) {
			return replmd_replicated_request_error(ar, ar->sub.search_ret);
		}
		if (ar->sub.search_msg) {
			return replmd_replicated_apply_merge(ar);
		}
		return replmd_replicated_apply_add(ar);
	}
#endif
	return LDB_SUCCESS;
}

static int replmd_replicated_apply_search(struct replmd_replicated_request *ar)
{
	int ret;
	char *tmp_str;
	char *filter;

	tmp_str = ldb_binary_encode(ar->sub.mem_ctx, ar->objs->objects[ar->index_current].guid_value);
	if (!tmp_str) return replmd_replicated_request_werror(ar, WERR_NOMEM);

	filter = talloc_asprintf(ar->sub.mem_ctx, "(objectGUID=%s)", tmp_str);
	if (!filter) return replmd_replicated_request_werror(ar, WERR_NOMEM);
	talloc_free(tmp_str);

	ret = ldb_build_search_req(&ar->sub.search_req,
				   ar->module->ldb,
				   ar->sub.mem_ctx,
				   ar->objs->partition_dn,
				   LDB_SCOPE_SUBTREE,
				   filter,
				   NULL,
				   NULL,
				   ar,
				   replmd_replicated_apply_search_callback);
	if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);

#ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */ 
	return ldb_next_request(ar->module, ar->sub.search_req);
#else
	ret = ldb_next_request(ar->module, ar->sub.search_req);
	if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);

	ar->sub.search_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
	if (ar->sub.search_ret != LDB_SUCCESS && ar->sub.search_ret != LDB_ERR_NO_SUCH_OBJECT) {
		return replmd_replicated_request_error(ar, ar->sub.search_ret);
	}
	if (ar->sub.search_msg) {
		return replmd_replicated_apply_merge(ar);
	}

	return replmd_replicated_apply_add(ar);
#endif
}

static int replmd_replicated_apply_next(struct replmd_replicated_request *ar)
{
#ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */ 
	if (ar->index_current >= ar->objs->num_objects) {
		return replmd_replicated_uptodate_vector(ar);
	}
#endif

	ar->sub.mem_ctx = talloc_new(ar);
	if (!ar->sub.mem_ctx) return replmd_replicated_request_werror(ar, WERR_NOMEM);

	return replmd_replicated_apply_search(ar);
}

static int replmd_replicated_uptodate_modify_callback(struct ldb_context *ldb,
						      void *private_data,
						      struct ldb_reply *ares)
{
#ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */ 
	struct replmd_replicated_request *ar = talloc_get_type(private_data,
					       struct replmd_replicated_request);

	ar->sub.change_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
	if (ar->sub.change_ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ar->sub.change_ret);
	}

	talloc_free(ar->sub.mem_ctx);
	ZERO_STRUCT(ar->sub);

	return replmd_replicated_request_done(ar);
#else
	return LDB_SUCCESS;
#endif
}

static int replmd_drsuapi_DsReplicaCursor2_compare(const struct drsuapi_DsReplicaCursor2 *c1,
						   const struct drsuapi_DsReplicaCursor2 *c2)
{
	return GUID_compare(&c1->source_dsa_invocation_id, &c2->source_dsa_invocation_id);
}

static int replmd_replicated_uptodate_modify(struct replmd_replicated_request *ar)
{
	enum ndr_err_code ndr_err;
	struct ldb_message *msg;
	struct replUpToDateVectorBlob ouv;
	const struct ldb_val *ouv_value;
	const struct drsuapi_DsReplicaCursor2CtrEx *ruv;
	struct replUpToDateVectorBlob nuv;
	struct ldb_val nuv_value;
	struct ldb_message_element *nuv_el = NULL;
	const struct GUID *our_invocation_id;
	struct ldb_message_element *orf_el = NULL;
	struct repsFromToBlob nrf;
	struct ldb_val *nrf_value = NULL;
	struct ldb_message_element *nrf_el = NULL;
	uint32_t i,j,ni=0;
	uint64_t seq_num;
	bool found = false;
	time_t t = time(NULL);
	NTTIME now;
	int ret;

	ruv = ar->objs->uptodateness_vector;
	ZERO_STRUCT(ouv);
	ouv.version = 2;
	ZERO_STRUCT(nuv);
	nuv.version = 2;

	unix_to_nt_time(&now, t);

	/* 
	 * we use the next sequence number for our own highest_usn
	 * because we will do a modify request and this will increment
	 * our highest_usn
	 */
	ret = ldb_sequence_number(ar->module->ldb, LDB_SEQ_NEXT, &seq_num);
	if (ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ret);
	}

	/*
	 * first create the new replUpToDateVector
	 */
	ouv_value = ldb_msg_find_ldb_val(ar->sub.search_msg, "replUpToDateVector");
	if (ouv_value) {
		ndr_err = ndr_pull_struct_blob(ouv_value, ar->sub.mem_ctx, 
					       lp_iconv_convenience(ldb_get_opaque(ar->module->ldb, "loadparm")), &ouv,
					       (ndr_pull_flags_fn_t)ndr_pull_replUpToDateVectorBlob);
		if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
			NTSTATUS nt_status = ndr_map_error2ntstatus(ndr_err);
			return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
		}

		if (ouv.version != 2) {
			return replmd_replicated_request_werror(ar, WERR_DS_DRA_INTERNAL_ERROR);
		}
	}

	/*
	 * the new uptodateness vector will at least
	 * contain 1 entry, one for the source_dsa
	 *
	 * plus optional values from our old vector and the one from the source_dsa
	 */
	nuv.ctr.ctr2.count = 1 + ouv.ctr.ctr2.count;
	if (ruv) nuv.ctr.ctr2.count += ruv->count;
	nuv.ctr.ctr2.cursors = talloc_array(ar->sub.mem_ctx,
					    struct drsuapi_DsReplicaCursor2,
					    nuv.ctr.ctr2.count);
	if (!nuv.ctr.ctr2.cursors) return replmd_replicated_request_werror(ar, WERR_NOMEM);

	/* first copy the old vector */
	for (i=0; i < ouv.ctr.ctr2.count; i++) {
		nuv.ctr.ctr2.cursors[ni] = ouv.ctr.ctr2.cursors[i];
		ni++;
	}

	/* get our invocation_id if we have one already attached to the ldb */
	our_invocation_id = samdb_ntds_invocation_id(ar->module->ldb);

	/* merge in the source_dsa vector is available */
	for (i=0; (ruv && i < ruv->count); i++) {
		found = false;

		if (our_invocation_id &&
		    GUID_equal(&ruv->cursors[i].source_dsa_invocation_id,
			       our_invocation_id)) {
			continue;
		}

		for (j=0; j < ni; j++) {
			if (!GUID_equal(&ruv->cursors[i].source_dsa_invocation_id,
					&nuv.ctr.ctr2.cursors[j].source_dsa_invocation_id)) {