diff options
Diffstat (limited to 'source4/dsdb')
| -rw-r--r-- | source4/dsdb/common/util.c | 146 | ||||
| -rw-r--r-- | source4/dsdb/config.mk | 3 | ||||
| -rw-r--r-- | source4/dsdb/repl/drepl_notify.c | 428 | ||||
| -rw-r--r-- | source4/dsdb/repl/drepl_out_helpers.c | 4 | ||||
| -rw-r--r-- | source4/dsdb/repl/drepl_out_pull.c | 3 | ||||
| -rw-r--r-- | source4/dsdb/repl/drepl_periodic.c | 1 | ||||
| -rw-r--r-- | source4/dsdb/repl/drepl_service.c | 10 | ||||
| -rw-r--r-- | source4/dsdb/repl/drepl_service.h | 37 | ||||
| -rw-r--r-- | source4/dsdb/samdb/ldb_modules/partition.c | 19 | ||||
| -rw-r--r-- | source4/dsdb/samdb/ldb_modules/repl_meta_data.c | 372 | ||||
| -rw-r--r-- | source4/dsdb/samdb/ldb_modules/samba3sam.c | 1 |
11 files changed, 992 insertions, 32 deletions
diff --git a/source4/dsdb/common/util.c b/source4/dsdb/common/util.c index b5005444cc..c2636e127f 100644 --- a/source4/dsdb/common/util.c +++ b/source4/dsdb/common/util.c @@ -2257,3 +2257,149 @@ failed: } +/* + load the uSNHighest attribute from the @REPLCHANGED object for a + partition + */ +int dsdb_load_partition_usn(struct ldb_context *ldb, struct ldb_dn *dn, uint64_t *uSN) +{ + struct ldb_request *req; + int ret; + TALLOC_CTX *tmp_ctx = talloc_new(ldb); + struct dsdb_control_current_partition *p_ctrl; + struct ldb_result *res; + + res = talloc_zero(tmp_ctx, struct ldb_result); + if (!res) { + talloc_free(tmp_ctx); + return LDB_ERR_OPERATIONS_ERROR; + } + + ret = ldb_build_search_req(&req, ldb, tmp_ctx, + ldb_dn_new(tmp_ctx, ldb, "@REPLCHANGED"), + LDB_SCOPE_BASE, + NULL, NULL, + NULL, + res, ldb_search_default_callback, + NULL); + if (ret != LDB_SUCCESS) { + talloc_free(tmp_ctx); + return ret; + } + + p_ctrl = talloc(req, struct dsdb_control_current_partition); + if (p_ctrl == NULL) { + talloc_free(res); + return LDB_ERR_OPERATIONS_ERROR; + } + p_ctrl->version = DSDB_CONTROL_CURRENT_PARTITION_VERSION; + p_ctrl->dn = dn; + + + ret = ldb_request_add_control(req, + DSDB_CONTROL_CURRENT_PARTITION_OID, + false, p_ctrl); + if (ret != LDB_SUCCESS) { + talloc_free(tmp_ctx); + return ret; + } + + /* Run the new request */ + ret = ldb_request(ldb, req); + + if (ret == LDB_SUCCESS) { + ret = ldb_wait(req->handle, LDB_WAIT_ALL); + } + + if (ret != LDB_SUCCESS) { + talloc_free(tmp_ctx); + return ret; + } + + if (res->count < 1) { + *uSN = 0; + } else { + *uSN = ldb_msg_find_attr_as_uint64(res->msgs[0], "uSNHighest", 0); + } + + talloc_free(tmp_ctx); + + return LDB_SUCCESS; +} + +/* + save the uSNHighest attribute in the @REPLCHANGED object for a + partition + */ +int dsdb_save_partition_usn(struct ldb_context *ldb, struct ldb_dn *dn, uint64_t uSN) +{ + struct ldb_request *req; + struct ldb_message *msg; + struct dsdb_control_current_partition *p_ctrl; + int ret; + + msg = ldb_msg_new(ldb); + if (msg == NULL) { + return LDB_ERR_OPERATIONS_ERROR; + } + + msg->dn = ldb_dn_new(msg, ldb, "@REPLCHANGED"); + if (msg->dn == NULL) { + talloc_free(msg); + return LDB_ERR_OPERATIONS_ERROR; + } + + ret = ldb_msg_add_fmt(msg, "uSNHighest", "%llu", (unsigned long long)uSN); + if (ret != LDB_SUCCESS) { + talloc_free(msg); + return ret; + } + msg->elements[0].flags = LDB_FLAG_MOD_REPLACE; + + + p_ctrl = talloc(msg, struct dsdb_control_current_partition); + if (p_ctrl == NULL) { + talloc_free(msg); + return LDB_ERR_OPERATIONS_ERROR; + } + p_ctrl->version = DSDB_CONTROL_CURRENT_PARTITION_VERSION; + p_ctrl->dn = dn; + + ret = ldb_build_mod_req(&req, ldb, msg, + msg, + NULL, + NULL, ldb_op_default_callback, + NULL); +again: + if (ret != LDB_SUCCESS) { + talloc_free(msg); + return ret; + } + + ret = ldb_request_add_control(req, + DSDB_CONTROL_CURRENT_PARTITION_OID, + false, p_ctrl); + if (ret != LDB_SUCCESS) { + talloc_free(msg); + return ret; + } + + /* Run the new request */ + ret = ldb_request(ldb, req); + + if (ret == LDB_SUCCESS) { + ret = ldb_wait(req->handle, LDB_WAIT_ALL); + } + if (ret == LDB_ERR_NO_SUCH_OBJECT) { + ret = ldb_build_add_req(&req, ldb, msg, + msg, + NULL, + NULL, ldb_op_default_callback, + NULL); + goto again; + } + + talloc_free(msg); + + return ret; +} diff --git a/source4/dsdb/config.mk b/source4/dsdb/config.mk index eb26f5b433..4150ba0d54 100644 --- a/source4/dsdb/config.mk +++ b/source4/dsdb/config.mk @@ -62,7 +62,8 @@ DREPL_SRV_OBJ_FILES = $(addprefix $(dsdbsrcdir)/repl/, \ drepl_periodic.o \ drepl_partitions.o \ drepl_out_pull.o \ - drepl_out_helpers.o) + drepl_out_helpers.o \ + drepl_notify.o) $(eval $(call proto_header_template,$(dsdbsrcdir)/repl/drepl_service_proto.h,$(DREPL_SRV_OBJ_FILES:.o=.c))) diff --git a/source4/dsdb/repl/drepl_notify.c b/source4/dsdb/repl/drepl_notify.c new file mode 100644 index 0000000000..73280917c5 --- /dev/null +++ b/source4/dsdb/repl/drepl_notify.c @@ -0,0 +1,428 @@ +/* + Unix SMB/CIFS mplementation. + + DSDB replication service periodic notification handling + + Copyright (C) Andrew Tridgell 2009 + based on drepl_periodic + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. + +*/ + +#include "includes.h" +#include "lib/events/events.h" +#include "dsdb/samdb/samdb.h" +#include "auth/auth.h" +#include "smbd/service.h" +#include "lib/messaging/irpc.h" +#include "dsdb/repl/drepl_service.h" +#include "lib/ldb/include/ldb_errors.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 "libcli/composite/composite.h" + + +struct dreplsrv_op_notify_state { + struct composite_context *creq; + + struct dreplsrv_out_connection *conn; + + struct dreplsrv_drsuapi_connection *drsuapi; + + struct drsuapi_DsBindInfoCtr bind_info_ctr; + struct drsuapi_DsBind bind_r; + struct dreplsrv_notify_operation *op; +}; + +/* + receive a DsReplicaSync reply + */ +static void dreplsrv_op_notify_replica_sync_recv(struct rpc_request *req) +{ + struct dreplsrv_op_notify_state *st = talloc_get_type(req->async.private_data, + struct dreplsrv_op_notify_state); + struct composite_context *c = st->creq; + struct drsuapi_DsReplicaSync *r = talloc_get_type(req->ndr.struct_ptr, + struct drsuapi_DsReplicaSync); + + c->status = dcerpc_ndr_request_recv(req); + if (!composite_is_ok(c)) return; + + if (!W_ERROR_IS_OK(r->out.result)) { + composite_error(c, werror_to_ntstatus(r->out.result)); + return; + } + + composite_done(c); +} + +/* + send a DsReplicaSync +*/ +static void dreplsrv_op_notify_replica_sync_send(struct dreplsrv_op_notify_state *st) +{ + struct composite_context *c = st->creq; + struct dreplsrv_partition *partition = st->op->source_dsa->partition; + struct dreplsrv_drsuapi_connection *drsuapi = st->op->source_dsa->conn->drsuapi; + struct rpc_request *req; + struct drsuapi_DsReplicaSync *r; + + r = talloc_zero(st, struct drsuapi_DsReplicaSync); + if (composite_nomem(r, c)) return; + + r->in.bind_handle = &drsuapi->bind_handle; + r->in.level = 1; + r->in.req.req1.naming_context = &partition->nc; + r->in.req.req1.source_dsa_guid = st->op->service->ntds_guid; + r->in.req.req1.options = + DRSUAPI_DS_REPLICA_SYNC_ASYNCHRONOUS_OPERATION | + DRSUAPI_DS_REPLICA_SYNC_WRITEABLE | + DRSUAPI_DS_REPLICA_SYNC_ALL_SOURCES; + + + req = dcerpc_drsuapi_DsReplicaSync_send(drsuapi->pipe, r, r); + composite_continue_rpc(c, req, dreplsrv_op_notify_replica_sync_recv, st); +} + +/* + called when we have an established connection + */ +static void dreplsrv_op_notify_connect_recv(struct composite_context *creq) +{ + struct dreplsrv_op_notify_state *st = talloc_get_type(creq->async.private_data, + struct dreplsrv_op_notify_state); + struct composite_context *c = st->creq; + + c->status = dreplsrv_out_drsuapi_recv(creq); + if (!composite_is_ok(c)) return; + + dreplsrv_op_notify_replica_sync_send(st); +} + +/* + start the ReplicaSync async call + */ +static struct composite_context *dreplsrv_op_notify_send(struct dreplsrv_notify_operation *op) +{ + struct composite_context *c; + struct composite_context *creq; + struct dreplsrv_op_notify_state *st; + + c = composite_create(op, op->service->task->event_ctx); + if (c == NULL) return NULL; + + st = talloc_zero(c, struct dreplsrv_op_notify_state); + if (composite_nomem(st, c)) return c; + + st->creq = c; + st->op = op; + + creq = dreplsrv_out_drsuapi_send(op->source_dsa->conn); + composite_continue(c, creq, dreplsrv_op_notify_connect_recv, st); + + return c; +} + +static void dreplsrv_notify_del_repsTo(struct dreplsrv_notify_operation *op) +{ + uint32_t count; + struct repsFromToBlob *reps; + WERROR werr; + struct dreplsrv_service *s = op->service; + int i; + + werr = dsdb_loadreps(s->samdb, op, op->source_dsa->partition->dn, "repsTo", &reps, &count); + if (!W_ERROR_IS_OK(werr)) { + DEBUG(0,(__location__ ": Failed to load repsTo for %s\n", + ldb_dn_get_linearized(op->source_dsa->partition->dn))); + return; + } + + for (i=0; i<count; i++) { + if (GUID_compare(&reps[i].ctr.ctr1.source_dsa_obj_guid, + &op->source_dsa->repsFrom1->source_dsa_obj_guid) == 0) { + memmove(&reps[i], &reps[i+1], + sizeof(reps[i])*(count-(i+1))); + count--; + } + } + + werr = dsdb_savereps(s->samdb, op, op->source_dsa->partition->dn, "repsTo", reps, count); + if (!W_ERROR_IS_OK(werr)) { + DEBUG(0,(__location__ ": Failed to save repsTo for %s\n", + ldb_dn_get_linearized(op->source_dsa->partition->dn))); + return; + } +} + +/* + called when a notify operation has completed + */ +static void dreplsrv_notify_op_callback(struct dreplsrv_notify_operation *op) +{ + NTSTATUS status; + struct dreplsrv_service *s = op->service; + + status = composite_wait(op->creq); + if (!NT_STATUS_IS_OK(status)) { + DEBUG(0,("dreplsrv_notify: Failed to send DsReplicaSync to %s - %s\n", + op->source_dsa->repsFrom1->other_info->dns_name, + nt_errstr(status))); + } else { + DEBUG(2,("dreplsrv_notify: DsReplicaSync OK for %s\n", + op->source_dsa->repsFrom1->other_info->dns_name)); + op->source_dsa->notify_uSN = op->uSN; + /* delete the repsTo for this replication partner in the + partition, as we have successfully told him to sync */ + dreplsrv_notify_del_repsTo(op); + } + talloc_free(op->creq); + + talloc_free(op); + s->ops.n_current = NULL; + dreplsrv_notify_run_ops(s); +} + + +static void dreplsrv_notify_op_callback_creq(struct composite_context *creq) +{ + struct dreplsrv_notify_operation *op = talloc_get_type(creq->async.private_data, + struct dreplsrv_notify_operation); + dreplsrv_notify_op_callback(op); +} + +/* + run any pending replica sync calls + */ +void dreplsrv_notify_run_ops(struct dreplsrv_service *s) +{ + struct dreplsrv_notify_operation *op; + + if (s->ops.n_current || s->ops.current) { + /* if there's still one running, we're done */ + return; + } + + if (!s->ops.notifies) { + /* if there're no pending operations, we're done */ + return; + } + + op = s->ops.notifies; + s->ops.n_current = op; + DLIST_REMOVE(s->ops.notifies, op); + + op->creq = dreplsrv_op_notify_send(op); + if (!op->creq) { + dreplsrv_notify_op_callback(op); + return; + } + + op->creq->async.fn = dreplsrv_notify_op_callback_creq; + op->creq->async.private_data = op; +} + + +/* + find a source_dsa for a given guid + */ +static struct dreplsrv_partition_source_dsa *dreplsrv_find_source_dsa(struct dreplsrv_partition *p, + struct GUID *guid) +{ + struct dreplsrv_partition_source_dsa *s; + + for (s=p->sources; s; s=s->next) { + if (GUID_compare(&s->repsFrom1->source_dsa_obj_guid, guid) == 0) { + return s; + } + } + return NULL; +} + + +/* + schedule a replicaSync message + */ +static WERROR dreplsrv_schedule_notify_sync(struct dreplsrv_service *service, + struct dreplsrv_partition *p, + struct repsFromToBlob *reps, + TALLOC_CTX *mem_ctx, + uint64_t uSN) +{ + struct dreplsrv_notify_operation *op; + struct dreplsrv_partition_source_dsa *s; + + s = dreplsrv_find_source_dsa(p, &reps->ctr.ctr1.source_dsa_obj_guid); + if (s == NULL) { + DEBUG(0,(__location__ ": Unable to find source_dsa for %s\n", + GUID_string(mem_ctx, &reps->ctr.ctr1.source_dsa_obj_guid))); + return WERR_DS_UNAVAILABLE; + } + + op = talloc_zero(mem_ctx, struct dreplsrv_notify_operation); + W_ERROR_HAVE_NO_MEMORY(op); + + op->service = service; + op->source_dsa = s; + op->uSN = uSN; + + DLIST_ADD_END(service->ops.notifies, op, struct dreplsrv_notify_operation *); + talloc_steal(service, op); + return WERR_OK; +} + +/* + see if a partition has a hugher uSN than what is in the repsTo and + if so then send a DsReplicaSync + */ +static WERROR dreplsrv_notify_check(struct dreplsrv_service *s, + struct dreplsrv_partition *p, + TALLOC_CTX *mem_ctx) +{ + uint32_t count=0; + struct repsFromToBlob *reps; + WERROR werr; + uint64_t uSN; + int ret, i; + + werr = dsdb_loadreps(s->samdb, mem_ctx, p->dn, "repsTo", &reps, &count); + if (count == 0) { + werr = dsdb_loadreps(s->samdb, mem_ctx, p->dn, "repsFrom", &reps, &count); + } + if (!W_ERROR_IS_OK(werr)) { + DEBUG(0,(__location__ ": Failed to load repsTo for %s\n", + ldb_dn_get_linearized(p->dn))); + return werr; + } + + /* loads the partition uSNHighest */ + ret = dsdb_load_partition_usn(s->samdb, p->dn, &uSN); + if (ret != LDB_SUCCESS || uSN == 0) { + /* nothing to do */ + return WERR_OK; + } + + /* see if any of our partners need some of our objects */ + for (i=0; i<count; i++) { + struct dreplsrv_partition_source_dsa *sdsa; + sdsa = dreplsrv_find_source_dsa(p, &reps[i].ctr.ctr1.source_dsa_obj_guid); + if (sdsa == NULL) continue; + if (sdsa->notify_uSN < uSN) { + /* we need to tell this partner to replicate + with us */ + werr = dreplsrv_schedule_notify_sync(s, p, &reps[i], mem_ctx, uSN); + if (!W_ERROR_IS_OK(werr)) { + DEBUG(0,(__location__ ": Failed to setup notify to %s for %s\n", + reps[i].ctr.ctr1.other_info->dns_name, + ldb_dn_get_linearized(p->dn))); + return werr; + } + } + } + + return WERR_OK; +} + +/* + see if any of the partitions have changed, and if so then send a + DsReplicaSync to all the replica partners in the repsTo object + */ +static WERROR dreplsrv_notify_check_all(struct dreplsrv_service *s, TALLOC_CTX *mem_ctx) +{ + WERROR status; + struct dreplsrv_partition *p; + + for (p = s->partitions; p; p = p->next) { + status = dreplsrv_notify_check(s, p, mem_ctx); + W_ERROR_NOT_OK_RETURN(status); + } + + return WERR_OK; +} + +static void dreplsrv_notify_run(struct dreplsrv_service *service); + +static void dreplsrv_notify_handler_te(struct tevent_context *ev, struct tevent_timer *te, + struct timeval t, void *ptr) +{ + struct dreplsrv_service *service = talloc_get_type(ptr, struct dreplsrv_service); + WERROR status; + + service->notify.te = NULL; + + dreplsrv_notify_run(service); + + status = dreplsrv_notify_schedule(service, service->notify.interval); + if (!W_ERROR_IS_OK(status)) { + task_server_terminate(service->task, win_errstr(status)); + return; + } +} + +WERROR dreplsrv_notify_schedule(struct dreplsrv_service *service, uint32_t next_interval) +{ + TALLOC_CTX *tmp_mem; + struct tevent_timer *new_te; + struct timeval next_time; + + /* prevent looping */ + if (next_interval == 0) next_interval = 1; + + next_time = timeval_current_ofs(next_interval, 50); + + if (service->notify.te) { + /* + * if the timestamp of the new event is higher, + * as current next we don't need to reschedule + */ + if (timeval_compare(&next_time, &service->notify.next_event) > 0) { + return WERR_OK; + } + } + + /* reset the next scheduled timestamp */ + service->notify.next_event = next_time; + + new_te = event_add_timed(service->task->event_ctx, service, + service->notify.next_event, + dreplsrv_notify_handler_te, service); + W_ERROR_HAVE_NO_MEMORY(new_te); + + tmp_mem = talloc_new(service); + DEBUG(2,("dreplsrv_notify_schedule(%u) %sscheduled for: %s\n", + next_interval, + (service->notify.te?"re":""), + nt_time_string(tmp_mem, timeval_to_nttime(&next_time)))); + talloc_free(tmp_mem); + + talloc_free(service->notify.te); + service->notify.te = new_te; + + return WERR_OK; +} + +static void dreplsrv_notify_run(struct dreplsrv_service *service) +{ + TALLOC_CTX *mem_ctx; + + mem_ctx = talloc_new(service); + dreplsrv_notify_check_all(service, mem_ctx); + talloc_free(mem_ctx); + + dreplsrv_run_pending_ops(service); + dreplsrv_notify_run_ops(service); +} diff --git a/source4/dsdb/repl/drepl_out_helpers.c b/source4/dsdb/repl/drepl_out_helpers.c index e90f2783b1..168aacdde9 100644 --- a/source4/dsdb/repl/drepl_out_helpers.c +++ b/source4/dsdb/repl/drepl_out_helpers.c @@ -48,7 +48,7 @@ struct dreplsrv_out_drsuapi_state { static void dreplsrv_out_drsuapi_connect_recv(struct composite_context *creq); -static struct composite_context *dreplsrv_out_drsuapi_send(struct dreplsrv_out_connection *conn) +struct composite_context *dreplsrv_out_drsuapi_send(struct dreplsrv_out_connection *conn) { struct composite_context *c; struct composite_context *creq; @@ -165,7 +165,7 @@ static void dreplsrv_out_drsuapi_bind_recv(struct rpc_request *req) composite_done(c); } -static NTSTATUS dreplsrv_out_drsuapi_recv(struct composite_context *c) +NTSTATUS dreplsrv_out_drsuapi_recv(struct composite_context *c) { NTSTATUS status; struct dreplsrv_out_drsuapi_state *st = talloc_get_type(c->private_data, diff --git a/source4/dsdb/repl/drepl_out_pull.c b/source4/dsdb/repl/drepl_out_pull.c index 54dbd29730..b073d265c3 100644 --- a/source4/dsdb/repl/drepl_out_pull.c +++ b/source4/dsdb/repl/drepl_out_pull.c @@ -125,6 +125,7 @@ done: talloc_free(op); s->ops.current = NULL; dreplsrv_run_pending_ops(s); + dreplsrv_notify_run_ops(s); } static void dreplsrv_pending_op_callback_creq(struct composite_context *creq) @@ -140,7 +141,7 @@ void dreplsrv_run_pending_ops(struct dreplsrv_service *s) time_t t; NTTIME now; - if (s->ops.current) { + if (s->ops.current || s->ops.n_current) { /* if there's still one running, we're done */ return; } diff --git a/source4/dsdb/repl/drepl_periodic.c b/source4/dsdb/repl/drepl_periodic.c index 2cfcb310dc..377cecbe99 100644 --- a/source4/dsdb/repl/drepl_periodic.c +++ b/source4/dsdb/repl/drepl_periodic.c @@ -110,4 +110,5 @@ static void dreplsrv_periodic_run(struct dreplsrv_service *service) dreplsrv_refresh_partitions(service); dreplsrv_run_pending_ops(service); + dreplsrv_notify_run_ops(service); } diff --git a/source4/dsdb/repl/drepl_service.c b/source4/dsdb/repl/drepl_service.c index eb49da3d41..cb415b65e6 100644 --- a/source4/dsdb/repl/drepl_service.c +++ b/source4/dsdb/repl/drepl_service.c @@ -196,6 +196,16 @@ static void dreplsrv_task_init(struct task_server *task) return; } + service->notify.interval = lp_parm_int(task->lp_ctx, NULL, "dreplsrv", + "notify_interval", 5); /* in seconds */ + status = dreplsrv_notify_schedule(service, service->notify.interval); + if (!W_ERROR_IS_OK(status)) { + task_server_terminate(task, talloc_asprintf(task, + "dreplsrv: Failed to setup notify schedule: %s\n", + win_errstr(status))); + return; + } + irpc_add_name(task->msg_ctx, "dreplsrv"); IRPC_REGISTER(task->msg_ctx, drsuapi, DRSUAPI_DSREPLICASYNC, drepl_replica_sync, service); diff --git a/source4/dsdb/repl/drepl_service.h b/source4/dsdb/repl/drepl_service.h index a9eea30719..0f9684fa78 100644 --- a/source4/dsdb/repl/drepl_service.h +++ b/source4/dsdb/repl/drepl_service.h @@ -69,6 +69,9 @@ struct dreplsrv_partition_source_dsa { struct repsFromToBlob _repsFromBlob; struct repsFromTo1 *repsFrom1; + /* the last uSN when we sent a notify */ + uint64_t notify_uSN; + /* the reference to the source_dsa and its outgoing connection */ struct dreplsrv_out_connection *conn; }; @@ -107,6 +110,17 @@ struct dreplsrv_out_operation { struct composite_context *creq; }; +struct dreplsrv_notify_operation { + struct dreplsrv_notify_operation *prev, *next; + + struct dreplsrv_service *service; + uint64_t uSN; + + struct dreplsrv_partition_source_dsa *source_dsa; + + struct composite_context *creq; +}; + struct dreplsrv_service { /* the whole drepl service is in one task */ struct task_server *task; @@ -150,6 +164,23 @@ struct dreplsrv_service { struct tevent_timer *te; } periodic; + /* some stuff for notify processing */ + struct { + /* + * the interval between notify runs + */ + uint32_t interval; + + /* + * the timestamp for the next event, + * this is the timstamp passed to event_add_timed() + */ + struct timeval next_event; + + /* here we have a reference to the timed event the schedules the notifies */ + struct tevent_timer *te; + } notify; + /* * the list of partitions we need to replicate */ @@ -166,6 +197,12 @@ struct dreplsrv_service { /* the list of pending operations */ struct dreplsrv_out_operation *pending; + + /* the list of pending notify operations */ + struct dreplsrv_notify_operation *notifies; + + /* an active notify operation */ + struct dreplsrv_notify_operation *n_current; } ops; }; diff --git a/source4/dsdb/samdb/ldb_modules/partition.c b/source4/dsdb/samdb/ldb_modules/partition.c index d4a69952e8..79d11fd958 100644 --- a/source4/dsdb/samdb/ldb_modules/partition.c +++ b/source4/dsdb/samdb/ldb_modules/partition.c @@ -161,6 +161,10 @@ static struct dsdb_partition *find_partition(struct partition_private_data *data } } + if (dn == NULL) { + return NULL; + } + /* Look at base DN */ /* Figure out which partition it is under */ /* Skip the lot if 'data' isn't here yet (initialisation) */ @@ -470,10 +474,10 @@ static int partition_replicate(struct ldb_module *module, struct ldb_request *re static int partition_search(struct ldb_module *module, struct ldb_request *req) { struct ldb_control **saved_controls; - /* Find backend */ struct partition_private_data *data = talloc_get_type(module->private_data, struct partition_private_data); + /* issue request */ /* (later) consider if we should be searching multiple @@ -483,6 +487,17 @@ static int partition_search(struct ldb_module *module, struct ldb_request *req) struct ldb_control *domain_scope_control = ldb_request_get_control(req, LDB_CONTROL_DOMAIN_SCOPE_OID); struct ldb_search_options_control *search_options = NULL; + struct dsdb_partition *p; + + p = find_partition(data, NULL, req); + if (p != NULL) { + /* the caller specified what partition they want the + * search - just pass it on + */ + return ldb_next_request(p->module, req); + } + + if (search_control) { search_options = talloc_get_type(search_control->data, struct ldb_search_options_control); } @@ -765,7 +780,7 @@ static int partition_sequence_number(struct ldb_module *module, struct ldb_reque struct ldb_result *res; struct dsdb_partition *p; - p = find_partition(NULL, NULL, req); + p = find_partition(data, NULL, req); if (p != NULL) { /* the caller specified what partition they want the * sequence number operation on - just pass it on diff --git a/source4/dsdb/samdb/ldb_modules/repl_meta_data.c b/source4/dsdb/samdb/ldb_modules/repl_meta_data.c index 2a16c2bb82..f9411fe95f 100644 --- a/source4/dsdb/samdb/ldb_modules/repl_meta_data.c +++ b/source4/dsdb/samdb/ldb_modules/repl_meta_data.c @@ -47,7 +47,14 @@ #include "lib/util/dlinklist.h" struct replmd_private { + TALLOC_CTX *la_ctx; struct la_entry *la_list; + uint32_t num_ncs; + struct nc_entry { + struct ldb_dn *dn; + struct GUID guid; + uint64_t mod_usn; + } *ncs; }; struct la_entry { @@ -71,6 +78,188 @@ struct replmd_replicated_request { struct ldb_message *search_msg; }; + +/* + initialise the module + allocate the private structure and build the list + of partition DNs for use by replmd_notify() + */ +static int replmd_init(struct ldb_module *module) +{ + struct replmd_private *replmd_private; + struct ldb_context *ldb = ldb_module_get_ctx(module); + + replmd_private = talloc_zero(module, struct replmd_private); + if (replmd_private == NULL) { + ldb_oom(ldb); + return LDB_ERR_OPERATIONS_ERROR; + } + ldb_module_set_private(module, replmd_private); + + return ldb_next_init(module); +} + + +static int nc_compare(struct nc_entry *n1, struct nc_entry *n2) +{ + return ldb_dn_compare(n1->dn, n2->dn); +} + +/* + build the list of partition DNs for use by replmd_notify() + */ +static int replmd_load_NCs(struct ldb_module *module) +{ + const char *attrs[] = { "namingContexts", NULL }; + struct ldb_result *res = NULL; + int i, ret; + TALLOC_CTX *tmp_ctx; + struct ldb_context *ldb; + struct ldb_message_element *el; + struct replmd_private *replmd_private = + talloc_get_type(ldb_module_get_private(module), struct replmd_private); + + if (replmd_private->ncs != NULL) { + return LDB_SUCCESS; + } + + ldb = ldb_module_get_ctx(module); + tmp_ctx = talloc_new(module); + + /* load the list of naming contexts */ + ret = ldb_search(ldb, tmp_ctx, &res, ldb_dn_new(tmp_ctx, ldb, ""), + LDB_SCOPE_BASE, attrs, NULL); + if (ret != LDB_SUCCESS || + res->count != 1) { + DEBUG(0,(__location__ ": Failed to load rootDSE\n")); + return LDB_ERR_OPERATIONS_ERROR; + } + + el = ldb_msg_find_element(res->msgs[0], "namingContexts"); + if (el == NULL) { + DEBUG(0,(__location__ ": Failed to load namingContexts\n")); + return LDB_ERR_OPERATIONS_ERROR; + } + + replmd_private->num_ncs = el->num_values; + replmd_private->ncs = talloc_array(replmd_private, struct nc_entry, + replmd_private->num_ncs); + if (replmd_private->ncs == NULL) { + ldb_oom(ldb); + return LDB_ERR_OPERATIONS_ERROR; + } + + for (i=0; i<replmd_private->num_ncs; i++) { + replmd_private->ncs[i].dn = + ldb_dn_from_ldb_val(replmd_private->ncs, + ldb, &el->values[i]); + replmd_private->ncs[i].mod_usn = 0; + } + + talloc_free(res); + + /* now find the GUIDs of each of those DNs */ + for (i=0; i<replmd_private->num_ncs; i++) { + const char *attrs2[] = { "objectGUID", NULL }; + ret = ldb_search(ldb, tmp_ctx, &res, replmd_private->ncs[i].dn, + LDB_SCOPE_BASE, attrs2, NULL); + if (ret != LDB_SUCCESS || + res->count != 1) { + /* this happens when the schema is first being + setup */ + talloc_free(replmd_private->ncs); + replmd_private->ncs = NULL; + replmd_private->num_ncs = 0; + talloc_free(tmp_ctx); + return LDB_SUCCESS; + } + replmd_private->ncs[i].guid = + samdb_result_guid(res->msgs[0], "objectGUID"); + talloc_free(res); + } + + /* sort the NCs into order, most to least specific */ + qsort(replmd_private->ncs, replmd_private->num_ncs, + sizeof(replmd_private->ncs[0]), QSORT_CAST nc_compare); + + + talloc_free(tmp_ctx); + + return LDB_SUCCESS; +} + + +/* + * notify the repl task that a object has changed. The notifies are + * gathered up in the replmd_private structure then written to the + * @REPLCHANGED object in each partition during the prepare_commit + */ +static int replmd_notify(struct ldb_module *module, struct ldb_dn *dn, uint64_t uSN) +{ + int ret, i; + struct replmd_private *replmd_private = + talloc_get_type(ldb_module_get_private(module), struct replmd_private); + + ret = replmd_load_NCs(module); + if (ret != LDB_SUCCESS) { + return ret; + } + if (replmd_private->num_ncs == 0) { + return LDB_SUCCESS; + } + + for (i=0; i<replmd_private->num_ncs; i++) { + if (ldb_dn_compare_base(replmd_private->ncs[i].dn, dn) == 0) { + break; + } + } + if (i == replmd_private->num_ncs) { + DEBUG(0,(__location__ ": DN not within known NCs '%s'\n", + ldb_dn_get_linearized(dn))); + return LDB_ERR_OPERATIONS_ERROR; + } + + if (uSN > replmd_private->ncs[i].mod_usn) { + replmd_private->ncs[i].mod_usn = uSN; + } + + return LDB_SUCCESS; +} + + +/* + * update a @REPLCHANGED record in each partition if there have been + * any writes of replicated data in the partition + */ +static int replmd_notify_store(struct ldb_module *module) +{ + int i; + struct replmd_private *replmd_private = + talloc_get_type(ldb_module_get_private(module), struct replmd_private); + struct ldb_context *ldb = ldb_module_get_ctx(module); + + for (i=0; i<replmd_private->num_ncs; i++) { + int ret; + + if (replmd_private->ncs[i].mod_usn == 0) { + /* this partition has not changed in this + transaction */ + continue; + } + + ret = dsdb_save_partition_usn(ldb, replmd_private->ncs[i].dn, + replmd_private->ncs[i].mod_usn); + if (ret != LDB_SUCCESS) { + DEBUG(0,(__location__ ": Failed to save partition uSN for %s\n", + ldb_dn_get_linearized(replmd_private->ncs[i].dn))); + return ret; + } + } + + return LDB_SUCCESS; +} + + /* created a replmd_replicated_request context */ @@ -458,6 +647,11 @@ static int replmd_add(struct ldb_module *module, struct ldb_request *req) return ret; } + ret = replmd_notify(module, msg->dn, seq_num); + if (ret != LDB_SUCCESS) { + return ret; + } + /* go on with the call chain */ return ldb_next_request(module, down_req); } @@ -505,6 +699,7 @@ static int replmd_update_rpmd_element(struct ldb_context *ldb, return LDB_ERR_OPERATIONS_ERROR; } omd->ctr.ctr1.count++; + ZERO_STRUCT(omd->ctr.ctr1.array[i]); } /* Get a new sequence number from the backend. We only do this @@ -519,7 +714,7 @@ static int replmd_update_rpmd_element(struct ldb_context *ldb, } md1 = &omd->ctr.ctr1.array[i]; - md1->version = 1; + md1->version++; md1->attid = a->attributeID_id; md1->originating_change_time = now; md1->originating_invocation_id = *our_invocation_id; @@ -534,8 +729,8 @@ static int replmd_update_rpmd_element(struct ldb_context *ldb, * object. This is needed for DRS replication, as the merge on the * client is based on this object */ -static int replmd_update_rpmd(struct ldb_context *ldb, struct ldb_message *msg, - uint64_t *seq_num) +static int replmd_update_rpmd(struct ldb_module *module, + struct ldb_message *msg, uint64_t *seq_num) { const struct ldb_val *omd_value; enum ndr_err_code ndr_err; @@ -548,6 +743,9 @@ static int replmd_update_rpmd(struct ldb_context *ldb, struct ldb_message *msg, int ret; const char *attrs[] = { "replPropertyMetaData" , NULL }; struct ldb_result *res; + struct ldb_context *ldb; + + ldb = ldb_module_get_ctx(module); our_invocation_id = samdb_ntds_invocation_id(ldb); if (!our_invocation_id) { @@ -631,6 +829,11 @@ static int replmd_update_rpmd(struct ldb_context *ldb, struct ldb_message *msg, return ret; } + ret = replmd_notify(module, msg->dn, *seq_num); + if (ret != LDB_SUCCESS) { + return ret; + } + el->num_values = 1; el->values = md_value; } @@ -691,7 +894,7 @@ static int replmd_modify(struct ldb_module *module, struct ldb_request *req) * attribute was changed */ - ret = replmd_update_rpmd(ldb, msg, &seq_num); + ret = replmd_update_rpmd(module, msg, &seq_num); if (ret != LDB_SUCCESS) { return ret; } @@ -729,6 +932,102 @@ static int replmd_modify(struct ldb_module *module, struct ldb_request *req) return ldb_next_request(module, down_req); } + +/* + handle a rename request + + On a rename we need to do an extra ldb_modify which sets the + whenChanged and uSNChanged attributes + */ +static int replmd_rename(struct ldb_module *module, struct ldb_request *req) +{ + struct ldb_context *ldb; + int ret, i; + time_t t = time(NULL); + uint64_t seq_num = 0; + struct ldb_message *msg; + struct replmd_private *replmd_private = + talloc_get_type(ldb_module_get_private(module), struct replmd_private); + + /* do not manipulate our control entries */ + if (ldb_dn_is_special(req->op.mod.message->dn)) { + return ldb_next_request(module, req); + } + + ldb = ldb_module_get_ctx(module); + + ldb_debug(ldb, LDB_DEBUG_TRACE, "replmd_rename\n"); + + /* Get a sequence number from the backend */ + ret = ldb_sequence_number(ldb, LDB_SEQ_NEXT, &seq_num); + if (ret != LDB_SUCCESS) { + return ret; + } + + msg = ldb_msg_new(req); + if (msg == NULL) { + ldb_oom(ldb); + return LDB_ERR_OPERATIONS_ERROR; + } + + msg->dn = req->op.rename.olddn; + + if (add_time_element(msg, "whenChanged", t) != LDB_SUCCESS) { + talloc_free(msg); + return LDB_ERR_OPERATIONS_ERROR; + } + msg->elements[0].flags = LDB_FLAG_MOD_REPLACE; + + if (add_uint64_element(msg, "uSNChanged", seq_num) != LDB_SUCCESS) { + talloc_free(msg); + return LDB_ERR_OPERATIONS_ERROR; + } + msg->elements[1].flags = LDB_FLAG_MOD_REPLACE; + + ret = ldb_modify(ldb, msg); + talloc_free(msg); + if (ret != LDB_SUCCESS) { + return ret; + } + + ret = replmd_load_NCs(module); + if (ret != 0) { + return ret; + } + + /* now update the highest uSNs of the partitions that are + affected. Note that two partitions could be changing */ + for (i=0; i<replmd_private->num_ncs; i++) { + if (ldb_dn_compare_base(replmd_private->ncs[i].dn, + req->op.rename.olddn) == 0) { + break; + } + } + if (i == replmd_private->num_ncs) { + DEBUG(0,(__location__ ": rename olddn outside tree? %s\n", + ldb_dn_get_linearized(req->op.rename.olddn))); + return LDB_ERR_OPERATIONS_ERROR; + } + replmd_private->ncs[i].mod_usn = seq_num; + + for (i=0; i<replmd_private->num_ncs; i++) { + if (ldb_dn_compare_base(replmd_private->ncs[i].dn, + req->op.rename.newdn) == 0) { + break; + } + } + if (i == replmd_private->num_ncs) { + DEBUG(0,(__location__ ": rename newdn outside tree? %s\n", + ldb_dn_get_linearized(req->op.rename.newdn))); + return LDB_ERR_OPERATIONS_ERROR; + } + replmd_private->ncs[i].mod_usn = seq_num; + + /* go on with the call chain */ + return ldb_next_request(module, req); +} + + static int replmd_replicated_request_error(struct replmd_replicated_request *ar, int ret) { return ret; @@ -829,6 +1128,11 @@ static int replmd_replicated_apply_add(struct replmd_replicated_request *ar) return replmd_replicated_request_error(ar, ret); } + ret = replmd_notify(ar->module, msg->dn, seq_num); + if (ret != LDB_SUCCESS) { + return replmd_replicated_request_error(ar, ret); + } + /* * the meta data array is already sorted by the caller */ @@ -1105,6 +1409,11 @@ static int replmd_replicated_apply_merge(struct replmd_replicated_request *ar) msg->elements[i].flags = LDB_FLAG_MOD_REPLACE; } + ret = replmd_notify(ar->module, msg->dn, seq_num); + if (ret != LDB_SUCCESS) { + return replmd_replicated_request_error(ar, ret); + } + if (DEBUGLVL(4)) { char *s = ldb_ldif_message_string(ldb, ar, LDB_CHANGETYPE_MODIFY, msg); DEBUG(4, ("DRS replication modify message:\n%s\n", s)); @@ -1684,18 +1993,17 @@ static int replmd_extended_replicated_objects(struct ldb_module *module, struct for (i=0; i<ar->objs->linked_attributes_count; i++) { struct la_entry *la_entry; - if (replmd_private == NULL) { - DEBUG(0,(__location__ ": repl_meta_data not called from within a transaction\n")); - return LDB_ERR_OPERATIONS_ERROR; + if (replmd_private->la_ctx == NULL) { + replmd_private->la_ctx = talloc_new(replmd_private); } - - la_entry = talloc(replmd_private, struct la_entry); + la_entry = talloc(replmd_private->la_ctx, struct la_entry); if (la_entry == NULL) { ldb_oom(ldb); return LDB_ERR_OPERATIONS_ERROR; } la_entry->la = talloc(la_entry, struct drsuapi_DsReplicaLinkedAttribute); if (la_entry->la == NULL) { + talloc_free(la_entry); ldb_oom(ldb); return LDB_ERR_OPERATIONS_ERROR; } @@ -1891,15 +2199,17 @@ static int replmd_extended(struct ldb_module *module, struct ldb_request *req) static int replmd_start_transaction(struct ldb_module *module) { /* create our private structure for this transaction */ + int i; struct replmd_private *replmd_private = talloc_get_type(ldb_module_get_private(module), struct replmd_private); - talloc_free(replmd_private); - replmd_private = talloc(module, struct replmd_private); - if (replmd_private == NULL) { - return LDB_ERR_OPERATIONS_ERROR; - } + talloc_free(replmd_private->la_ctx); replmd_private->la_list = NULL; - ldb_module_set_private(module, replmd_private); + replmd_private->la_ctx = NULL; + + for (i=0; i<replmd_private->num_ncs; i++) { + replmd_private->ncs[i].mod_usn = 0; + } + return ldb_next_start_trans(module); } @@ -1911,23 +2221,32 @@ static int replmd_prepare_commit(struct ldb_module *module) { struct replmd_private *replmd_private = talloc_get_type(ldb_module_get_private(module), struct replmd_private); - struct la_entry *la; + struct la_entry *la, *prev; + int ret; /* walk the list backwards, to do the first entry first, as we * added the entries with DLIST_ADD() which puts them at the * start of the list */ for (la = replmd_private->la_list; la && la->next; la=la->next) ; - for (; la; la=la->prev) { - int ret; + for (; la; la=prev) { + prev = la->prev; + DLIST_REMOVE(replmd_private->la_list, la); ret = replmd_process_linked_attribute(module, la); if (ret != LDB_SUCCESS) { return ret; } } - talloc_free(replmd_private); - ldb_module_set_private(module, NULL); + talloc_free(replmd_private->la_ctx); + replmd_private->la_list = NULL; + replmd_private->la_ctx = NULL; + + /* possibly change @REPLCHANGED */ + ret = replmd_notify_store(module); + if (ret != LDB_SUCCESS) { + return ret; + } return ldb_next_prepare_commit(module); } @@ -1936,17 +2255,20 @@ static int replmd_del_transaction(struct ldb_module *module) { struct replmd_private *replmd_private = talloc_get_type(ldb_module_get_private(module), struct replmd_private); - talloc_free(replmd_private); - ldb_module_set_private(module, NULL); + talloc_free(replmd_private->la_ctx); + replmd_private->la_list = NULL; + replmd_private->la_ctx = NULL; return ldb_next_del_trans(module); } _PUBLIC_ const struct ldb_module_ops ldb_repl_meta_data_module_ops = { .name = "repl_meta_data", - .add = replmd_add, - .modify = replmd_modify, - .extended = replmd_extended, + .init_context = replmd_init, + .add = replmd_add, + .modify = replmd_modify, + .rename = replmd_rename, + .extended = replmd_extended, .start_transaction = replmd_start_transaction, .prepare_commit = replmd_prepare_commit, .del_transaction = replmd_del_transaction, diff --git a/source4/dsdb/samdb/ldb_modules/samba3sam.c b/source4/dsdb/samdb/ldb_modules/samba3sam.c index 59cb9de717..f5ddff95ed 100644 --- a/source4/dsdb/samdb/ldb_modules/samba3sam.c +++ b/source4/dsdb/samdb/ldb_modules/samba3sam.c @@ -24,7 +24,6 @@ * sambaTrustPassword * sambaUnixIdPool * sambaIdmapEntry - * sambaAccountPolicy * sambaSidEntry * sambaAcctFlags -> systemFlags ? * sambaPasswordHistory -> ntPwdHistory*/ |