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