s3: smbd: named pipe reads are async. Use the same logic as for named pipe transacts...
[Samba.git] / source4 / wrepl_server / wrepl_in_connection.c
blobc68214dcbd248a5ef0ad72ca0735b1531f7360d5
1 /*
2 Unix SMB/CIFS implementation.
4 WINS Replication server
6 Copyright (C) Stefan Metzmacher 2005
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.
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.
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/>.
22 #include "includes.h"
23 #include "lib/socket/socket.h"
24 #include "lib/stream/packet.h"
25 #include "samba/service_task.h"
26 #include "samba/service_stream.h"
27 #include "samba/service.h"
28 #include "lib/messaging/irpc.h"
29 #include "librpc/gen_ndr/ndr_winsrepl.h"
30 #include "wrepl_server/wrepl_server.h"
31 #include "samba/process_model.h"
32 #include "system/network.h"
33 #include "lib/socket/netif.h"
34 #include "lib/tsocket/tsocket.h"
35 #include "libcli/util/tstream.h"
36 #include "param/param.h"
38 void wreplsrv_terminate_in_connection(struct wreplsrv_in_connection *wreplconn, const char *reason)
40 stream_terminate_connection(wreplconn->conn, reason);
44 receive some data on a WREPL connection
46 static NTSTATUS wreplsrv_process(struct wreplsrv_in_connection *wrepl_conn,
47 struct wreplsrv_in_call **_call)
49 struct wrepl_wrap packet_out_wrap;
50 NTSTATUS status;
51 enum ndr_err_code ndr_err;
52 struct wreplsrv_in_call *call = *_call;
54 ndr_err = ndr_pull_struct_blob(&call->in, call,
55 &call->req_packet,
56 (ndr_pull_flags_fn_t)ndr_pull_wrepl_packet);
57 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
58 return ndr_map_error2ntstatus(ndr_err);
61 if (DEBUGLVL(10)) {
62 DEBUG(10,("Received WINS-Replication packet of length %u\n",
63 (unsigned int) call->in.length + 4));
64 NDR_PRINT_DEBUG(wrepl_packet, &call->req_packet);
67 status = wreplsrv_in_call(call);
68 if (NT_STATUS_IS_ERR(status)) {
69 return status;
71 if (!NT_STATUS_IS_OK(status)) {
72 /* w2k just ignores invalid packets, so we do */
73 DEBUG(10,("Received WINS-Replication packet was invalid, we just ignore it\n"));
74 TALLOC_FREE(call);
75 *_call = NULL;
76 return NT_STATUS_OK;
79 /* and now encode the reply */
80 packet_out_wrap.packet = call->rep_packet;
81 ndr_err = ndr_push_struct_blob(&call->out, call,
82 &packet_out_wrap,
83 (ndr_push_flags_fn_t) ndr_push_wrepl_wrap);
84 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
85 return ndr_map_error2ntstatus(ndr_err);
88 if (DEBUGLVL(10)) {
89 DEBUG(10,("Sending WINS-Replication packet of length %u\n",
90 (unsigned int) call->out.length));
91 NDR_PRINT_DEBUG(wrepl_packet, &call->rep_packet);
94 return NT_STATUS_OK;
97 static void wreplsrv_call_loop(struct tevent_req *subreq);
100 called when we get a new connection
102 static void wreplsrv_accept(struct stream_connection *conn)
104 struct wreplsrv_service *service = talloc_get_type(conn->private_data, struct wreplsrv_service);
105 struct wreplsrv_in_connection *wrepl_conn;
106 struct tsocket_address *peer_addr;
107 char *peer_ip;
108 struct tevent_req *subreq;
109 int rc;
111 wrepl_conn = talloc_zero(conn, struct wreplsrv_in_connection);
112 if (wrepl_conn == NULL) {
113 stream_terminate_connection(conn,
114 "wreplsrv_accept: out of memory");
115 return;
118 wrepl_conn->send_queue = tevent_queue_create(conn, "wrepl_accept");
119 if (wrepl_conn->send_queue == NULL) {
120 stream_terminate_connection(conn,
121 "wrepl_accept: out of memory");
122 return;
125 TALLOC_FREE(conn->event.fde);
127 rc = tstream_bsd_existing_socket(wrepl_conn,
128 socket_get_fd(conn->socket),
129 &wrepl_conn->tstream);
130 if (rc < 0) {
131 stream_terminate_connection(conn,
132 "wrepl_accept: out of memory");
133 return;
135 socket_set_flags(conn->socket, SOCKET_FLAG_NOCLOSE);
137 wrepl_conn->conn = conn;
138 wrepl_conn->service = service;
140 peer_addr = conn->remote_address;
142 if (!tsocket_address_is_inet(peer_addr, "ipv4")) {
143 DEBUG(0,("wreplsrv_accept: non ipv4 peer addr '%s'\n",
144 tsocket_address_string(peer_addr, wrepl_conn)));
145 wreplsrv_terminate_in_connection(wrepl_conn, "wreplsrv_accept: "
146 "invalid peer IP");
147 return;
150 peer_ip = tsocket_address_inet_addr_string(peer_addr, wrepl_conn);
151 if (peer_ip == NULL) {
152 wreplsrv_terminate_in_connection(wrepl_conn, "wreplsrv_accept: "
153 "could not convert peer IP into a string");
154 return;
157 wrepl_conn->partner = wreplsrv_find_partner(service, peer_ip);
158 irpc_add_name(conn->msg_ctx, "wreplsrv_connection");
161 * The wrepl pdu's has the length as 4 byte (initial_read_size),
162 * packet_full_request_u32 provides the pdu length then.
164 subreq = tstream_read_pdu_blob_send(wrepl_conn,
165 wrepl_conn->conn->event.ctx,
166 wrepl_conn->tstream,
167 4, /* initial_read_size */
168 packet_full_request_u32,
169 wrepl_conn);
170 if (subreq == NULL) {
171 wreplsrv_terminate_in_connection(wrepl_conn, "wrepl_accept: "
172 "no memory for tstream_read_pdu_blob_send");
173 return;
175 tevent_req_set_callback(subreq, wreplsrv_call_loop, wrepl_conn);
178 static void wreplsrv_call_writev_done(struct tevent_req *subreq);
180 static void wreplsrv_call_loop(struct tevent_req *subreq)
182 struct wreplsrv_in_connection *wrepl_conn = tevent_req_callback_data(subreq,
183 struct wreplsrv_in_connection);
184 struct wreplsrv_in_call *call;
185 NTSTATUS status;
187 call = talloc_zero(wrepl_conn, struct wreplsrv_in_call);
188 if (call == NULL) {
189 wreplsrv_terminate_in_connection(wrepl_conn, "wreplsrv_call_loop: "
190 "no memory for wrepl_samba3_call");
191 return;
193 call->wreplconn = wrepl_conn;
195 status = tstream_read_pdu_blob_recv(subreq,
196 call,
197 &call->in);
198 TALLOC_FREE(subreq);
199 if (!NT_STATUS_IS_OK(status)) {
200 const char *reason;
202 reason = talloc_asprintf(call, "wreplsrv_call_loop: "
203 "tstream_read_pdu_blob_recv() - %s",
204 nt_errstr(status));
205 if (!reason) {
206 reason = nt_errstr(status);
209 wreplsrv_terminate_in_connection(wrepl_conn, reason);
210 return;
213 DEBUG(10,("Received wrepl packet of length %lu from %s\n",
214 (long) call->in.length,
215 tsocket_address_string(wrepl_conn->conn->remote_address, call)));
217 /* skip length header */
218 call->in.data += 4;
219 call->in.length -= 4;
221 status = wreplsrv_process(wrepl_conn, &call);
222 if (!NT_STATUS_IS_OK(status)) {
223 const char *reason;
225 reason = talloc_asprintf(call, "wreplsrv_call_loop: "
226 "tstream_read_pdu_blob_recv() - %s",
227 nt_errstr(status));
228 if (reason == NULL) {
229 reason = nt_errstr(status);
232 wreplsrv_terminate_in_connection(wrepl_conn, reason);
233 return;
236 /* We handed over the connection so we're done here */
237 if (wrepl_conn->tstream == NULL) {
238 return;
241 /* Invalid WINS-Replication packet, we just ignore it */
242 if (call == NULL) {
243 goto noreply;
246 call->out_iov[0].iov_base = (char *) call->out.data;
247 call->out_iov[0].iov_len = call->out.length;
249 subreq = tstream_writev_queue_send(call,
250 wrepl_conn->conn->event.ctx,
251 wrepl_conn->tstream,
252 wrepl_conn->send_queue,
253 call->out_iov, 1);
254 if (subreq == NULL) {
255 wreplsrv_terminate_in_connection(wrepl_conn, "wreplsrv_call_loop: "
256 "no memory for tstream_writev_queue_send");
257 return;
259 tevent_req_set_callback(subreq, wreplsrv_call_writev_done, call);
261 noreply:
263 * The wrepl pdu's has the length as 4 byte (initial_read_size),
264 * provides the pdu length then.
266 subreq = tstream_read_pdu_blob_send(wrepl_conn,
267 wrepl_conn->conn->event.ctx,
268 wrepl_conn->tstream,
269 4, /* initial_read_size */
270 packet_full_request_u32,
271 wrepl_conn);
272 if (subreq == NULL) {
273 wreplsrv_terminate_in_connection(wrepl_conn, "wreplsrv_call_loop: "
274 "no memory for tstream_read_pdu_blob_send");
275 return;
277 tevent_req_set_callback(subreq, wreplsrv_call_loop, wrepl_conn);
280 static void wreplsrv_call_writev_done(struct tevent_req *subreq)
282 struct wreplsrv_in_call *call = tevent_req_callback_data(subreq,
283 struct wreplsrv_in_call);
284 int sys_errno;
285 int rc;
287 rc = tstream_writev_queue_recv(subreq, &sys_errno);
288 TALLOC_FREE(subreq);
289 if (rc == -1) {
290 const char *reason;
292 reason = talloc_asprintf(call, "wreplsrv_call_writev_done: "
293 "tstream_writev_queue_recv() - %d:%s",
294 sys_errno, strerror(sys_errno));
295 if (reason == NULL) {
296 reason = "wreplsrv_call_writev_done: "
297 "tstream_writev_queue_recv() failed";
300 wreplsrv_terminate_in_connection(call->wreplconn, reason);
301 return;
304 if (call->terminate_after_send) {
305 wreplsrv_terminate_in_connection(call->wreplconn,
306 "wreplsrv_in_connection: terminate_after_send");
307 return;
310 talloc_free(call);
314 called on a tcp recv
316 static void wreplsrv_recv(struct stream_connection *conn, uint16_t flags)
318 struct wreplsrv_in_connection *wrepl_conn = talloc_get_type(conn->private_data,
319 struct wreplsrv_in_connection);
320 /* this should never be triggered! */
321 DEBUG(0,("Terminating connection - '%s'\n", "wrepl_recv: called"));
322 wreplsrv_terminate_in_connection(wrepl_conn, "wrepl_recv: called");
326 called when we can write to a connection
328 static void wreplsrv_send(struct stream_connection *conn, uint16_t flags)
330 struct wreplsrv_in_connection *wrepl_conn = talloc_get_type(conn->private_data,
331 struct wreplsrv_in_connection);
332 /* this should never be triggered! */
333 DEBUG(0,("Terminating connection - '%s'\n", "wrepl_send: called"));
334 wreplsrv_terminate_in_connection(wrepl_conn, "wrepl_send: called");
337 static const struct stream_server_ops wreplsrv_stream_ops = {
338 .name = "wreplsrv",
339 .accept_connection = wreplsrv_accept,
340 .recv_handler = wreplsrv_recv,
341 .send_handler = wreplsrv_send,
345 called when we get a new connection
347 NTSTATUS wreplsrv_in_connection_merge(struct wreplsrv_partner *partner,
348 uint32_t peer_assoc_ctx,
349 struct tstream_context **stream,
350 struct wreplsrv_in_connection **_wrepl_in,
351 void* process_context)
353 struct wreplsrv_service *service = partner->service;
354 struct wreplsrv_in_connection *wrepl_in;
355 struct stream_connection *conn;
356 struct tevent_req *subreq;
357 NTSTATUS status;
359 wrepl_in = talloc_zero(partner, struct wreplsrv_in_connection);
360 NT_STATUS_HAVE_NO_MEMORY(wrepl_in);
362 wrepl_in->service = service;
363 wrepl_in->partner = partner;
364 wrepl_in->tstream = talloc_move(wrepl_in, stream);
365 wrepl_in->assoc_ctx.peer_ctx = peer_assoc_ctx;
367 status = stream_new_connection_merge(service->task->event_ctx,
368 service->task->lp_ctx,
369 service->task->model_ops,
370 &wreplsrv_stream_ops,
371 service->task->msg_ctx,
372 wrepl_in,
373 &conn,
374 process_context);
375 NT_STATUS_NOT_OK_RETURN(status);
378 * make the wreplsrv_in_connection structure a child of the
379 * stream_connection, to match the hierarchy of wreplsrv_accept
381 wrepl_in->conn = conn;
382 talloc_steal(conn, wrepl_in);
384 wrepl_in->send_queue = tevent_queue_create(wrepl_in, "wreplsrv_in_connection_merge");
385 if (wrepl_in->send_queue == NULL) {
386 stream_terminate_connection(conn,
387 "wreplsrv_in_connection_merge: out of memory");
388 return NT_STATUS_NO_MEMORY;
392 * The wrepl pdu's has the length as 4 byte (initial_read_size),
393 * packet_full_request_u32 provides the pdu length then.
395 subreq = tstream_read_pdu_blob_send(wrepl_in,
396 wrepl_in->conn->event.ctx,
397 wrepl_in->tstream,
398 4, /* initial_read_size */
399 packet_full_request_u32,
400 wrepl_in);
401 if (subreq == NULL) {
402 wreplsrv_terminate_in_connection(wrepl_in, "wreplsrv_in_connection_merge: "
403 "no memory for tstream_read_pdu_blob_send");
404 return NT_STATUS_NO_MEMORY;
406 tevent_req_set_callback(subreq, wreplsrv_call_loop, wrepl_in);
408 *_wrepl_in = wrepl_in;
410 return NT_STATUS_OK;
414 startup the wrepl port 42 server sockets
416 NTSTATUS wreplsrv_setup_sockets(struct wreplsrv_service *service, struct loadparm_context *lp_ctx)
418 NTSTATUS status;
419 struct task_server *task = service->task;
420 const char *address;
421 uint16_t port = WINS_REPLICATION_PORT;
423 if (lpcfg_interfaces(lp_ctx) && lpcfg_bind_interfaces_only(lp_ctx)) {
424 int num_interfaces;
425 int i;
426 struct interface *ifaces;
428 load_interface_list(task, lp_ctx, &ifaces);
430 num_interfaces = iface_list_count(ifaces);
432 /* We have been given an interfaces line, and been
433 told to only bind to those interfaces. Create a
434 socket per interface and bind to only these.
436 for(i = 0; i < num_interfaces; i++) {
437 if (!iface_list_n_is_v4(ifaces, i)) {
438 continue;
440 address = iface_list_n_ip(ifaces, i);
441 status = stream_setup_socket(
442 task, task->event_ctx,
443 task->lp_ctx,
444 task->model_ops,
445 &wreplsrv_stream_ops,
446 "ipv4", address, &port,
447 lpcfg_socket_options(task->lp_ctx),
448 service, task->process_context);
449 if (!NT_STATUS_IS_OK(status)) {
450 DEBUG(0,("stream_setup_socket(address=%s,port=%u) failed - %s\n",
451 address, port, nt_errstr(status)));
452 return status;
455 } else {
456 address = "0.0.0.0";
457 status = stream_setup_socket(task, task->event_ctx,
458 task->lp_ctx, task->model_ops,
459 &wreplsrv_stream_ops,
460 "ipv4", address, &port,
461 lpcfg_socket_options(task->lp_ctx),
462 service, task->process_context);
463 if (!NT_STATUS_IS_OK(status)) {
464 DEBUG(0,("stream_setup_socket(address=%s,port=%u) failed - %s\n",
465 address, port, nt_errstr(status)));
466 return status;
470 return NT_STATUS_OK;