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 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 #include "libcli/resolve/resolve.h"
  33 #include "param/param.h"
  34
  35 enum wreplsrv_out_connect_stage {
  36         WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET,
  37         WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX,
  38         WREPLSRV_OUT_CONNECT_STAGE_DONE
  39 };
  40
  41 struct wreplsrv_out_connect_state {
  42         enum wreplsrv_out_connect_stage stage;
  43         struct composite_context *c;
  44         struct wrepl_request *req;
  45         struct composite_context *c_req;
  46         struct wrepl_associate assoc_io;
  47         enum winsrepl_partner_type type;
  48         struct wreplsrv_out_connection *wreplconn;
  49 };
  50
  51 static void wreplsrv_out_connect_handler_creq(struct composite_context *c_req);
  52 static void wreplsrv_out_connect_handler_req(struct wrepl_request *req);
  53
  54 static NTSTATUS wreplsrv_out_connect_wait_socket(struct wreplsrv_out_connect_state *state)
  55 {
  56         NTSTATUS status;
  57
  58         status = wrepl_connect_recv(state->c_req);
  59         NT_STATUS_NOT_OK_RETURN(status);
  60
  61         state->req = wrepl_associate_send(state->wreplconn->sock, &state->assoc_io);
  62         NT_STATUS_HAVE_NO_MEMORY(state->req);
  63
  64         state->req->async.fn            = wreplsrv_out_connect_handler_req;
  65         state->req->async.private       = state;
  66
  67         state->stage = WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX;
  68
  69         return NT_STATUS_OK;
  70 }
  71
  72 static NTSTATUS wreplsrv_out_connect_wait_assoc_ctx(struct wreplsrv_out_connect_state *state)
  73 {
  74         NTSTATUS status;
  75
  76         status = wrepl_associate_recv(state->req, &state->assoc_io);
  77         NT_STATUS_NOT_OK_RETURN(status);
  78
  79         state->wreplconn->assoc_ctx.peer_ctx = state->assoc_io.out.assoc_ctx;
  80
  81         if (state->type == WINSREPL_PARTNER_PUSH) {
  82                 state->wreplconn->partner->push.wreplconn = state->wreplconn;
  83                 talloc_steal(state->wreplconn->partner, state->wreplconn);
  84         } else if (state->type == WINSREPL_PARTNER_PULL) {
  85                 state->wreplconn->partner->pull.wreplconn = state->wreplconn;
  86                 talloc_steal(state->wreplconn->partner, state->wreplconn);
  87         }
  88
  89         state->stage = WREPLSRV_OUT_CONNECT_STAGE_DONE;
  90
  91         return NT_STATUS_OK;
  92 }
  93
  94 static void wreplsrv_out_connect_handler(struct wreplsrv_out_connect_state *state)
  95 {
  96         struct composite_context *c = state->c;
  97
  98         switch (state->stage) {
  99         case WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET:
 100                 c->status = wreplsrv_out_connect_wait_socket(state);
 101                 break;
 102         case WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX:
 103                 c->status = wreplsrv_out_connect_wait_assoc_ctx(state);
 104                 c->state  = COMPOSITE_STATE_DONE;
 105                 break;
 106         case WREPLSRV_OUT_CONNECT_STAGE_DONE:
 107                 c->status = NT_STATUS_INTERNAL_ERROR;
 108         }
 109
 110         if (!NT_STATUS_IS_OK(c->status)) {
 111                 c->state = COMPOSITE_STATE_ERROR;
 112         }
 113
 114         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
 115                 c->async.fn(c);
 116         }
 117 }
 118
 119 static void wreplsrv_out_connect_handler_creq(struct composite_context *creq)
 120 {
 121         struct wreplsrv_out_connect_state *state = talloc_get_type(creq->async.private_data,
 122                                                    struct wreplsrv_out_connect_state);
 123         wreplsrv_out_connect_handler(state);
 124         return;
 125 }
 126
 127 static void wreplsrv_out_connect_handler_req(struct wrepl_request *req)
 128 {
 129         struct wreplsrv_out_connect_state *state = talloc_get_type(req->async.private,
 130                                                    struct wreplsrv_out_connect_state);
 131         wreplsrv_out_connect_handler(state);
 132         return;
 133 }
 134
 135 static struct composite_context *wreplsrv_out_connect_send(struct wreplsrv_partner *partner,
 136                                                            enum winsrepl_partner_type type,
 137                                                            struct wreplsrv_out_connection *wreplconn)
 138 {
 139         struct composite_context *c = NULL;
 140         struct wreplsrv_service *service = partner->service;
 141         struct wreplsrv_out_connect_state *state = NULL;
 142         struct wreplsrv_out_connection **wreplconnp = &wreplconn;
 143         bool cached_connection = false;
 144
 145         c = talloc_zero(partner, struct composite_context);
 146         if (!c) goto failed;
 147
 148         state = talloc_zero(c, struct wreplsrv_out_connect_state);
 149         if (!state) goto failed;
 150         state->c        = c;
 151         state->type     = type;
 152
 153         c->state        = COMPOSITE_STATE_IN_PROGRESS;
 154         c->event_ctx    = service->task->event_ctx;
 155         c->private_data = state;
 156
 157         if (type == WINSREPL_PARTNER_PUSH) {
 158                 cached_connection       = true;
 159                 wreplconn               = partner->push.wreplconn;
 160                 wreplconnp              = &partner->push.wreplconn;
 161         } else if (type == WINSREPL_PARTNER_PULL) {
 162                 cached_connection       = true;
 163                 wreplconn               = partner->pull.wreplconn;
 164                 wreplconnp              = &partner->pull.wreplconn;
 165         }
 166
 167         /* we have a connection already, so use it */
 168         if (wreplconn) {
 169                 if (!wreplconn->sock->dead) {
 170                         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_DONE;
 171                         state->wreplconn= wreplconn;
 172                         composite_done(c);
 173                         return c;
 174                 } else if (!cached_connection) {
 175                         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_DONE;
 176                         state->wreplconn= NULL;
 177                         composite_done(c);
 178                         return c;
 179                 } else {
 180                         talloc_free(wreplconn);
 181                         *wreplconnp = NULL;
 182                 }
 183         }
 184
 185         wreplconn = talloc_zero(state, struct wreplsrv_out_connection);
 186         if (!wreplconn) goto failed;
 187
 188         wreplconn->service      = service;
 189         wreplconn->partner      = partner;
 190         wreplconn->sock         = wrepl_socket_init(wreplconn, service->task->event_ctx, lp_iconv_convenience(service->task->lp_ctx));
 191         if (!wreplconn->sock) goto failed;
 192
 193         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET;
 194         state->wreplconn= wreplconn;
 195         state->c_req    = wrepl_connect_send(wreplconn->sock,
 196                                              partner->our_address?partner->our_address:wrepl_best_ip(service->task->lp_ctx, partner->address),
 197                                              partner->address);
 198         if (!state->c_req) goto failed;
 199
 200         state->c_req->async.fn                  = wreplsrv_out_connect_handler_creq;
 201         state->c_req->async.private_data        = state;
 202
 203         return c;
 204 failed:
 205         talloc_free(c);
 206         return NULL;
 207 }
 208
 209 static NTSTATUS wreplsrv_out_connect_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
 210                                           struct wreplsrv_out_connection **wreplconn)
 211 {
 212         NTSTATUS status;
 213
 214         status = composite_wait(c);
 215
 216         if (NT_STATUS_IS_OK(status)) {
 217                 struct wreplsrv_out_connect_state *state = talloc_get_type(c->private_data,
 218                                                            struct wreplsrv_out_connect_state);
 219                 if (state->wreplconn) {
 220                         *wreplconn = talloc_reference(mem_ctx, state->wreplconn);
 221                         if (!*wreplconn) status = NT_STATUS_NO_MEMORY;
 222                 } else {
 223                         status = NT_STATUS_INVALID_CONNECTION;
 224                 }
 225         }
 226
 227         talloc_free(c);
 228         return status;
 229
 230 }
 231
 232 struct wreplsrv_pull_table_io {
 233         struct {
 234                 struct wreplsrv_partner *partner;
 235                 uint32_t num_owners;
 236                 struct wrepl_wins_owner *owners;
 237         } in;
 238         struct {
 239                 uint32_t num_owners;
 240                 struct wrepl_wins_owner *owners;
 241         } out;
 242 };
 243
 244 enum wreplsrv_pull_table_stage {
 245         WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION,
 246         WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY,
 247         WREPLSRV_PULL_TABLE_STAGE_DONE
 248 };
 249
 250 struct wreplsrv_pull_table_state {
 251         enum wreplsrv_pull_table_stage stage;
 252         struct composite_context *c;
 253         struct wrepl_request *req;
 254         struct wrepl_pull_table table_io;
 255         struct wreplsrv_pull_table_io *io;
 256         struct composite_context *creq;
 257         struct wreplsrv_out_connection *wreplconn;
 258 };
 259
 260 static void wreplsrv_pull_table_handler_req(struct wrepl_request *req);
 261
 262 static NTSTATUS wreplsrv_pull_table_wait_connection(struct wreplsrv_pull_table_state *state)
 263 {
 264         NTSTATUS status;
 265
 266         status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
 267         NT_STATUS_NOT_OK_RETURN(status);
 268
 269         state->table_io.in.assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
 270         state->req = wrepl_pull_table_send(state->wreplconn->sock, &state->table_io);
 271         NT_STATUS_HAVE_NO_MEMORY(state->req);
 272
 273         state->req->async.fn            = wreplsrv_pull_table_handler_req;
 274         state->req->async.private       = state;
 275
 276         state->stage = WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY;
 277
 278         return NT_STATUS_OK;
 279 }
 280
 281 static NTSTATUS wreplsrv_pull_table_wait_table_reply(struct wreplsrv_pull_table_state *state)
 282 {
 283         NTSTATUS status;
 284
 285         status = wrepl_pull_table_recv(state->req, state, &state->table_io);
 286         NT_STATUS_NOT_OK_RETURN(status);
 287
 288         state->stage = WREPLSRV_PULL_TABLE_STAGE_DONE;
 289
 290         return NT_STATUS_OK;
 291 }
 292
 293 static void wreplsrv_pull_table_handler(struct wreplsrv_pull_table_state *state)
 294 {
 295         struct composite_context *c = state->c;
 296
 297         switch (state->stage) {
 298         case WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION:
 299                 c->status = wreplsrv_pull_table_wait_connection(state);
 300                 break;
 301         case WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY:
 302                 c->status = wreplsrv_pull_table_wait_table_reply(state);
 303                 c->state  = COMPOSITE_STATE_DONE;
 304                 break;
 305         case WREPLSRV_PULL_TABLE_STAGE_DONE:
 306                 c->status = NT_STATUS_INTERNAL_ERROR;
 307         }
 308
 309         if (!NT_STATUS_IS_OK(c->status)) {
 310                 c->state = COMPOSITE_STATE_ERROR;
 311         }
 312
 313         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
 314                 c->async.fn(c);
 315         }
 316 }
 317
 318 static void wreplsrv_pull_table_handler_creq(struct composite_context *creq)
 319 {
 320         struct wreplsrv_pull_table_state *state = talloc_get_type(creq->async.private_data,
 321                                                   struct wreplsrv_pull_table_state);
 322         wreplsrv_pull_table_handler(state);
 323         return;
 324 }
 325
 326 static void wreplsrv_pull_table_handler_req(struct wrepl_request *req)
 327 {
 328         struct wreplsrv_pull_table_state *state = talloc_get_type(req->async.private,
 329                                                   struct wreplsrv_pull_table_state);
 330         wreplsrv_pull_table_handler(state);
 331         return;
 332 }
 333
 334 static struct composite_context *wreplsrv_pull_table_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_table_io *io)
 335 {
 336         struct composite_context *c = NULL;
 337         struct wreplsrv_service *service = io->in.partner->service;
 338         struct wreplsrv_pull_table_state *state = NULL;
 339
 340         c = talloc_zero(mem_ctx, struct composite_context);
 341         if (!c) goto failed;
 342
 343         state = talloc_zero(c, struct wreplsrv_pull_table_state);
 344         if (!state) goto failed;
 345         state->c        = c;
 346         state->io       = io;
 347
 348         c->state        = COMPOSITE_STATE_IN_PROGRESS;
 349         c->event_ctx    = service->task->event_ctx;
 350         c->private_data = state;
 351
 352         if (io->in.num_owners) {
 353                 state->table_io.out.num_partners        = io->in.num_owners;
 354                 state->table_io.out.partners            = io->in.owners;
 355                 state->stage                            = WREPLSRV_PULL_TABLE_STAGE_DONE;
 356                 composite_done(c);
 357                 return c;
 358         }
 359
 360         state->stage    = WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION;
 361         state->creq     = wreplsrv_out_connect_send(io->in.partner, WINSREPL_PARTNER_PULL, NULL);
 362         if (!state->creq) goto failed;
 363
 364         state->creq->async.fn           = wreplsrv_pull_table_handler_creq;
 365         state->creq->async.private_data = state;
 366
 367         return c;
 368 failed:
 369         talloc_free(c);
 370         return NULL;
 371 }
 372
 373 static NTSTATUS wreplsrv_pull_table_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
 374                                          struct wreplsrv_pull_table_io *io)
 375 {
 376         NTSTATUS status;
 377
 378         status = composite_wait(c);
 379
 380         if (NT_STATUS_IS_OK(status)) {
 381                 struct wreplsrv_pull_table_state *state = talloc_get_type(c->private_data,
 382                                                           struct wreplsrv_pull_table_state);
 383                 io->out.num_owners      = state->table_io.out.num_partners;
 384                 io->out.owners          = talloc_reference(mem_ctx, state->table_io.out.partners);
 385         }
 386
 387         talloc_free(c);
 388         return status;
 389 }
 390
 391 struct wreplsrv_pull_names_io {
 392         struct {
 393                 struct wreplsrv_partner *partner;
 394                 struct wreplsrv_out_connection *wreplconn;
 395                 struct wrepl_wins_owner owner;
 396         } in;
 397         struct {
 398                 uint32_t num_names;
 399                 struct wrepl_name *names;
 400         } out;
 401 };
 402
 403 enum wreplsrv_pull_names_stage {
 404         WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION,
 405         WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY,
 406         WREPLSRV_PULL_NAMES_STAGE_DONE
 407 };
 408
 409 struct wreplsrv_pull_names_state {
 410         enum wreplsrv_pull_names_stage stage;
 411         struct composite_context *c;
 412         struct wrepl_request *req;
 413         struct wrepl_pull_names pull_io;
 414         struct wreplsrv_pull_names_io *io;
 415         struct composite_context *creq;
 416         struct wreplsrv_out_connection *wreplconn;
 417 };
 418
 419 static void wreplsrv_pull_names_handler_req(struct wrepl_request *req);
 420
 421 static NTSTATUS wreplsrv_pull_names_wait_connection(struct wreplsrv_pull_names_state *state)
 422 {
 423         NTSTATUS status;
 424
 425         status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
 426         NT_STATUS_NOT_OK_RETURN(status);
 427
 428         state->pull_io.in.assoc_ctx     = state->wreplconn->assoc_ctx.peer_ctx;
 429         state->pull_io.in.partner       = state->io->in.owner;
 430         state->req = wrepl_pull_names_send(state->wreplconn->sock, &state->pull_io);
 431         NT_STATUS_HAVE_NO_MEMORY(state->req);
 432
 433         state->req->async.fn            = wreplsrv_pull_names_handler_req;
 434         state->req->async.private       = state;
 435
 436         state->stage = WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY;
 437
 438         return NT_STATUS_OK;
 439 }
 440
 441 static NTSTATUS wreplsrv_pull_names_wait_send_reply(struct wreplsrv_pull_names_state *state)
 442 {
 443         NTSTATUS status;
 444
 445         status = wrepl_pull_names_recv(state->req, state, &state->pull_io);
 446         NT_STATUS_NOT_OK_RETURN(status);
 447
 448         state->stage = WREPLSRV_PULL_NAMES_STAGE_DONE;
 449
 450         return NT_STATUS_OK;
 451 }
 452
 453 static void wreplsrv_pull_names_handler(struct wreplsrv_pull_names_state *state)
 454 {
 455         struct composite_context *c = state->c;
 456
 457         switch (state->stage) {
 458         case WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION:
 459                 c->status = wreplsrv_pull_names_wait_connection(state);
 460                 break;
 461         case WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY:
 462                 c->status = wreplsrv_pull_names_wait_send_reply(state);
 463                 c->state  = COMPOSITE_STATE_DONE;
 464                 break;
 465         case WREPLSRV_PULL_NAMES_STAGE_DONE:
 466                 c->status = NT_STATUS_INTERNAL_ERROR;
 467         }
 468
 469         if (!NT_STATUS_IS_OK(c->status)) {
 470                 c->state = COMPOSITE_STATE_ERROR;
 471         }
 472
 473         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
 474                 c->async.fn(c);
 475         }
 476 }
 477
 478 static void wreplsrv_pull_names_handler_creq(struct composite_context *creq)
 479 {
 480         struct wreplsrv_pull_names_state *state = talloc_get_type(creq->async.private_data,
 481                                                   struct wreplsrv_pull_names_state);
 482         wreplsrv_pull_names_handler(state);
 483         return;
 484 }
 485
 486 static void wreplsrv_pull_names_handler_req(struct wrepl_request *req)
 487 {
 488         struct wreplsrv_pull_names_state *state = talloc_get_type(req->async.private,
 489                                                   struct wreplsrv_pull_names_state);
 490         wreplsrv_pull_names_handler(state);
 491         return;
 492 }
 493
 494 static struct composite_context *wreplsrv_pull_names_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_names_io *io)
 495 {
 496         struct composite_context *c = NULL;
 497         struct wreplsrv_service *service = io->in.partner->service;
 498         struct wreplsrv_pull_names_state *state = NULL;
 499         enum winsrepl_partner_type partner_type = WINSREPL_PARTNER_PULL;
 500
 501         if (io->in.wreplconn) partner_type = WINSREPL_PARTNER_NONE;
 502
 503         c = talloc_zero(mem_ctx, struct composite_context);
 504         if (!c) goto failed;
 505
 506         state = talloc_zero(c, struct wreplsrv_pull_names_state);
 507         if (!state) goto failed;
 508         state->c        = c;
 509         state->io       = io;
 510
 511         c->state        = COMPOSITE_STATE_IN_PROGRESS;
 512         c->event_ctx    = service->task->event_ctx;
 513         c->private_data = state;
 514
 515         state->stage    = WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION;
 516         state->creq     = wreplsrv_out_connect_send(io->in.partner, partner_type, io->in.wreplconn);
 517         if (!state->creq) goto failed;
 518
 519         state->creq->async.fn           = wreplsrv_pull_names_handler_creq;
 520         state->creq->async.private_data = state;
 521
 522         return c;
 523 failed:
 524         talloc_free(c);
 525         return NULL;
 526 }
 527
 528 static NTSTATUS wreplsrv_pull_names_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
 529                                          struct wreplsrv_pull_names_io *io)
 530 {
 531         NTSTATUS status;
 532
 533         status = composite_wait(c);
 534
 535         if (NT_STATUS_IS_OK(status)) {
 536                 struct wreplsrv_pull_names_state *state = talloc_get_type(c->private_data,
 537                                                           struct wreplsrv_pull_names_state);
 538                 io->out.num_names       = state->pull_io.out.num_names;
 539                 io->out.names           = 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 }