source4/dsdb/repl/drepl_notify.c

   1 /*
   2    Unix SMB/CIFS mplementation.
   3
   4    DSDB replication service periodic notification handling
   5
   6    Copyright (C) Andrew Tridgell 2009
   7    based on drepl_periodic
   8
   9    This program is free software; you can redistribute it and/or modify
  10    it under the terms of the GNU General Public License as published by
  11    the Free Software Foundation; either version 3 of the License, or
  12    (at your option) any later version.
  13
  14    This program is distributed in the hope that it will be useful,
  15    but WITHOUT ANY WARRANTY; without even the implied warranty of
  16    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17    GNU General Public License for more details.
  18
  19    You should have received a copy of the GNU General Public License
  20    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  21
  22 */
  23
  24 #include "includes.h"
  25 #include "lib/events/events.h"
  26 #include "dsdb/samdb/samdb.h"
  27 #include "auth/auth.h"
  28 #include "smbd/service.h"
  29 #include "lib/messaging/irpc.h"
  30 #include "dsdb/repl/drepl_service.h"
  31 #include "lib/ldb/include/ldb_errors.h"
  32 #include "../lib/util/dlinklist.h"
  33 #include "librpc/gen_ndr/ndr_misc.h"
  34 #include "librpc/gen_ndr/ndr_drsuapi.h"
  35 #include "librpc/gen_ndr/ndr_drsblobs.h"
  36 #include "libcli/composite/composite.h"
  37 #include "../lib/util/tevent_ntstatus.h"
  38
  39
  40 struct dreplsrv_op_notify_state {
  41         struct tevent_context *ev;
  42         struct dreplsrv_notify_operation *op;
  43         void *ndr_struct_ptr;
  44 };
  45
  46 static void dreplsrv_op_notify_connect_done(struct tevent_req *subreq);
  47
  48 /*
  49   start the ReplicaSync async call
  50  */
  51 static struct tevent_req *dreplsrv_op_notify_send(TALLOC_CTX *mem_ctx,
  52                                                   struct tevent_context *ev,
  53                                                   struct dreplsrv_notify_operation *op)
  54 {
  55         struct tevent_req *req;
  56         struct dreplsrv_op_notify_state *state;
  57         struct tevent_req *subreq;
  58
  59         req = tevent_req_create(mem_ctx, &state,
  60                                 struct dreplsrv_op_notify_state);
  61         if (req == NULL) {
  62                 return NULL;
  63         }
  64         state->ev = ev;
  65         state->op = op;
  66
  67         subreq = dreplsrv_out_drsuapi_send(state,
  68                                            ev,
  69                                            op->source_dsa->conn);
  70         if (tevent_req_nomem(subreq, req)) {
  71                 return tevent_req_post(req, ev);
  72         }
  73         tevent_req_set_callback(subreq, dreplsrv_op_notify_connect_done, req);
  74
  75         return req;
  76 }
  77
  78 static void dreplsrv_op_notify_replica_sync_trigger(struct tevent_req *req);
  79
  80 static void dreplsrv_op_notify_connect_done(struct tevent_req *subreq)
  81 {
  82         struct tevent_req *req = tevent_req_callback_data(subreq,
  83                                                           struct tevent_req);
  84         NTSTATUS status;
  85
  86         status = dreplsrv_out_drsuapi_recv(subreq);
  87         TALLOC_FREE(subreq);
  88         if (tevent_req_nterror(req, status)) {
  89                 return;
  90         }
  91
  92         dreplsrv_op_notify_replica_sync_trigger(req);
  93 }
  94
  95 static void dreplsrv_op_notify_replica_sync_done(struct tevent_req *subreq);
  96
  97 static void dreplsrv_op_notify_replica_sync_trigger(struct tevent_req *req)
  98 {
  99         struct dreplsrv_op_notify_state *state =
 100                 tevent_req_data(req,
 101                 struct dreplsrv_op_notify_state);
 102         struct dreplsrv_partition *partition = state->op->source_dsa->partition;
 103         struct dreplsrv_drsuapi_connection *drsuapi = state->op->source_dsa->conn->drsuapi;
 104         struct drsuapi_DsReplicaSync *r;
 105         struct tevent_req *subreq;
 106
 107         r = talloc_zero(state, struct drsuapi_DsReplicaSync);
 108         if (tevent_req_nomem(r, req)) {
 109                 return;
 110         }
 111         r->in.req = talloc_zero(r, union drsuapi_DsReplicaSyncRequest);
 112         if (tevent_req_nomem(r, req)) {
 113                 return;
 114         }
 115         r->in.bind_handle       = &drsuapi->bind_handle;
 116         r->in.level = 1;
 117         r->in.req->req1.naming_context = &partition->nc;
 118         r->in.req->req1.source_dsa_guid = state->op->service->ntds_guid;
 119         r->in.req->req1.options =
 120                 DRSUAPI_DRS_ASYNC_OP |
 121                 DRSUAPI_DRS_UPDATE_NOTIFICATION |
 122                 DRSUAPI_DRS_WRIT_REP;
 123
 124         if (state->op->is_urgent) {
 125                 r->in.req->req1.options |= DRSUAPI_DRS_SYNC_URGENT;
 126         }
 127
 128         state->ndr_struct_ptr = r;
 129
 130         if (DEBUGLVL(10)) {
 131                 NDR_PRINT_IN_DEBUG(drsuapi_DsReplicaSync, r);
 132         }
 133
 134         subreq = dcerpc_drsuapi_DsReplicaSync_r_send(state,
 135                                                      state->ev,
 136                                                      drsuapi->drsuapi_handle,
 137                                                      r);
 138         if (tevent_req_nomem(subreq, req)) {
 139                 return;
 140         }
 141         tevent_req_set_callback(subreq, dreplsrv_op_notify_replica_sync_done, req);
 142 }
 143
 144 static void dreplsrv_op_notify_replica_sync_done(struct tevent_req *subreq)
 145 {
 146         struct tevent_req *req =
 147                 tevent_req_callback_data(subreq,
 148                 struct tevent_req);
 149         struct dreplsrv_op_notify_state *state =
 150                 tevent_req_data(req,
 151                 struct dreplsrv_op_notify_state);
 152         struct drsuapi_DsReplicaSync *r = talloc_get_type(state->ndr_struct_ptr,
 153                                                           struct drsuapi_DsReplicaSync);
 154         NTSTATUS status;
 155
 156         state->ndr_struct_ptr = NULL;
 157
 158         status = dcerpc_drsuapi_DsReplicaSync_r_recv(subreq, r);
 159         TALLOC_FREE(subreq);
 160         if (tevent_req_nterror(req, status)) {
 161                 return;
 162         }
 163
 164         if (!W_ERROR_IS_OK(r->out.result)) {
 165                 status = werror_to_ntstatus(r->out.result);
 166                 tevent_req_nterror(req, status);
 167                 return;
 168         }
 169
 170         tevent_req_done(req);
 171 }
 172
 173 static NTSTATUS dreplsrv_op_notify_recv(struct tevent_req *req)
 174 {
 175         return tevent_req_simple_recv_ntstatus(req);
 176 }
 177
 178 /*
 179   called when a notify operation has completed
 180  */
 181 static void dreplsrv_notify_op_callback(struct tevent_req *subreq)
 182 {
 183         struct dreplsrv_notify_operation *op =
 184                 tevent_req_callback_data(subreq,
 185                 struct dreplsrv_notify_operation);
 186         NTSTATUS status;
 187         struct dreplsrv_service *s = op->service;
 188         WERROR werr;
 189
 190         status = dreplsrv_op_notify_recv(subreq);
 191         werr = ntstatus_to_werror(status);
 192         TALLOC_FREE(subreq);
 193         if (!NT_STATUS_IS_OK(status)) {
 194                 DEBUG(4,("dreplsrv_notify: Failed to send DsReplicaSync to %s for %s - %s : %s\n",
 195                          op->source_dsa->repsFrom1->other_info->dns_name,
 196                          ldb_dn_get_linearized(op->source_dsa->partition->dn),
 197                          nt_errstr(status), win_errstr(werr)));
 198                 if (W_ERROR_EQUAL(werr, WERR_DS_DRA_NO_REPLICA)) {
 199                         DEBUG(0,("Enabling SYNC_ALL workaround\n"));
 200                         op->service->syncall_workaround = true;
 201                 }
 202         } else {
 203                 DEBUG(2,("dreplsrv_notify: DsReplicaSync OK for %s\n",
 204                          op->source_dsa->repsFrom1->other_info->dns_name));
 205                 op->source_dsa->notify_uSN = op->uSN;
 206         }
 207
 208         drepl_reps_update(s, "repsTo", op->source_dsa->partition->dn,
 209                           &op->source_dsa->repsFrom1->source_dsa_obj_guid,
 210                           werr);
 211
 212         talloc_free(op);
 213         s->ops.n_current = NULL;
 214         dreplsrv_run_pending_ops(s);
 215 }
 216
 217 /*
 218   run any pending replica sync calls
 219  */
 220 void dreplsrv_notify_run_ops(struct dreplsrv_service *s)
 221 {
 222         struct dreplsrv_notify_operation *op;
 223         struct tevent_req *subreq;
 224
 225         if (s->ops.n_current || s->ops.current) {
 226                 /* if there's still one running, we're done */
 227                 return;
 228         }
 229
 230         if (!s->ops.notifies) {
 231                 /* if there're no pending operations, we're done */
 232                 return;
 233         }
 234
 235         op = s->ops.notifies;
 236         s->ops.n_current = op;
 237         DLIST_REMOVE(s->ops.notifies, op);
 238
 239         subreq = dreplsrv_op_notify_send(op, s->task->event_ctx, op);
 240         if (!subreq) {
 241                 DEBUG(0,("dreplsrv_notify_run_ops: dreplsrv_op_notify_send[%s][%s] - no memory\n",
 242                          op->source_dsa->repsFrom1->other_info->dns_name,
 243                          ldb_dn_get_linearized(op->source_dsa->partition->dn)));
 244                 return;
 245         }
 246         tevent_req_set_callback(subreq, dreplsrv_notify_op_callback, op);
 247         DEBUG(4,("started DsReplicaSync for %s to %s\n",
 248                  ldb_dn_get_linearized(op->source_dsa->partition->dn),
 249                  op->source_dsa->repsFrom1->other_info->dns_name));
 250 }
 251
 252
 253 /*
 254   find a source_dsa for a given guid
 255  */
 256 static struct dreplsrv_partition_source_dsa *dreplsrv_find_notify_dsa(struct dreplsrv_partition *p,
 257                                                                       struct GUID *guid)
 258 {
 259         struct dreplsrv_partition_source_dsa *s;
 260
 261         /* first check the sources list */
 262         for (s=p->sources; s; s=s->next) {
 263                 if (GUID_compare(&s->repsFrom1->source_dsa_obj_guid, guid) == 0) {
 264                         return s;
 265                 }
 266         }
 267
 268         /* then the notifies list */
 269         for (s=p->notifies; s; s=s->next) {
 270                 if (GUID_compare(&s->repsFrom1->source_dsa_obj_guid, guid) == 0) {
 271                         return s;
 272                 }
 273         }
 274         return NULL;
 275 }
 276
 277
 278 /*
 279   schedule a replicaSync message
 280  */
 281 static WERROR dreplsrv_schedule_notify_sync(struct dreplsrv_service *service,
 282                                             struct dreplsrv_partition *p,
 283                                             struct repsFromToBlob *reps,
 284                                             TALLOC_CTX *mem_ctx,
 285                                             uint64_t uSN,
 286                                             bool is_urgent,
 287                                             uint32_t replica_flags)
 288 {
 289         struct dreplsrv_notify_operation *op;
 290         struct dreplsrv_partition_source_dsa *s;
 291
 292         s = dreplsrv_find_notify_dsa(p, &reps->ctr.ctr1.source_dsa_obj_guid);
 293         if (s == NULL) {
 294                 DEBUG(0,(__location__ ": Unable to find source_dsa for %s\n",
 295                          GUID_string(mem_ctx, &reps->ctr.ctr1.source_dsa_obj_guid)));
 296                 return WERR_DS_UNAVAILABLE;
 297         }
 298
 299         /* first try to find an existing notify operation */
 300         for (op = service->ops.notifies; op; op = op->next) {
 301                 if (op->source_dsa != s) {
 302                         continue;
 303                 }
 304
 305                 if (op->is_urgent != is_urgent) {
 306                         continue;
 307                 }
 308
 309                 if (op->replica_flags != replica_flags) {
 310                         continue;
 311                 }
 312
 313                 if (op->uSN < uSN) {
 314                         op->uSN = uSN;
 315                 }
 316
 317                 /* reuse the notify operation, as it's not yet started */
 318                 return WERR_OK;
 319         }
 320
 321         op = talloc_zero(mem_ctx, struct dreplsrv_notify_operation);
 322         W_ERROR_HAVE_NO_MEMORY(op);
 323
 324         op->service       = service;
 325         op->source_dsa    = s;
 326         op->uSN           = uSN;
 327         op->is_urgent     = is_urgent;
 328         op->replica_flags = replica_flags;
 329         op->schedule_time = time(NULL);
 330
 331         DLIST_ADD_END(service->ops.notifies, op, struct dreplsrv_notify_operation *);
 332         talloc_steal(service, op);
 333         return WERR_OK;
 334 }
 335
 336 /*
 337   see if a partition has a hugher uSN than what is in the repsTo and
 338   if so then send a DsReplicaSync
 339  */
 340 static WERROR dreplsrv_notify_check(struct dreplsrv_service *s,
 341                                     struct dreplsrv_partition *p,
 342                                     TALLOC_CTX *mem_ctx)
 343 {
 344         uint32_t count=0;
 345         struct repsFromToBlob *reps;
 346         WERROR werr;
 347         uint64_t uSNHighest;
 348         uint64_t uSNUrgent;
 349         uint32_t i;
 350         int ret;
 351
 352         werr = dsdb_loadreps(s->samdb, mem_ctx, p->dn, "repsTo", &reps, &count);
 353         if (!W_ERROR_IS_OK(werr)) {
 354                 DEBUG(0,(__location__ ": Failed to load repsTo for %s\n",
 355                          ldb_dn_get_linearized(p->dn)));
 356                 return werr;
 357         }
 358
 359         /* loads the partition uSNHighest and uSNUrgent */
 360         ret = dsdb_load_partition_usn(s->samdb, p->dn, &uSNHighest, &uSNUrgent);
 361         if (ret != LDB_SUCCESS || uSNHighest == 0) {
 362                 /* nothing to do */
 363                 return WERR_OK;
 364         }
 365
 366         /* see if any of our partners need some of our objects */
 367         for (i=0; i<count; i++) {
 368                 struct dreplsrv_partition_source_dsa *sdsa;
 369                 uint32_t replica_flags;
 370                 sdsa = dreplsrv_find_notify_dsa(p, &reps[i].ctr.ctr1.source_dsa_obj_guid);
 371                 replica_flags = reps[i].ctr.ctr1.replica_flags;
 372                 if (sdsa == NULL) continue;
 373                 if (sdsa->notify_uSN < uSNHighest) {
 374                         /* we need to tell this partner to replicate
 375                            with us */
 376                         bool is_urgent = sdsa->notify_uSN < uSNUrgent;
 377
 378                         /* check if urgent replication is needed */
 379                         werr = dreplsrv_schedule_notify_sync(s, p, &reps[i], mem_ctx,
 380                                                              uSNHighest, is_urgent, replica_flags);
 381                         if (!W_ERROR_IS_OK(werr)) {
 382                                 DEBUG(0,(__location__ ": Failed to setup notify to %s for %s\n",
 383                                          reps[i].ctr.ctr1.other_info->dns_name,
 384                                          ldb_dn_get_linearized(p->dn)));
 385                                 return werr;
 386                         }
 387                         DEBUG(4,("queued DsReplicaSync for %s to %s (urgent=%s) uSN=%llu:%llu\n",
 388                                  ldb_dn_get_linearized(p->dn),
 389                                  reps[i].ctr.ctr1.other_info->dns_name,
 390                                  is_urgent?"true":"false",
 391                                  (unsigned long long)sdsa->notify_uSN,
 392                                  (unsigned long long)uSNHighest));
 393                 }
 394         }
 395
 396         return WERR_OK;
 397 }
 398
 399 /*
 400   see if any of the partitions have changed, and if so then send a
 401   DsReplicaSync to all the replica partners in the repsTo object
 402  */
 403 static WERROR dreplsrv_notify_check_all(struct dreplsrv_service *s, TALLOC_CTX *mem_ctx)
 404 {
 405         WERROR status;
 406         struct dreplsrv_partition *p;
 407
 408         for (p = s->partitions; p; p = p->next) {
 409                 status = dreplsrv_notify_check(s, p, mem_ctx);
 410                 W_ERROR_NOT_OK_RETURN(status);
 411         }
 412
 413         return WERR_OK;
 414 }
 415
 416 static void dreplsrv_notify_run(struct dreplsrv_service *service);
 417
 418 static void dreplsrv_notify_handler_te(struct tevent_context *ev, struct tevent_timer *te,
 419                                        struct timeval t, void *ptr)
 420 {
 421         struct dreplsrv_service *service = talloc_get_type(ptr, struct dreplsrv_service);
 422         WERROR status;
 423
 424         service->notify.te = NULL;
 425
 426         dreplsrv_notify_run(service);
 427
 428         status = dreplsrv_notify_schedule(service, service->notify.interval);
 429         if (!W_ERROR_IS_OK(status)) {
 430                 task_server_terminate(service->task, win_errstr(status), false);
 431                 return;
 432         }
 433 }
 434
 435 WERROR dreplsrv_notify_schedule(struct dreplsrv_service *service, uint32_t next_interval)
 436 {
 437         TALLOC_CTX *tmp_mem;
 438         struct tevent_timer *new_te;
 439         struct timeval next_time;
 440
 441         /* prevent looping */
 442         if (next_interval == 0) next_interval = 1;
 443
 444         next_time = timeval_current_ofs(next_interval, 50);
 445
 446         if (service->notify.te) {
 447                 /*
 448                  * if the timestamp of the new event is higher,
 449                  * as current next we don't need to reschedule
 450                  */
 451                 if (timeval_compare(&next_time, &service->notify.next_event) > 0) {
 452                         return WERR_OK;
 453                 }
 454         }
 455
 456         /* reset the next scheduled timestamp */
 457         service->notify.next_event = next_time;
 458
 459         new_te = event_add_timed(service->task->event_ctx, service,
 460                                  service->notify.next_event,
 461                                  dreplsrv_notify_handler_te, service);
 462         W_ERROR_HAVE_NO_MEMORY(new_te);
 463
 464         tmp_mem = talloc_new(service);
 465         DEBUG(4,("dreplsrv_notify_schedule(%u) %sscheduled for: %s\n",
 466                 next_interval,
 467                 (service->notify.te?"re":""),
 468                 nt_time_string(tmp_mem, timeval_to_nttime(&next_time))));
 469         talloc_free(tmp_mem);
 470
 471         talloc_free(service->notify.te);
 472         service->notify.te = new_te;
 473
 474         return WERR_OK;
 475 }
 476
 477 static void dreplsrv_notify_run(struct dreplsrv_service *service)
 478 {
 479         TALLOC_CTX *mem_ctx;
 480
 481         mem_ctx = talloc_new(service);
 482         dreplsrv_notify_check_all(service, mem_ctx);
 483         talloc_free(mem_ctx);
 484
 485         dreplsrv_run_pending_ops(service);
 486 }