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