source4/wrepl_server/wrepl_out_helpers.c

   1 /*
   2    Unix SMB/CIFS implementation.
   3
   4    WINS Replication server
   5
   6    Copyright (C) Stefan Metzmacher      2005
   7
   8    This program is free software; you can redistribute it and/or modify
   9    it under the terms of the GNU General Public License as published by
  10    the Free Software Foundation; either version 2 of the License, or
  11    (at your option) any later version.
  12
  13    This program is distributed in the hope that it will be useful,
  14    but WITHOUT ANY WARRANTY; without even the implied warranty of
  15    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16    GNU General Public License for more details.
  17
  18    You should have received a copy of the GNU General Public License
  19    along with this program; if not, write to the Free Software
  20    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  21 */
  22
  23 #include "includes.h"
  24 #include "dlinklist.h"
  25 #include "lib/events/events.h"
  26 #include "lib/socket/socket.h"
  27 #include "smbd/service_task.h"
  28 #include "smbd/service_stream.h"
  29 #include "lib/messaging/irpc.h"
  30 #include "librpc/gen_ndr/ndr_winsrepl.h"
  31 #include "wrepl_server/wrepl_server.h"
  32 #include "wrepl_server/wrepl_out_helpers.h"
  33 #include "nbt_server/wins/winsdb.h"
  34 #include "ldb/include/ldb.h"
  35 #include "libcli/composite/composite.h"
  36 #include "libcli/wrepl/winsrepl.h"
  37
  38 enum wreplsrv_out_connect_stage {
  39         WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET,
  40         WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX,
  41         WREPLSRV_OUT_CONNECT_STAGE_DONE
  42 };
  43
  44 struct wreplsrv_out_connect_state {
  45         enum wreplsrv_out_connect_stage stage;
  46         struct composite_context *c;
  47         struct wrepl_request *req;
  48         struct wrepl_associate assoc_io;
  49         enum winsrepl_partner_type type;
  50         struct wreplsrv_out_connection *wreplconn;
  51 };
  52
  53 static void wreplsrv_out_connect_handler(struct wrepl_request *req);
  54
  55 static NTSTATUS wreplsrv_out_connect_wait_socket(struct wreplsrv_out_connect_state *state)
  56 {
  57         NTSTATUS status;
  58
  59         status = wrepl_connect_recv(state->req);
  60         NT_STATUS_NOT_OK_RETURN(status);
  61
  62         state->req = wrepl_associate_send(state->wreplconn->sock, &state->assoc_io);
  63         NT_STATUS_HAVE_NO_MEMORY(state->req);
  64
  65         state->req->async.fn            = wreplsrv_out_connect_handler;
  66         state->req->async.private       = state;
  67
  68         state->stage = WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX;
  69
  70         return NT_STATUS_OK;
  71 }
  72
  73 static NTSTATUS wreplsrv_out_connect_wait_assoc_ctx(struct wreplsrv_out_connect_state *state)
  74 {
  75         NTSTATUS status;
  76
  77         status = wrepl_associate_recv(state->req, &state->assoc_io);
  78         NT_STATUS_NOT_OK_RETURN(status);
  79
  80         state->wreplconn->assoc_ctx.peer_ctx = state->assoc_io.out.assoc_ctx;
  81
  82         if (state->type == WINSREPL_PARTNER_PUSH) {
  83                 state->wreplconn->partner->push.wreplconn = state->wreplconn;
  84                 talloc_steal(state->wreplconn->partner, state->wreplconn);
  85         } else if (state->type == WINSREPL_PARTNER_PULL) {
  86                 state->wreplconn->partner->pull.wreplconn = state->wreplconn;
  87                 talloc_steal(state->wreplconn->partner, state->wreplconn);
  88         }
  89
  90         state->stage = WREPLSRV_OUT_CONNECT_STAGE_DONE;
  91
  92         return NT_STATUS_OK;
  93 }
  94
  95 static void wreplsrv_out_connect_handler(struct wrepl_request *req)
  96 {
  97         struct wreplsrv_out_connect_state *state = talloc_get_type(req->async.private,
  98                                                    struct wreplsrv_out_connect_state);
  99         struct composite_context *c = state->c;
 100
 101         switch (state->stage) {
 102         case WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET:
 103                 c->status = wreplsrv_out_connect_wait_socket(state);
 104                 break;
 105         case WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX:
 106                 c->status = wreplsrv_out_connect_wait_assoc_ctx(state);
 107                 c->state  = COMPOSITE_STATE_DONE;
 108                 break;
 109         case WREPLSRV_OUT_CONNECT_STAGE_DONE:
 110                 c->status = NT_STATUS_INTERNAL_ERROR;
 111         }
 112
 113         if (!NT_STATUS_IS_OK(c->status)) {
 114                 c->state = COMPOSITE_STATE_ERROR;
 115         }
 116
 117         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
 118                 c->async.fn(c);
 119         }
 120 }
 121
 122 static struct composite_context *wreplsrv_out_connect_send(struct wreplsrv_partner *partner,
 123                                                            enum winsrepl_partner_type type,
 124                                                            struct wreplsrv_out_connection *wreplconn)
 125 {
 126         struct composite_context *c = NULL;
 127         struct wreplsrv_service *service = partner->service;
 128         struct wreplsrv_out_connect_state *state = NULL;
 129         struct wreplsrv_out_connection **wreplconnp = &wreplconn;
 130         BOOL cached_connection = False;
 131
 132         c = talloc_zero(partner, struct composite_context);
 133         if (!c) goto failed;
 134
 135         state = talloc_zero(c, struct wreplsrv_out_connect_state);
 136         if (!state) goto failed;
 137         state->c        = c;
 138         state->type     = type;
 139
 140         c->state        = COMPOSITE_STATE_IN_PROGRESS;
 141         c->event_ctx    = service->task->event_ctx;
 142         c->private_data = state;
 143
 144         if (type == WINSREPL_PARTNER_PUSH) {
 145                 cached_connection       = True;
 146                 wreplconn               = partner->push.wreplconn;
 147                 wreplconnp              = &partner->push.wreplconn;
 148         } else if (type == WINSREPL_PARTNER_PULL) {
 149                 cached_connection       = True;
 150                 wreplconn               = partner->pull.wreplconn;
 151                 wreplconnp              = &partner->pull.wreplconn;
 152         }
 153
 154         /* we have a connection already, so use it */
 155         if (wreplconn) {
 156                 if (!wreplconn->sock->dead) {
 157                         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_DONE;
 158                         state->wreplconn= wreplconn;
 159                         composite_done(c);
 160                         return c;
 161                 } else if (!cached_connection) {
 162                         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_DONE;
 163                         state->wreplconn= NULL;
 164                         composite_done(c);
 165                         return c;
 166                 } else {
 167                         talloc_free(wreplconn);
 168                         *wreplconnp = NULL;
 169                 }
 170         }
 171
 172         wreplconn = talloc_zero(state, struct wreplsrv_out_connection);
 173         if (!wreplconn) goto failed;
 174
 175         wreplconn->service      = service;
 176         wreplconn->partner      = partner;
 177         wreplconn->sock         = wrepl_socket_init(wreplconn, service->task->event_ctx);
 178         if (!wreplconn->sock) goto failed;
 179
 180         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET;
 181         state->wreplconn= wreplconn;
 182         state->req      = wrepl_connect_send(wreplconn->sock,
 183                                              partner->our_address,
 184                                              partner->address);
 185         if (!state->req) goto failed;
 186
 187         state->req->async.fn            = wreplsrv_out_connect_handler;
 188         state->req->async.private       = state;
 189
 190         return c;
 191 failed:
 192         talloc_free(c);
 193         return NULL;
 194 }
 195
 196 static NTSTATUS wreplsrv_out_connect_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
 197                                           struct wreplsrv_out_connection **wreplconn)
 198 {
 199         NTSTATUS status;
 200
 201         status = composite_wait(c);
 202
 203         if (NT_STATUS_IS_OK(status)) {
 204                 struct wreplsrv_out_connect_state *state = talloc_get_type(c->private_data,
 205                                                            struct wreplsrv_out_connect_state);
 206                 if (state->wreplconn) {
 207                         *wreplconn = talloc_reference(mem_ctx, state->wreplconn);
 208                         if (!*wreplconn) status = NT_STATUS_NO_MEMORY;
 209                 } else {
 210                         status = NT_STATUS_INVALID_CONNECTION;
 211                 }
 212         }
 213
 214         talloc_free(c);
 215         return status;
 216
 217 }
 218
 219 enum wreplsrv_pull_table_stage {
 220         WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION,
 221         WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY,
 222         WREPLSRV_PULL_TABLE_STAGE_DONE
 223 };
 224
 225 struct wreplsrv_pull_table_state {
 226         enum wreplsrv_pull_table_stage stage;
 227         struct composite_context *c;
 228         struct wrepl_request *req;
 229         struct wrepl_pull_table table_io;
 230         struct wreplsrv_pull_table_io *io;
 231         struct composite_context *creq;
 232         struct wreplsrv_out_connection *wreplconn;
 233 };
 234
 235 static void wreplsrv_pull_table_handler_req(struct wrepl_request *req);
 236
 237 static NTSTATUS wreplsrv_pull_table_wait_connection(struct wreplsrv_pull_table_state *state)
 238 {
 239         NTSTATUS status;
 240
 241         status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
 242         NT_STATUS_NOT_OK_RETURN(status);
 243
 244         state->table_io.in.assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
 245         state->req = wrepl_pull_table_send(state->wreplconn->sock, &state->table_io);
 246         NT_STATUS_HAVE_NO_MEMORY(state->req);
 247
 248         state->req->async.fn            = wreplsrv_pull_table_handler_req;
 249         state->req->async.private       = state;
 250
 251         state->stage = WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY;
 252
 253         return NT_STATUS_OK;
 254 }
 255
 256 static NTSTATUS wreplsrv_pull_table_wait_table_reply(struct wreplsrv_pull_table_state *state)
 257 {
 258         NTSTATUS status;
 259
 260         status = wrepl_pull_table_recv(state->req, state, &state->table_io);
 261         NT_STATUS_NOT_OK_RETURN(status);
 262
 263         state->stage = WREPLSRV_PULL_TABLE_STAGE_DONE;
 264
 265         return NT_STATUS_OK;
 266 }
 267
 268 static void wreplsrv_pull_table_handler(struct wreplsrv_pull_table_state *state)
 269 {
 270         struct composite_context *c = state->c;
 271
 272         switch (state->stage) {
 273         case WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION:
 274                 c->status = wreplsrv_pull_table_wait_connection(state);
 275                 break;
 276         case WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY:
 277                 c->status = wreplsrv_pull_table_wait_table_reply(state);
 278                 c->state  = COMPOSITE_STATE_DONE;
 279                 break;
 280         case WREPLSRV_PULL_TABLE_STAGE_DONE:
 281                 c->status = NT_STATUS_INTERNAL_ERROR;
 282         }
 283
 284         if (!NT_STATUS_IS_OK(c->status)) {
 285                 c->state = COMPOSITE_STATE_ERROR;
 286         }
 287
 288         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
 289                 c->async.fn(c);
 290         }
 291 }
 292
 293 static void wreplsrv_pull_table_handler_creq(struct composite_context *creq)
 294 {
 295         struct wreplsrv_pull_table_state *state = talloc_get_type(creq->async.private_data,
 296                                                   struct wreplsrv_pull_table_state);
 297         wreplsrv_pull_table_handler(state);
 298         return;
 299 }
 300
 301 static void wreplsrv_pull_table_handler_req(struct wrepl_request *req)
 302 {
 303         struct wreplsrv_pull_table_state *state = talloc_get_type(req->async.private,
 304                                                   struct wreplsrv_pull_table_state);
 305         wreplsrv_pull_table_handler(state);
 306         return;
 307 }
 308
 309 struct composite_context *wreplsrv_pull_table_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_table_io *io)
 310 {
 311         struct composite_context *c = NULL;
 312         struct wreplsrv_service *service = io->in.partner->service;
 313         struct wreplsrv_pull_table_state *state = NULL;
 314
 315         c = talloc_zero(mem_ctx, struct composite_context);
 316         if (!c) goto failed;
 317
 318         state = talloc_zero(c, struct wreplsrv_pull_table_state);
 319         if (!state) goto failed;
 320         state->c        = c;
 321         state->io       = io;
 322
 323         c->state        = COMPOSITE_STATE_IN_PROGRESS;
 324         c->event_ctx    = service->task->event_ctx;
 325         c->private_data = state;
 326
 327         if (io->in.num_owners) {
 328                 state->table_io.out.num_partners        = io->in.num_owners;
 329                 state->table_io.out.partners            = io->in.owners;
 330                 state->stage                            = WREPLSRV_PULL_TABLE_STAGE_DONE;
 331                 composite_done(c);
 332                 return c;
 333         }
 334
 335         state->stage    = WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION;
 336         state->creq     = wreplsrv_out_connect_send(io->in.partner, WINSREPL_PARTNER_PULL, NULL);
 337         if (!state->creq) goto failed;
 338
 339         state->creq->async.fn           = wreplsrv_pull_table_handler_creq;
 340         state->creq->async.private_data = state;
 341
 342         return c;
 343 failed:
 344         talloc_free(c);
 345         return NULL;
 346 }
 347
 348 NTSTATUS wreplsrv_pull_table_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
 349                                   struct wreplsrv_pull_table_io *io)
 350 {
 351         NTSTATUS status;
 352
 353         status = composite_wait(c);
 354
 355         if (NT_STATUS_IS_OK(status)) {
 356                 struct wreplsrv_pull_table_state *state = talloc_get_type(c->private_data,
 357                                                           struct wreplsrv_pull_table_state);
 358                 io->out.num_owners      = state->table_io.out.num_partners;
 359                 io->out.owners          = state->table_io.out.partners;
 360                 talloc_reference(mem_ctx, state->table_io.out.partners);
 361         }
 362
 363         talloc_free(c);
 364         return status;
 365 }
 366
 367 enum wreplsrv_pull_names_stage {
 368         WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION,
 369         WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY,
 370         WREPLSRV_PULL_NAMES_STAGE_DONE
 371 };
 372
 373 struct wreplsrv_pull_names_state {
 374         enum wreplsrv_pull_names_stage stage;
 375         struct composite_context *c;
 376         struct wrepl_request *req;
 377         struct wrepl_pull_names pull_io;
 378         struct wreplsrv_pull_names_io *io;
 379         struct composite_context *creq;
 380         struct wreplsrv_out_connection *wreplconn;
 381 };
 382
 383 static void wreplsrv_pull_names_handler_req(struct wrepl_request *req);
 384
 385 static NTSTATUS wreplsrv_pull_names_wait_connection(struct wreplsrv_pull_names_state *state)
 386 {
 387         NTSTATUS status;
 388
 389         status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
 390         NT_STATUS_NOT_OK_RETURN(status);
 391
 392         state->pull_io.in.assoc_ctx     = state->wreplconn->assoc_ctx.peer_ctx;
 393         state->pull_io.in.partner       = state->io->in.owner;
 394         state->req = wrepl_pull_names_send(state->wreplconn->sock, &state->pull_io);
 395         NT_STATUS_HAVE_NO_MEMORY(state->req);
 396
 397         state->req->async.fn            = wreplsrv_pull_names_handler_req;
 398         state->req->async.private       = state;
 399
 400         state->stage = WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY;
 401
 402         return NT_STATUS_OK;
 403 }
 404
 405 static NTSTATUS wreplsrv_pull_names_wait_send_reply(struct wreplsrv_pull_names_state *state)
 406 {
 407         NTSTATUS status;
 408
 409         status = wrepl_pull_names_recv(state->req, state, &state->pull_io);
 410         NT_STATUS_NOT_OK_RETURN(status);
 411
 412         state->stage = WREPLSRV_PULL_NAMES_STAGE_DONE;
 413
 414         return NT_STATUS_OK;
 415 }
 416
 417 static void wreplsrv_pull_names_handler(struct wreplsrv_pull_names_state *state)
 418 {
 419         struct composite_context *c = state->c;
 420
 421         switch (state->stage) {
 422         case WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION:
 423                 c->status = wreplsrv_pull_names_wait_connection(state);
 424                 break;
 425         case WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY:
 426                 c->status = wreplsrv_pull_names_wait_send_reply(state);
 427                 c->state  = COMPOSITE_STATE_DONE;
 428                 break;
 429         case WREPLSRV_PULL_NAMES_STAGE_DONE:
 430                 c->status = NT_STATUS_INTERNAL_ERROR;
 431         }
 432
 433         if (!NT_STATUS_IS_OK(c->status)) {
 434                 c->state = COMPOSITE_STATE_ERROR;
 435         }
 436
 437         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
 438                 c->async.fn(c);
 439         }
 440 }
 441
 442 static void wreplsrv_pull_names_handler_creq(struct composite_context *creq)
 443 {
 444         struct wreplsrv_pull_names_state *state = talloc_get_type(creq->async.private_data,
 445                                                   struct wreplsrv_pull_names_state);
 446         wreplsrv_pull_names_handler(state);
 447         return;
 448 }
 449
 450 static void wreplsrv_pull_names_handler_req(struct wrepl_request *req)
 451 {
 452         struct wreplsrv_pull_names_state *state = talloc_get_type(req->async.private,
 453                                                   struct wreplsrv_pull_names_state);
 454         wreplsrv_pull_names_handler(state);
 455         return;
 456 }
 457
 458 struct composite_context *wreplsrv_pull_names_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_names_io *io)
 459 {
 460         struct composite_context *c = NULL;
 461         struct wreplsrv_service *service = io->in.partner->service;
 462         struct wreplsrv_pull_names_state *state = NULL;
 463         enum winsrepl_partner_type partner_type = WINSREPL_PARTNER_PULL;
 464
 465         if (io->in.wreplconn) partner_type = WINSREPL_PARTNER_NONE;
 466
 467         c = talloc_zero(mem_ctx, struct composite_context);
 468         if (!c) goto failed;
 469
 470         state = talloc_zero(c, struct wreplsrv_pull_names_state);
 471         if (!state) goto failed;
 472         state->c        = c;
 473         state->io       = io;
 474
 475         c->state        = COMPOSITE_STATE_IN_PROGRESS;
 476         c->event_ctx    = service->task->event_ctx;
 477         c->private_data = state;
 478
 479         state->stage    = WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION;
 480         state->creq     = wreplsrv_out_connect_send(io->in.partner, partner_type, io->in.wreplconn);
 481         if (!state->creq) goto failed;
 482
 483         state->creq->async.fn           = wreplsrv_pull_names_handler_creq;
 484         state->creq->async.private_data = state;
 485
 486         return c;
 487 failed:
 488         talloc_free(c);
 489         return NULL;
 490 }
 491
 492 NTSTATUS wreplsrv_pull_names_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
 493                                   struct wreplsrv_pull_names_io *io)
 494 {
 495         NTSTATUS status;
 496
 497         status = composite_wait(c);
 498
 499         if (NT_STATUS_IS_OK(status)) {
 500                 struct wreplsrv_pull_names_state *state = talloc_get_type(c->private_data,
 501                                                           struct wreplsrv_pull_names_state);
 502                 io->out.num_names       = state->pull_io.out.num_names;
 503                 io->out.names           = state->pull_io.out.names;
 504                 talloc_reference(mem_ctx, state->pull_io.out.names);
 505         }
 506
 507         talloc_free(c);
 508         return status;
 509
 510 }
 511
 512 enum wreplsrv_pull_cycle_stage {
 513         WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY,
 514         WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES,
 515         WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC,
 516         WREPLSRV_PULL_CYCLE_STAGE_DONE
 517 };
 518
 519 struct wreplsrv_pull_cycle_state {
 520         enum wreplsrv_pull_cycle_stage stage;
 521         struct composite_context *c;
 522         struct wreplsrv_pull_cycle_io *io;
 523         struct wreplsrv_pull_table_io table_io;
 524         uint32_t current;
 525         struct wreplsrv_pull_names_io names_io;
 526         struct composite_context *creq;
 527         struct wrepl_associate_stop assoc_stop_io;
 528         struct wrepl_request *req;
 529 };
 530
 531 static void wreplsrv_pull_cycle_handler_creq(struct composite_context *creq);
 532 static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req);
 533
 534 static NTSTATUS wreplsrv_pull_cycle_next_owner_do_work(struct wreplsrv_pull_cycle_state *state)
 535 {
 536         struct wreplsrv_owner *current_owner;
 537         struct wreplsrv_owner *local_owner;
 538         uint32_t i;
 539         uint64_t old_max_version = 0;
 540         BOOL do_pull = False;
 541
 542         for (i=state->current; i < state->table_io.out.num_owners; i++) {
 543                 current_owner = wreplsrv_find_owner(state->io->in.partner->pull.table,
 544                                                     state->table_io.out.owners[i].address);
 545
 546                 local_owner = wreplsrv_find_owner(state->io->in.partner->service->table,
 547                                                   state->table_io.out.owners[i].address);
 548                 /*
 549                  * this means we are ourself the current owner,
 550                  * and we don't want replicate ourself
 551                  */
 552                 if (!current_owner) continue;
 553
 554                 /*
 555                  * this means we don't have any records of this owner
 556                  * so fetch them
 557                  */
 558                 if (!local_owner) {
 559                         do_pull         = True;
 560
 561                         break;
 562                 }
 563
 564                 /*
 565                  * this means the remote partner has some new records of this owner
 566                  * fetch them
 567                  */
 568                 if (current_owner->owner.max_version > local_owner->owner.max_version) {
 569                         do_pull         = True;
 570                         old_max_version = local_owner->owner.max_version;
 571                         break;
 572                 }
 573         }
 574         state->current = i;
 575
 576         if (do_pull) {
 577                 state->names_io.in.partner              = state->io->in.partner;
 578                 state->names_io.in.wreplconn            = state->io->in.wreplconn;
 579                 state->names_io.in.owner                = current_owner->owner;
 580                 state->names_io.in.owner.min_version    = old_max_version + 1;
 581                 state->creq = wreplsrv_pull_names_send(state, &state->names_io);
 582                 NT_STATUS_HAVE_NO_MEMORY(state->creq);
 583
 584                 state->creq->async.fn           = wreplsrv_pull_cycle_handler_creq;
 585                 state->creq->async.private_data = state;
 586
 587                 return STATUS_MORE_ENTRIES;
 588         }
 589
 590         return NT_STATUS_OK;
 591 }
 592
 593 static NTSTATUS wreplsrv_pull_cycle_next_owner_wrapper(struct wreplsrv_pull_cycle_state *state)
 594 {
 595         NTSTATUS status;
 596
 597         status = wreplsrv_pull_cycle_next_owner_do_work(state);
 598         if (NT_STATUS_IS_OK(status)) {
 599                 state->stage = WREPLSRV_PULL_CYCLE_STAGE_DONE;
 600         } else if (NT_STATUS_EQUAL(STATUS_MORE_ENTRIES, status)) {
 601                 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES;
 602                 status = NT_STATUS_OK;
 603         }
 604
 605         if (state->stage == WREPLSRV_PULL_CYCLE_STAGE_DONE && state->io->in.wreplconn) {
 606                 state->assoc_stop_io.in.assoc_ctx       = state->io->in.wreplconn->assoc_ctx.peer_ctx;
 607                 state->assoc_stop_io.in.reason          = 0;
 608                 state->req = wrepl_associate_stop_send(state->io->in.wreplconn->sock, &state->assoc_stop_io);
 609                 NT_STATUS_HAVE_NO_MEMORY(state->req);
 610
 611                 state->req->async.fn            = wreplsrv_pull_cycle_handler_req;
 612                 state->req->async.private       = state;
 613
 614                 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC;
 615         }
 616
 617         return status;
 618 }
 619
 620 static NTSTATUS wreplsrv_pull_cycle_wait_table_reply(struct wreplsrv_pull_cycle_state *state)
 621 {
 622         NTSTATUS status;
 623         uint32_t i;
 624
 625         status = wreplsrv_pull_table_recv(state->creq, state, &state->table_io);
 626         NT_STATUS_NOT_OK_RETURN(status);
 627
 628         /* update partner table */
 629         for (i=0; i < state->table_io.out.num_owners; i++) {
 630                 BOOL is_our_addr;
 631
 632                 is_our_addr = wreplsrv_is_our_address(state->io->in.partner->service,
 633                                                       state->table_io.out.owners[i].address);
 634                 if (is_our_addr) continue;
 635
 636                 status = wreplsrv_add_table(state->io->in.partner->service,
 637                                             state->io->in.partner,
 638                                             &state->io->in.partner->pull.table,
 639                                             state->table_io.out.owners[i].address,
 640                                             state->table_io.out.owners[i].max_version);
 641                 NT_STATUS_NOT_OK_RETURN(status);
 642         }
 643
 644         status = wreplsrv_pull_cycle_next_owner_wrapper(state);
 645         NT_STATUS_NOT_OK_RETURN(status);
 646
 647         return status;
 648 }
 649
 650 static NTSTATUS wreplsrv_pull_cycle_apply_records(struct wreplsrv_pull_cycle_state *state)
 651 {
 652         NTSTATUS status;
 653
 654         status = wreplsrv_apply_records(state->io->in.partner, &state->names_io);
 655         NT_STATUS_NOT_OK_RETURN(status);
 656
 657         talloc_free(state->names_io.out.names);
 658         ZERO_STRUCT(state->names_io);
 659
 660         return NT_STATUS_OK;
 661 }
 662
 663 static NTSTATUS wreplsrv_pull_cycle_wait_send_replies(struct wreplsrv_pull_cycle_state *state)
 664 {
 665         NTSTATUS status;
 666
 667         status = wreplsrv_pull_names_recv(state->creq, state, &state->names_io);
 668         NT_STATUS_NOT_OK_RETURN(status);
 669
 670         /*
 671          * TODO: this should maybe an async call,
 672          *       because we may need some network access
 673          *       for conflict resolving
 674          */
 675         status = wreplsrv_pull_cycle_apply_records(state);
 676         NT_STATUS_NOT_OK_RETURN(status);
 677
 678         status = wreplsrv_pull_cycle_next_owner_wrapper(state);
 679         NT_STATUS_NOT_OK_RETURN(status);
 680
 681         return status;
 682 }
 683
 684 static NTSTATUS wreplsrv_pull_cycle_wait_stop_assoc(struct wreplsrv_pull_cycle_state *state)
 685 {
 686         NTSTATUS status;
 687
 688         status = wrepl_associate_stop_recv(state->req, &state->assoc_stop_io);
 689         NT_STATUS_NOT_OK_RETURN(status);
 690
 691         state->stage = WREPLSRV_PULL_CYCLE_STAGE_DONE;
 692
 693         return status;
 694 }
 695
 696 static void wreplsrv_pull_cycle_handler(struct wreplsrv_pull_cycle_state *state)
 697 {
 698         struct composite_context *c = state->c;
 699
 700         switch (state->stage) {
 701         case WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY:
 702                 c->status = wreplsrv_pull_cycle_wait_table_reply(state);
 703                 break;
 704         case WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES:
 705                 c->status = wreplsrv_pull_cycle_wait_send_replies(state);
 706                 break;
 707         case WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC:
 708                 c->status = wreplsrv_pull_cycle_wait_stop_assoc(state);
 709                 break;
 710         case WREPLSRV_PULL_CYCLE_STAGE_DONE:
 711                 c->status = NT_STATUS_INTERNAL_ERROR;
 712         }
 713
 714         if (state->stage == WREPLSRV_PULL_CYCLE_STAGE_DONE) {
 715                 c->state  = COMPOSITE_STATE_DONE;
 716         }
 717
 718         if (!NT_STATUS_IS_OK(c->status)) {
 719                 c->state = COMPOSITE_STATE_ERROR;
 720         }
 721
 722         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
 723                 c->async.fn(c);
 724         }
 725 }
 726
 727 static void wreplsrv_pull_cycle_handler_creq(struct composite_context *creq)
 728 {
 729         struct wreplsrv_pull_cycle_state *state = talloc_get_type(creq->async.private_data,
 730                                                   struct wreplsrv_pull_cycle_state);
 731         wreplsrv_pull_cycle_handler(state);
 732         return;
 733 }
 734
 735 static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req)
 736 {
 737         struct wreplsrv_pull_cycle_state *state = talloc_get_type(req->async.private,
 738                                                   struct wreplsrv_pull_cycle_state);
 739         wreplsrv_pull_cycle_handler(state);
 740         return;
 741 }
 742
 743 struct composite_context *wreplsrv_pull_cycle_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_cycle_io *io)
 744 {
 745         struct composite_context *c = NULL;
 746         struct wreplsrv_service *service = io->in.partner->service;
 747         struct wreplsrv_pull_cycle_state *state = NULL;
 748
 749         c = talloc_zero(mem_ctx, struct composite_context);
 750         if (!c) goto failed;
 751
 752         state = talloc_zero(c, struct wreplsrv_pull_cycle_state);
 753         if (!state) goto failed;
 754         state->c        = c;
 755         state->io       = io;
 756
 757         c->state        = COMPOSITE_STATE_IN_PROGRESS;
 758         c->event_ctx    = service->task->event_ctx;
 759         c->private_data = state;
 760
 761         state->stage    = WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY;
 762         state->table_io.in.partner      = io->in.partner;
 763         state->table_io.in.num_owners   = io->in.num_owners;
 764         state->table_io.in.owners       = io->in.owners;
 765         state->creq = wreplsrv_pull_table_send(state, &state->table_io);
 766         if (!state->creq) goto failed;
 767
 768         state->creq->async.fn           = wreplsrv_pull_cycle_handler_creq;
 769         state->creq->async.private_data = state;
 770
 771         return c;
 772 failed:
 773         talloc_free(c);
 774         return NULL;
 775 }
 776
 777 NTSTATUS wreplsrv_pull_cycle_recv(struct composite_context *c)
 778 {
 779         NTSTATUS status;
 780
 781         status = composite_wait(c);
 782
 783         talloc_free(c);
 784         return status;
 785 }
 786
 787 enum wreplsrv_push_notify_stage {
 788         WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT,
 789         WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_UPDATE,
 790         WREPLSRV_PUSH_NOTIFY_STAGE_DONE
 791 };
 792
 793 struct wreplsrv_push_notify_state {
 794         enum wreplsrv_push_notify_stage stage;
 795         struct composite_context *c;
 796         struct wreplsrv_push_notify_io *io;
 797         enum wrepl_replication_cmd command;
 798         BOOL full_table;
 799         struct wrepl_request *req;
 800         struct wrepl_packet req_packet;
 801         struct wrepl_packet *rep_packet;
 802         struct composite_context *creq;
 803         struct wreplsrv_out_connection *wreplconn;
 804 };
 805
 806 static void wreplsrv_push_notify_handler_creq(struct composite_context *creq);
 807 static void wreplsrv_push_notify_handler_req(struct wrepl_request *req);
 808
 809 static NTSTATUS wreplsrv_push_notify_update(struct wreplsrv_push_notify_state *state)
 810 {
 811         struct wreplsrv_service *service = state->io->in.partner->service;
 812         struct wrepl_packet *req = &state->req_packet;
 813         struct wrepl_replication *repl_out = &state->req_packet.message.replication;
 814         struct wrepl_table *table_out = &state->req_packet.message.replication.info.table;
 815         struct wreplsrv_in_connection *wrepl_in;
 816         NTSTATUS status;
 817         struct socket_context *sock;
 818         struct data_blob_list_item *update_rep;
 819         const char *our_ip;
 820         DATA_BLOB update_blob;
 821
 822         req->opcode     = WREPL_OPCODE_BITS;
 823         req->assoc_ctx  = state->wreplconn->assoc_ctx.peer_ctx;
 824         req->mess_type  = WREPL_REPLICATION;
 825
 826         repl_out->command = state->command;
 827
 828         our_ip = socket_get_my_addr(state->wreplconn->sock->sock, state);
 829         NT_STATUS_HAVE_NO_MEMORY(our_ip);
 830
 831         status = wreplsrv_fill_wrepl_table(service, state, table_out,
 832                                            our_ip, our_ip, state->full_table);
 833         NT_STATUS_NOT_OK_RETURN(status);
 834
 835         state->req = wrepl_request_send(state->wreplconn->sock, req);
 836         NT_STATUS_HAVE_NO_MEMORY(state->req);
 837
 838         sock = state->wreplconn->sock->sock;
 839         talloc_steal(state, state->wreplconn->sock->sock);
 840         state->wreplconn->sock->sock = NULL;
 841
 842         update_blob = state->req->buffer;
 843         talloc_steal(state, state->req->buffer.data);
 844
 845         talloc_free(state->wreplconn->sock);
 846         state->wreplconn->sock = NULL;
 847
 848         status = wreplsrv_in_connection_merge(state->io->in.partner,
 849                                               sock, &wrepl_in);
 850         NT_STATUS_NOT_OK_RETURN(status);
 851
 852         wrepl_in->assoc_ctx.peer_ctx    = state->wreplconn->assoc_ctx.peer_ctx;
 853         wrepl_in->assoc_ctx.our_ctx     = 0;
 854
 855         update_rep = talloc(wrepl_in, struct data_blob_list_item);
 856         NT_STATUS_HAVE_NO_MEMORY(update_rep);
 857
 858         update_rep->blob = update_blob;
 859         talloc_steal(update_rep, update_blob.data);
 860
 861         talloc_free(state->wreplconn);
 862         state->wreplconn = NULL;
 863
 864         if (!wrepl_in->send_queue) {
 865                 EVENT_FD_WRITEABLE(wrepl_in->conn->event.fde);
 866         }
 867         DLIST_ADD_END(wrepl_in->send_queue, update_rep, struct data_blob_list_item *);
 868
 869         state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
 870
 871         return NT_STATUS_OK;
 872 }
 873
 874 static NTSTATUS wreplsrv_push_notify_inform(struct wreplsrv_push_notify_state *state)
 875 {
 876         struct wreplsrv_service *service = state->io->in.partner->service;
 877         struct wrepl_packet *req = &state->req_packet;
 878         struct wrepl_replication *repl_out = &state->req_packet.message.replication;
 879         struct wrepl_table *table_out = &state->req_packet.message.replication.info.table;
 880         NTSTATUS status;
 881         const char *our_ip;
 882
 883         req->opcode     = WREPL_OPCODE_BITS;
 884         req->assoc_ctx  = state->wreplconn->assoc_ctx.peer_ctx;
 885         req->mess_type  = WREPL_REPLICATION;
 886
 887         repl_out->command = state->command;
 888
 889         our_ip = socket_get_my_addr(state->wreplconn->sock->sock, state);
 890         NT_STATUS_HAVE_NO_MEMORY(our_ip);
 891
 892         status = wreplsrv_fill_wrepl_table(service, state, table_out,
 893                                            our_ip, our_ip, state->full_table);
 894         NT_STATUS_NOT_OK_RETURN(status);
 895
 896         state->req = wrepl_request_send(state->wreplconn->sock, req);
 897         NT_STATUS_HAVE_NO_MEMORY(state->req);
 898
 899         /* we won't get a reply to a inform message */
 900         state->req->send_only           = True;
 901         state->req->async.fn            = wreplsrv_push_notify_handler_req;
 902         state->req->async.private       = state;
 903
 904         state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_UPDATE;
 905
 906         return NT_STATUS_OK;
 907 }
 908
 909 static NTSTATUS wreplsrv_push_notify_wait_connect(struct wreplsrv_push_notify_state *state)
 910 {
 911         NTSTATUS status;
 912
 913         status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
 914         NT_STATUS_NOT_OK_RETURN(status);
 915
 916         switch (state->command) {
 917         case WREPL_REPL_UPDATE:
 918                 state->full_table = True;
 919                 return wreplsrv_push_notify_update(state);
 920         case WREPL_REPL_UPDATE2:
 921                 state->full_table = False;
 922                 return wreplsrv_push_notify_update(state);
 923         case WREPL_REPL_INFORM:
 924                 state->full_table = True;
 925                 return wreplsrv_push_notify_inform(state);
 926         case WREPL_REPL_INFORM2:
 927                 state->full_table = False;
 928                 return wreplsrv_push_notify_inform(state);
 929         default:
 930                 return NT_STATUS_INTERNAL_ERROR;
 931         }
 932
 933         return NT_STATUS_INTERNAL_ERROR;
 934 }
 935
 936 static NTSTATUS wreplsrv_push_notify_wait_update(struct wreplsrv_push_notify_state *state)
 937 {
 938         return NT_STATUS_FOOBAR;
 939 }
 940
 941 static void wreplsrv_push_notify_handler(struct wreplsrv_push_notify_state *state)
 942 {
 943         struct composite_context *c = state->c;
 944
 945         switch (state->stage) {
 946         case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT:
 947                 c->status = wreplsrv_push_notify_wait_connect(state);
 948                 break;
 949         case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_UPDATE:
 950                 c->status = wreplsrv_push_notify_wait_update(state);
 951                 break;
 952         case WREPLSRV_PUSH_NOTIFY_STAGE_DONE:
 953                 c->status = NT_STATUS_INTERNAL_ERROR;
 954         }
 955
 956         if (state->stage == WREPLSRV_PUSH_NOTIFY_STAGE_DONE) {
 957                 c->state  = COMPOSITE_STATE_DONE;
 958         }
 959
 960         if (!NT_STATUS_IS_OK(c->status)) {
 961                 c->state = COMPOSITE_STATE_ERROR;
 962         }
 963
 964         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
 965                 c->async.fn(c);
 966         }
 967 }
 968
 969 static void wreplsrv_push_notify_handler_creq(struct composite_context *creq)
 970 {
 971         struct wreplsrv_push_notify_state *state = talloc_get_type(creq->async.private_data,
 972                                                    struct wreplsrv_push_notify_state);
 973         wreplsrv_push_notify_handler(state);
 974         return;
 975 }
 976
 977 static void wreplsrv_push_notify_handler_req(struct wrepl_request *req)
 978 {
 979         struct wreplsrv_push_notify_state *state = talloc_get_type(req->async.private,
 980                                                    struct wreplsrv_push_notify_state);
 981         wreplsrv_push_notify_handler(state);
 982         return;
 983 }
 984
 985 struct composite_context *wreplsrv_push_notify_send(TALLOC_CTX *mem_ctx, struct wreplsrv_push_notify_io *io)
 986 {
 987         struct composite_context *c = NULL;
 988         struct wreplsrv_service *service = io->in.partner->service;
 989         struct wreplsrv_push_notify_state *state = NULL;
 990         enum winsrepl_partner_type partner_type;
 991
 992         c = talloc_zero(mem_ctx, struct composite_context);
 993         if (!c) goto failed;
 994
 995         state = talloc_zero(c, struct wreplsrv_push_notify_state);
 996         if (!state) goto failed;
 997         state->c        = c;
 998         state->io       = io;
 999
1000         if (io->in.inform) {
1001                 /* we can cache the connection in partner->push->wreplconn */
1002                 partner_type = WINSREPL_PARTNER_PUSH;
1003                 if (io->in.propagate) {
1004                         state->command  = WREPL_REPL_INFORM2;
1005                 } else {
1006                         state->command  = WREPL_REPL_INFORM;
1007                 }
1008         } else {
1009                 /* we can NOT cache the connection */
1010                 partner_type = WINSREPL_PARTNER_NONE;
1011                 if (io->in.propagate) {
1012                         state->command  = WREPL_REPL_UPDATE2;
1013                 } else {
1014                         state->command  = WREPL_REPL_UPDATE;
1015                 }
1016         }
1017
1018         c->state        = COMPOSITE_STATE_IN_PROGRESS;
1019         c->event_ctx    = service->task->event_ctx;
1020         c->private_data = state;
1021
1022         state->stage    = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT;
1023         state->creq     = wreplsrv_out_connect_send(io->in.partner, partner_type, NULL);
1024         if (!state->creq) goto failed;
1025
1026         state->creq->async.fn           = wreplsrv_push_notify_handler_creq;
1027         state->creq->async.private_data = state;
1028
1029         return c;
1030 failed:
1031         talloc_free(c);
1032         return NULL;
1033 }
1034
1035 NTSTATUS wreplsrv_push_notify_recv(struct composite_context *c)
1036 {
1037         NTSTATUS status;
1038
1039         status = composite_wait(c);
1040
1041         talloc_free(c);
1042         return status;
1043 }