2 Unix SMB/CIFS implementation.
4 Samba internal messaging functions
6 Copyright (C) Andrew Tridgell 2004
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/>.
23 #include "lib/events/events.h"
24 #include "system/filesys.h"
25 #include "messaging/messaging.h"
26 #include "../lib/util/dlinklist.h"
27 #include "lib/socket/socket.h"
28 #include "librpc/gen_ndr/ndr_irpc.h"
29 #include "lib/messaging/irpc.h"
30 #include "lib/tdb_wrap/tdb_wrap.h"
31 #include "../lib/util/unix_privs.h"
32 #include "librpc/rpc/dcerpc.h"
33 #include "../lib/tdb_compat/tdb_compat.h"
34 #include "../lib/util/util_tdb.h"
35 #include "cluster/cluster.h"
36 #include "../lib/util/tevent_ntstatus.h"
37 #include "lib/param/param.h"
39 /* change the message version with any incompatible changes in the protocol */
40 #define IMESSAGING_VERSION 1
46 struct imessaging_context
*msg_ctx
;
49 void (*handler
)(struct irpc_request
*irpc
, struct irpc_message
*m
);
54 struct imessaging_context
{
55 struct server_id server_id
;
56 struct socket_context
*sock
;
57 const char *base_path
;
59 struct loadparm_context
*lp_ctx
;
60 struct dispatch_fn
**dispatch
;
62 struct idr_context
*dispatch_tree
;
63 struct imessaging_rec
*pending
;
64 struct imessaging_rec
*retry_queue
;
65 struct irpc_list
*irpc
;
66 struct idr_context
*idr
;
68 struct timeval start_time
;
69 struct tevent_timer
*retry_te
;
71 struct tevent_context
*ev
;
72 struct tevent_fd
*fde
;
76 /* we have a linked list of dispatch handlers for each msg_type that
77 this messaging server can deal with */
79 struct dispatch_fn
*next
, *prev
;
85 /* an individual message */
86 struct imessaging_rec
{
87 struct imessaging_rec
*next
, *prev
;
88 struct imessaging_context
*msg
;
91 struct imessaging_header
{
94 struct server_id from
;
104 static void irpc_handler(struct imessaging_context
*, void *,
105 uint32_t, struct server_id
, DATA_BLOB
*);
109 A useful function for testing the message system.
111 static void ping_message(struct imessaging_context
*msg
, void *private_data
,
112 uint32_t msg_type
, struct server_id src
, DATA_BLOB
*data
)
114 char *task_id
= server_id_str(NULL
, &src
);
115 DEBUG(1,("INFO: Received PING message from server %s [%.*s]\n",
116 task_id
, (int)data
->length
,
117 data
->data
?(const char *)data
->data
:""));
118 talloc_free(task_id
);
119 imessaging_send(msg
, src
, MSG_PONG
, data
);
123 return uptime of messaging server via irpc
125 static NTSTATUS
irpc_uptime(struct irpc_message
*msg
,
126 struct irpc_uptime
*r
)
128 struct imessaging_context
*ctx
= talloc_get_type(msg
->private_data
, struct imessaging_context
);
129 *r
->out
.start_time
= timeval_to_nttime(&ctx
->start_time
);
134 return the path to a messaging socket
136 static char *imessaging_path(struct imessaging_context
*msg
, struct server_id server_id
)
138 TALLOC_CTX
*tmp_ctx
= talloc_new(msg
);
139 const char *id
= server_id_str(tmp_ctx
, &server_id
);
144 s
= talloc_asprintf(msg
, "%s/msg.%s", msg
->base_path
, id
);
145 talloc_steal(s
, tmp_ctx
);
150 dispatch a fully received message
152 note that this deliberately can match more than one message handler
153 per message. That allows a single messasging context to register
154 (for example) a debug handler for more than one piece of code
156 static void imessaging_dispatch(struct imessaging_context
*msg
, struct imessaging_rec
*rec
)
158 struct dispatch_fn
*d
, *next
;
160 /* temporary IDs use an idtree, the rest use a array of pointers */
161 if (rec
->header
->msg_type
>= MSG_TMP_BASE
) {
162 d
= (struct dispatch_fn
*)idr_find(msg
->dispatch_tree
,
163 rec
->header
->msg_type
);
164 } else if (rec
->header
->msg_type
< msg
->num_types
) {
165 d
= msg
->dispatch
[rec
->header
->msg_type
];
170 for (; d
; d
= next
) {
173 data
.data
= rec
->packet
.data
+ sizeof(*rec
->header
);
174 data
.length
= rec
->header
->length
;
175 d
->fn(msg
, d
->private_data
, d
->msg_type
, rec
->header
->from
, &data
);
177 rec
->header
->length
= 0;
181 handler for messages that arrive from other nodes in the cluster
183 static void cluster_message_handler(struct imessaging_context
*msg
, DATA_BLOB packet
)
185 struct imessaging_rec
*rec
;
187 rec
= talloc(msg
, struct imessaging_rec
);
189 smb_panic("Unable to allocate imessaging_rec");
193 rec
->path
= msg
->path
;
194 rec
->header
= (struct imessaging_header
*)packet
.data
;
195 rec
->packet
= packet
;
198 if (packet
.length
!= sizeof(*rec
->header
) + rec
->header
->length
) {
199 DEBUG(0,("messaging: bad message header size %d should be %d\n",
200 rec
->header
->length
, (int)(packet
.length
- sizeof(*rec
->header
))));
205 imessaging_dispatch(msg
, rec
);
212 try to send the message
214 static NTSTATUS
try_send(struct imessaging_rec
*rec
)
216 struct imessaging_context
*msg
= rec
->msg
;
220 struct socket_address
*path
;
222 /* rec->path is the path of the *other* socket, where we want
224 path
= socket_address_from_strings(msg
, msg
->sock
->backend_name
,
227 return NT_STATUS_NO_MEMORY
;
230 /* we send with privileges so messages work from any context */
231 priv
= root_privileges();
232 status
= socket_sendto(msg
->sock
, &rec
->packet
, &nsent
, path
);
240 retry backed off messages
242 static void msg_retry_timer(struct tevent_context
*ev
, struct tevent_timer
*te
,
243 struct timeval t
, void *private_data
)
245 struct imessaging_context
*msg
= talloc_get_type(private_data
,
246 struct imessaging_context
);
247 msg
->retry_te
= NULL
;
249 /* put the messages back on the main queue */
250 while (msg
->retry_queue
) {
251 struct imessaging_rec
*rec
= msg
->retry_queue
;
252 DLIST_REMOVE(msg
->retry_queue
, rec
);
253 DLIST_ADD_END(msg
->pending
, rec
, struct imessaging_rec
*);
256 TEVENT_FD_WRITEABLE(msg
->event
.fde
);
260 handle a socket write event
262 static void imessaging_send_handler(struct imessaging_context
*msg
)
264 while (msg
->pending
) {
265 struct imessaging_rec
*rec
= msg
->pending
;
267 status
= try_send(rec
);
268 if (NT_STATUS_EQUAL(status
, STATUS_MORE_ENTRIES
)) {
270 if (rec
->retries
> 3) {
271 /* we're getting continuous write errors -
272 backoff this record */
273 DLIST_REMOVE(msg
->pending
, rec
);
274 DLIST_ADD_END(msg
->retry_queue
, rec
,
275 struct imessaging_rec
*);
276 if (msg
->retry_te
== NULL
) {
278 tevent_add_timer(msg
->event
.ev
, msg
,
279 timeval_current_ofs(1, 0),
280 msg_retry_timer
, msg
);
286 if (!NT_STATUS_IS_OK(status
)) {
287 TALLOC_CTX
*tmp_ctx
= talloc_new(msg
);
288 DEBUG(1,("messaging: Lost message from %s to %s of type %u - %s\n",
289 server_id_str(tmp_ctx
, &rec
->header
->from
),
290 server_id_str(tmp_ctx
, &rec
->header
->to
),
291 rec
->header
->msg_type
,
293 talloc_free(tmp_ctx
);
295 DLIST_REMOVE(msg
->pending
, rec
);
298 if (msg
->pending
== NULL
) {
299 TEVENT_FD_NOT_WRITEABLE(msg
->event
.fde
);
304 handle a new incoming packet
306 static void imessaging_recv_handler(struct imessaging_context
*msg
)
308 struct imessaging_rec
*rec
;
313 /* see how many bytes are in the next packet */
314 status
= socket_pending(msg
->sock
, &msize
);
315 if (!NT_STATUS_IS_OK(status
)) {
316 DEBUG(0,("socket_pending failed in messaging - %s\n",
321 packet
= data_blob_talloc(msg
, NULL
, msize
);
322 if (packet
.data
== NULL
) {
323 /* assume this is temporary and retry */
327 status
= socket_recv(msg
->sock
, packet
.data
, msize
, &msize
);
328 if (!NT_STATUS_IS_OK(status
)) {
329 data_blob_free(&packet
);
333 if (msize
< sizeof(*rec
->header
)) {
334 DEBUG(0,("messaging: bad message of size %d\n", (int)msize
));
335 data_blob_free(&packet
);
339 rec
= talloc(msg
, struct imessaging_rec
);
341 smb_panic("Unable to allocate imessaging_rec");
344 talloc_steal(rec
, packet
.data
);
346 rec
->path
= msg
->path
;
347 rec
->header
= (struct imessaging_header
*)packet
.data
;
348 rec
->packet
= packet
;
351 if (msize
!= sizeof(*rec
->header
) + rec
->header
->length
) {
352 DEBUG(0,("messaging: bad message header size %d should be %d\n",
353 rec
->header
->length
, (int)(msize
- sizeof(*rec
->header
))));
358 imessaging_dispatch(msg
, rec
);
364 handle a socket event
366 static void imessaging_handler(struct tevent_context
*ev
, struct tevent_fd
*fde
,
367 uint16_t flags
, void *private_data
)
369 struct imessaging_context
*msg
= talloc_get_type(private_data
,
370 struct imessaging_context
);
371 if (flags
& TEVENT_FD_WRITE
) {
372 imessaging_send_handler(msg
);
374 if (flags
& TEVENT_FD_READ
) {
375 imessaging_recv_handler(msg
);
381 Register a dispatch function for a particular message type.
383 NTSTATUS
imessaging_register(struct imessaging_context
*msg
, void *private_data
,
384 uint32_t msg_type
, msg_callback_t fn
)
386 struct dispatch_fn
*d
;
388 /* possibly expand dispatch array */
389 if (msg_type
>= msg
->num_types
) {
390 struct dispatch_fn
**dp
;
392 dp
= talloc_realloc(msg
, msg
->dispatch
, struct dispatch_fn
*, msg_type
+1);
393 NT_STATUS_HAVE_NO_MEMORY(dp
);
395 for (i
=msg
->num_types
;i
<=msg_type
;i
++) {
396 msg
->dispatch
[i
] = NULL
;
398 msg
->num_types
= msg_type
+1;
401 d
= talloc_zero(msg
->dispatch
, struct dispatch_fn
);
402 NT_STATUS_HAVE_NO_MEMORY(d
);
403 d
->msg_type
= msg_type
;
404 d
->private_data
= private_data
;
407 DLIST_ADD(msg
->dispatch
[msg_type
], d
);
413 register a temporary message handler. The msg_type is allocated
416 NTSTATUS
imessaging_register_tmp(struct imessaging_context
*msg
, void *private_data
,
417 msg_callback_t fn
, uint32_t *msg_type
)
419 struct dispatch_fn
*d
;
422 d
= talloc_zero(msg
->dispatch
, struct dispatch_fn
);
423 NT_STATUS_HAVE_NO_MEMORY(d
);
424 d
->private_data
= private_data
;
427 id
= idr_get_new_above(msg
->dispatch_tree
, d
, MSG_TMP_BASE
, UINT16_MAX
);
430 return NT_STATUS_TOO_MANY_CONTEXT_IDS
;
433 d
->msg_type
= (uint32_t)id
;
434 (*msg_type
) = d
->msg_type
;
440 De-register the function for a particular message type.
442 void imessaging_deregister(struct imessaging_context
*msg
, uint32_t msg_type
, void *private_data
)
444 struct dispatch_fn
*d
, *next
;
446 if (msg_type
>= msg
->num_types
) {
447 d
= (struct dispatch_fn
*)idr_find(msg
->dispatch_tree
,
450 idr_remove(msg
->dispatch_tree
, msg_type
);
455 for (d
= msg
->dispatch
[msg_type
]; d
; d
= next
) {
457 if (d
->private_data
== private_data
) {
458 DLIST_REMOVE(msg
->dispatch
[msg_type
], d
);
465 Send a message to a particular server
467 NTSTATUS
imessaging_send(struct imessaging_context
*msg
, struct server_id server
,
468 uint32_t msg_type
, const DATA_BLOB
*data
)
470 struct imessaging_rec
*rec
;
472 size_t dlength
= data
?data
->length
:0;
474 rec
= talloc(msg
, struct imessaging_rec
);
476 return NT_STATUS_NO_MEMORY
;
479 rec
->packet
= data_blob_talloc(rec
, NULL
, sizeof(*rec
->header
) + dlength
);
480 if (rec
->packet
.data
== NULL
) {
482 return NT_STATUS_NO_MEMORY
;
487 rec
->header
= (struct imessaging_header
*)rec
->packet
.data
;
489 ZERO_STRUCTP(rec
->header
);
490 rec
->header
->version
= IMESSAGING_VERSION
;
491 rec
->header
->msg_type
= msg_type
;
492 rec
->header
->from
= msg
->server_id
;
493 rec
->header
->to
= server
;
494 rec
->header
->length
= dlength
;
496 memcpy(rec
->packet
.data
+ sizeof(*rec
->header
),
497 data
->data
, dlength
);
500 if (!cluster_node_equal(&msg
->server_id
, &server
)) {
501 /* the destination is on another node - dispatch via
503 status
= cluster_message_send(server
, &rec
->packet
);
508 rec
->path
= imessaging_path(msg
, server
);
509 talloc_steal(rec
, rec
->path
);
511 if (msg
->pending
!= NULL
) {
512 status
= STATUS_MORE_ENTRIES
;
514 status
= try_send(rec
);
517 if (NT_STATUS_EQUAL(status
, STATUS_MORE_ENTRIES
)) {
518 if (msg
->pending
== NULL
) {
519 TEVENT_FD_WRITEABLE(msg
->event
.fde
);
521 DLIST_ADD_END(msg
->pending
, rec
, struct imessaging_rec
*);
531 Send a message to a particular server, with the message containing a single pointer
533 NTSTATUS
imessaging_send_ptr(struct imessaging_context
*msg
, struct server_id server
,
534 uint32_t msg_type
, void *ptr
)
538 blob
.data
= (uint8_t *)&ptr
;
539 blob
.length
= sizeof(void *);
541 return imessaging_send(msg
, server
, msg_type
, &blob
);
546 remove our messaging socket and database entry
548 int imessaging_cleanup(struct imessaging_context
*msg
)
554 DEBUG(5,("imessaging: cleaning up %s\n", msg
->path
));
556 while (msg
->names
&& msg
->names
[0]) {
557 irpc_remove_name(msg
, msg
->names
[0]);
563 create the listening socket and setup the dispatcher
565 use temporary=true when you want a destructor to remove the
566 associated messaging socket and database entry on talloc free. Don't
567 use this in processes that may fork and a child may talloc free this
570 struct imessaging_context
*imessaging_init(TALLOC_CTX
*mem_ctx
,
571 struct loadparm_context
*lp_ctx
,
572 struct server_id server_id
,
573 struct tevent_context
*ev
,
576 struct imessaging_context
*msg
;
578 struct socket_address
*path
;
584 msg
= talloc_zero(mem_ctx
, struct imessaging_context
);
589 /* setup a handler for messages from other cluster nodes, if appropriate */
590 status
= cluster_message_init(msg
, server_id
, cluster_message_handler
);
591 if (!NT_STATUS_IS_OK(status
)) {
596 /* create the messaging directory if needed */
598 msg
->lp_ctx
= talloc_reference(msg
, lp_ctx
);
604 msg
->base_path
= lpcfg_imessaging_path(msg
, lp_ctx
);
606 mkdir(msg
->base_path
, 0700);
608 msg
->path
= imessaging_path(msg
, server_id
);
609 msg
->server_id
= server_id
;
610 msg
->idr
= idr_init(msg
);
611 msg
->dispatch_tree
= idr_init(msg
);
612 msg
->start_time
= timeval_current();
614 status
= socket_create("unix", SOCKET_TYPE_DGRAM
, &msg
->sock
, 0);
615 if (!NT_STATUS_IS_OK(status
)) {
620 /* by stealing here we ensure that the socket is cleaned up (and even
622 talloc_steal(msg
, msg
->sock
);
624 path
= socket_address_from_strings(msg
, msg
->sock
->backend_name
,
631 status
= socket_listen(msg
->sock
, path
, 50, 0);
632 if (!NT_STATUS_IS_OK(status
)) {
633 DEBUG(0,("Unable to setup messaging listener for '%s':%s\n", msg
->path
, nt_errstr(status
)));
638 /* it needs to be non blocking for sends */
639 set_blocking(socket_get_fd(msg
->sock
), false);
642 msg
->event
.fde
= tevent_add_fd(ev
, msg
, socket_get_fd(msg
->sock
),
643 TEVENT_FD_READ
, imessaging_handler
, msg
);
644 tevent_fd_set_auto_close(msg
->event
.fde
);
647 talloc_set_destructor(msg
, imessaging_cleanup
);
650 imessaging_register(msg
, NULL
, MSG_PING
, ping_message
);
651 imessaging_register(msg
, NULL
, MSG_IRPC
, irpc_handler
);
652 IRPC_REGISTER(msg
, irpc
, IRPC_UPTIME
, irpc_uptime
, msg
);
658 A hack, for the short term until we get 'client only' messaging in place
660 struct imessaging_context
*imessaging_client_init(TALLOC_CTX
*mem_ctx
,
661 struct loadparm_context
*lp_ctx
,
662 struct tevent_context
*ev
)
667 id
.task_id
= generate_random();
668 id
.vnn
= NONCLUSTER_VNN
;
670 /* This is because we are not in the s3 serverid database */
671 id
.unique_id
= SERVERID_UNIQUE_ID_NOT_TO_VERIFY
;
673 return imessaging_init(mem_ctx
, lp_ctx
, id
, ev
, true);
676 a list of registered irpc server functions
679 struct irpc_list
*next
, *prev
;
681 const struct ndr_interface_table
*table
;
689 register a irpc server function
691 NTSTATUS
irpc_register(struct imessaging_context
*msg_ctx
,
692 const struct ndr_interface_table
*table
,
693 int callnum
, irpc_function_t fn
, void *private_data
)
695 struct irpc_list
*irpc
;
697 /* override an existing handler, if any */
698 for (irpc
=msg_ctx
->irpc
; irpc
; irpc
=irpc
->next
) {
699 if (irpc
->table
== table
&& irpc
->callnum
== callnum
) {
704 irpc
= talloc(msg_ctx
, struct irpc_list
);
705 NT_STATUS_HAVE_NO_MEMORY(irpc
);
706 DLIST_ADD(msg_ctx
->irpc
, irpc
);
710 irpc
->callnum
= callnum
;
712 irpc
->private_data
= private_data
;
713 irpc
->uuid
= irpc
->table
->syntax_id
.uuid
;
720 handle an incoming irpc reply message
722 static void irpc_handler_reply(struct imessaging_context
*msg_ctx
, struct irpc_message
*m
)
724 struct irpc_request
*irpc
;
726 irpc
= (struct irpc_request
*)idr_find(msg_ctx
->idr
, m
->header
.callid
);
727 if (irpc
== NULL
) return;
729 irpc
->incoming
.handler(irpc
, m
);
735 NTSTATUS
irpc_send_reply(struct irpc_message
*m
, NTSTATUS status
)
737 struct ndr_push
*push
;
739 enum ndr_err_code ndr_err
;
741 m
->header
.status
= status
;
743 /* setup the reply */
744 push
= ndr_push_init_ctx(m
->ndr
);
746 status
= NT_STATUS_NO_MEMORY
;
750 m
->header
.flags
|= IRPC_FLAG_REPLY
;
751 m
->header
.creds
.token
= NULL
;
753 /* construct the packet */
754 ndr_err
= ndr_push_irpc_header(push
, NDR_SCALARS
|NDR_BUFFERS
, &m
->header
);
755 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err
)) {
756 status
= ndr_map_error2ntstatus(ndr_err
);
760 ndr_err
= m
->irpc
->table
->calls
[m
->irpc
->callnum
].ndr_push(push
, NDR_OUT
, m
->data
);
761 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err
)) {
762 status
= ndr_map_error2ntstatus(ndr_err
);
766 /* send the reply message */
767 packet
= ndr_push_blob(push
);
768 status
= imessaging_send(m
->msg_ctx
, m
->from
, MSG_IRPC
, &packet
);
769 if (!NT_STATUS_IS_OK(status
)) goto failed
;
777 handle an incoming irpc request message
779 static void irpc_handler_request(struct imessaging_context
*msg_ctx
,
780 struct irpc_message
*m
)
784 enum ndr_err_code ndr_err
;
786 for (i
=msg_ctx
->irpc
; i
; i
=i
->next
) {
787 if (GUID_equal(&i
->uuid
, &m
->header
.uuid
) &&
788 i
->table
->syntax_id
.if_version
== m
->header
.if_version
&&
789 i
->callnum
== m
->header
.callnum
) {
795 /* no registered handler for this message */
800 /* allocate space for the structure */
801 r
= talloc_zero_size(m
->ndr
, i
->table
->calls
[m
->header
.callnum
].struct_size
);
802 if (r
== NULL
) goto failed
;
804 m
->ndr
->flags
|= LIBNDR_FLAG_REF_ALLOC
;
806 /* parse the request data */
807 ndr_err
= i
->table
->calls
[i
->callnum
].ndr_pull(m
->ndr
, NDR_IN
, r
);
808 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err
)) goto failed
;
811 m
->private_data
= i
->private_data
;
812 m
->defer_reply
= false;
814 m
->msg_ctx
= msg_ctx
;
817 m
->ev
= msg_ctx
->event
.ev
;
819 m
->header
.status
= i
->fn(m
, r
);
822 /* the server function won't ever be replying to this request */
827 if (m
->defer_reply
) {
828 /* the server function has asked to defer the reply to later */
829 talloc_steal(msg_ctx
, m
);
833 irpc_send_reply(m
, m
->header
.status
);
841 handle an incoming irpc message
843 static void irpc_handler(struct imessaging_context
*msg_ctx
, void *private_data
,
844 uint32_t msg_type
, struct server_id src
, DATA_BLOB
*packet
)
846 struct irpc_message
*m
;
847 enum ndr_err_code ndr_err
;
849 m
= talloc(msg_ctx
, struct irpc_message
);
850 if (m
== NULL
) goto failed
;
854 m
->ndr
= ndr_pull_init_blob(packet
, m
);
855 if (m
->ndr
== NULL
) goto failed
;
857 m
->ndr
->flags
|= LIBNDR_FLAG_REF_ALLOC
;
859 ndr_err
= ndr_pull_irpc_header(m
->ndr
, NDR_BUFFERS
|NDR_SCALARS
, &m
->header
);
860 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err
)) goto failed
;
862 if (m
->header
.flags
& IRPC_FLAG_REPLY
) {
863 irpc_handler_reply(msg_ctx
, m
);
865 irpc_handler_request(msg_ctx
, m
);
875 destroy a irpc request
877 static int irpc_destructor(struct irpc_request
*irpc
)
879 if (irpc
->callid
!= -1) {
880 idr_remove(irpc
->msg_ctx
->idr
, irpc
->callid
);
888 open the naming database
890 static struct tdb_wrap
*irpc_namedb_open(struct imessaging_context
*msg_ctx
)
893 char *path
= talloc_asprintf(msg_ctx
, "%s/names.tdb", msg_ctx
->base_path
);
897 t
= tdb_wrap_open(msg_ctx
, path
, 0, 0, O_RDWR
|O_CREAT
, 0660, msg_ctx
->lp_ctx
);
904 add a string name that this irpc server can be called on
906 NTSTATUS
irpc_add_name(struct imessaging_context
*msg_ctx
, const char *name
)
911 NTSTATUS status
= NT_STATUS_OK
;
913 t
= irpc_namedb_open(msg_ctx
);
914 NT_STATUS_HAVE_NO_MEMORY(t
);
916 if (tdb_lock_bystring(t
->tdb
, name
) != 0) {
918 return NT_STATUS_LOCK_NOT_GRANTED
;
920 rec
= tdb_fetch_bystring(t
->tdb
, name
);
921 count
= rec
.dsize
/ sizeof(struct server_id
);
922 rec
.dptr
= (unsigned char *)realloc_p(rec
.dptr
, struct server_id
, count
+1);
923 rec
.dsize
+= sizeof(struct server_id
);
924 if (rec
.dptr
== NULL
) {
925 tdb_unlock_bystring(t
->tdb
, name
);
927 return NT_STATUS_NO_MEMORY
;
929 ((struct server_id
*)rec
.dptr
)[count
] = msg_ctx
->server_id
;
930 if (tdb_store_bystring(t
->tdb
, name
, rec
, 0) != 0) {
931 status
= NT_STATUS_INTERNAL_ERROR
;
934 tdb_unlock_bystring(t
->tdb
, name
);
937 msg_ctx
->names
= str_list_add(msg_ctx
->names
, name
);
938 talloc_steal(msg_ctx
, msg_ctx
->names
);
944 return a list of server ids for a server name
946 struct server_id
*irpc_servers_byname(struct imessaging_context
*msg_ctx
,
953 struct server_id
*ret
;
955 t
= irpc_namedb_open(msg_ctx
);
960 if (tdb_lock_bystring(t
->tdb
, name
) != 0) {
964 rec
= tdb_fetch_bystring(t
->tdb
, name
);
965 if (rec
.dptr
== NULL
) {
966 tdb_unlock_bystring(t
->tdb
, name
);
970 count
= rec
.dsize
/ sizeof(struct server_id
);
971 ret
= talloc_array(mem_ctx
, struct server_id
, count
+1);
973 tdb_unlock_bystring(t
->tdb
, name
);
977 for (i
=0;i
<count
;i
++) {
978 ret
[i
] = ((struct server_id
*)rec
.dptr
)[i
];
980 ret
[i
] = cluster_id(0, 0);
982 tdb_unlock_bystring(t
->tdb
, name
);
988 static int all_servers_func(struct tdb_context
*tdb
, TDB_DATA key
, TDB_DATA data
, void *state
)
990 struct irpc_name_records
*name_records
= talloc_get_type(state
, struct irpc_name_records
);
991 struct irpc_name_record
*name_record
;
995 = talloc_realloc(name_records
, name_records
->names
,
996 struct irpc_name_record
*, name_records
->num_records
+1);
997 if (!name_records
->names
) {
1001 name_records
->names
[name_records
->num_records
] = name_record
1002 = talloc(name_records
->names
,
1003 struct irpc_name_record
);
1008 name_records
->num_records
++;
1011 = talloc_strndup(name_record
,
1012 (const char *)key
.dptr
, key
.dsize
);
1013 if (!name_record
->name
) {
1017 name_record
->count
= data
.dsize
/ sizeof(struct server_id
);
1018 name_record
->ids
= talloc_array(name_record
,
1020 name_record
->count
);
1021 if (name_record
->ids
== NULL
) {
1024 for (i
=0;i
<name_record
->count
;i
++) {
1025 name_record
->ids
[i
] = ((struct server_id
*)data
.dptr
)[i
];
1031 return a list of server ids for a server name
1033 struct irpc_name_records
*irpc_all_servers(struct imessaging_context
*msg_ctx
,
1034 TALLOC_CTX
*mem_ctx
)
1038 struct irpc_name_records
*name_records
= talloc_zero(mem_ctx
, struct irpc_name_records
);
1039 if (name_records
== NULL
) {
1043 t
= irpc_namedb_open(msg_ctx
);
1048 ret
= tdb_traverse_read(t
->tdb
, all_servers_func
, name_records
);
1056 return name_records
;
1060 remove a name from a messaging context
1062 void irpc_remove_name(struct imessaging_context
*msg_ctx
, const char *name
)
1067 struct server_id
*ids
;
1069 str_list_remove(msg_ctx
->names
, name
);
1071 t
= irpc_namedb_open(msg_ctx
);
1076 if (tdb_lock_bystring(t
->tdb
, name
) != 0) {
1080 rec
= tdb_fetch_bystring(t
->tdb
, name
);
1081 if (rec
.dptr
== NULL
) {
1082 tdb_unlock_bystring(t
->tdb
, name
);
1086 count
= rec
.dsize
/ sizeof(struct server_id
);
1089 tdb_unlock_bystring(t
->tdb
, name
);
1093 ids
= (struct server_id
*)rec
.dptr
;
1094 for (i
=0;i
<count
;i
++) {
1095 if (cluster_id_equal(&ids
[i
], &msg_ctx
->server_id
)) {
1097 memmove(ids
+i
, ids
+i
+1,
1098 sizeof(struct server_id
) * (count
-(i
+1)));
1100 rec
.dsize
-= sizeof(struct server_id
);
1104 tdb_store_bystring(t
->tdb
, name
, rec
, 0);
1106 tdb_unlock_bystring(t
->tdb
, name
);
1110 struct server_id
imessaging_get_server_id(struct imessaging_context
*msg_ctx
)
1112 return msg_ctx
->server_id
;
1115 struct irpc_bh_state
{
1116 struct imessaging_context
*msg_ctx
;
1117 struct server_id server_id
;
1118 const struct ndr_interface_table
*table
;
1120 struct security_token
*token
;
1123 static bool irpc_bh_is_connected(struct dcerpc_binding_handle
*h
)
1125 struct irpc_bh_state
*hs
= dcerpc_binding_handle_data(h
,
1126 struct irpc_bh_state
);
1135 static uint32_t irpc_bh_set_timeout(struct dcerpc_binding_handle
*h
,
1138 struct irpc_bh_state
*hs
= dcerpc_binding_handle_data(h
,
1139 struct irpc_bh_state
);
1140 uint32_t old
= hs
->timeout
;
1142 hs
->timeout
= timeout
;
1147 struct irpc_bh_raw_call_state
{
1148 struct irpc_request
*irpc
;
1151 DATA_BLOB in_packet
;
1155 static void irpc_bh_raw_call_incoming_handler(struct irpc_request
*irpc
,
1156 struct irpc_message
*m
);
1158 static struct tevent_req
*irpc_bh_raw_call_send(TALLOC_CTX
*mem_ctx
,
1159 struct tevent_context
*ev
,
1160 struct dcerpc_binding_handle
*h
,
1161 const struct GUID
*object
,
1164 const uint8_t *in_data
,
1167 struct irpc_bh_state
*hs
=
1168 dcerpc_binding_handle_data(h
,
1169 struct irpc_bh_state
);
1170 struct tevent_req
*req
;
1171 struct irpc_bh_raw_call_state
*state
;
1173 struct irpc_header header
;
1174 struct ndr_push
*ndr
;
1176 enum ndr_err_code ndr_err
;
1178 req
= tevent_req_create(mem_ctx
, &state
,
1179 struct irpc_bh_raw_call_state
);
1183 state
->opnum
= opnum
;
1184 state
->in_data
.data
= discard_const_p(uint8_t, in_data
);
1185 state
->in_data
.length
= in_length
;
1187 ok
= irpc_bh_is_connected(h
);
1189 tevent_req_nterror(req
, NT_STATUS_CONNECTION_DISCONNECTED
);
1190 return tevent_req_post(req
, ev
);
1193 state
->irpc
= talloc_zero(state
, struct irpc_request
);
1194 if (tevent_req_nomem(state
->irpc
, req
)) {
1195 return tevent_req_post(req
, ev
);
1198 state
->irpc
->msg_ctx
= hs
->msg_ctx
;
1199 state
->irpc
->callid
= idr_get_new(hs
->msg_ctx
->idr
,
1200 state
->irpc
, UINT16_MAX
);
1201 if (state
->irpc
->callid
== -1) {
1202 tevent_req_nterror(req
, NT_STATUS_INSUFFICIENT_RESOURCES
);
1203 return tevent_req_post(req
, ev
);
1205 state
->irpc
->incoming
.handler
= irpc_bh_raw_call_incoming_handler
;
1206 state
->irpc
->incoming
.private_data
= req
;
1208 talloc_set_destructor(state
->irpc
, irpc_destructor
);
1210 /* setup the header */
1211 header
.uuid
= hs
->table
->syntax_id
.uuid
;
1213 header
.if_version
= hs
->table
->syntax_id
.if_version
;
1214 header
.callid
= state
->irpc
->callid
;
1215 header
.callnum
= state
->opnum
;
1217 header
.status
= NT_STATUS_OK
;
1218 header
.creds
.token
= hs
->token
;
1220 /* construct the irpc packet */
1221 ndr
= ndr_push_init_ctx(state
->irpc
);
1222 if (tevent_req_nomem(ndr
, req
)) {
1223 return tevent_req_post(req
, ev
);
1226 ndr_err
= ndr_push_irpc_header(ndr
, NDR_SCALARS
|NDR_BUFFERS
, &header
);
1227 status
= ndr_map_error2ntstatus(ndr_err
);
1228 if (!NT_STATUS_IS_OK(status
)) {
1229 tevent_req_nterror(req
, status
);
1230 return tevent_req_post(req
, ev
);
1233 ndr_err
= ndr_push_bytes(ndr
, in_data
, in_length
);
1234 status
= ndr_map_error2ntstatus(ndr_err
);
1235 if (!NT_STATUS_IS_OK(status
)) {
1236 tevent_req_nterror(req
, status
);
1237 return tevent_req_post(req
, ev
);
1241 state
->in_packet
= ndr_push_blob(ndr
);
1242 status
= imessaging_send(hs
->msg_ctx
, hs
->server_id
,
1243 MSG_IRPC
, &state
->in_packet
);
1244 if (!NT_STATUS_IS_OK(status
)) {
1245 tevent_req_nterror(req
, status
);
1246 return tevent_req_post(req
, ev
);
1249 if (hs
->timeout
!= IRPC_CALL_TIMEOUT_INF
) {
1250 /* set timeout-callback in case caller wants that */
1251 ok
= tevent_req_set_endtime(req
, ev
, timeval_current_ofs(hs
->timeout
, 0));
1253 return tevent_req_post(req
, ev
);
1260 static void irpc_bh_raw_call_incoming_handler(struct irpc_request
*irpc
,
1261 struct irpc_message
*m
)
1263 struct tevent_req
*req
=
1264 talloc_get_type_abort(irpc
->incoming
.private_data
,
1266 struct irpc_bh_raw_call_state
*state
=
1267 tevent_req_data(req
,
1268 struct irpc_bh_raw_call_state
);
1270 talloc_steal(state
, m
);
1272 if (!NT_STATUS_IS_OK(m
->header
.status
)) {
1273 tevent_req_nterror(req
, m
->header
.status
);
1277 state
->out_data
= data_blob_talloc(state
,
1278 m
->ndr
->data
+ m
->ndr
->offset
,
1279 m
->ndr
->data_size
- m
->ndr
->offset
);
1280 if ((m
->ndr
->data_size
- m
->ndr
->offset
) > 0 && !state
->out_data
.data
) {
1281 tevent_req_oom(req
);
1285 tevent_req_done(req
);
1288 static NTSTATUS
irpc_bh_raw_call_recv(struct tevent_req
*req
,
1289 TALLOC_CTX
*mem_ctx
,
1292 uint32_t *out_flags
)
1294 struct irpc_bh_raw_call_state
*state
=
1295 tevent_req_data(req
,
1296 struct irpc_bh_raw_call_state
);
1299 if (tevent_req_is_nterror(req
, &status
)) {
1300 tevent_req_received(req
);
1304 *out_data
= talloc_move(mem_ctx
, &state
->out_data
.data
);
1305 *out_length
= state
->out_data
.length
;
1307 tevent_req_received(req
);
1308 return NT_STATUS_OK
;
1311 struct irpc_bh_disconnect_state
{
1315 static struct tevent_req
*irpc_bh_disconnect_send(TALLOC_CTX
*mem_ctx
,
1316 struct tevent_context
*ev
,
1317 struct dcerpc_binding_handle
*h
)
1319 struct irpc_bh_state
*hs
= dcerpc_binding_handle_data(h
,
1320 struct irpc_bh_state
);
1321 struct tevent_req
*req
;
1322 struct irpc_bh_disconnect_state
*state
;
1325 req
= tevent_req_create(mem_ctx
, &state
,
1326 struct irpc_bh_disconnect_state
);
1331 ok
= irpc_bh_is_connected(h
);
1333 tevent_req_nterror(req
, NT_STATUS_CONNECTION_DISCONNECTED
);
1334 return tevent_req_post(req
, ev
);
1339 tevent_req_done(req
);
1340 return tevent_req_post(req
, ev
);
1343 static NTSTATUS
irpc_bh_disconnect_recv(struct tevent_req
*req
)
1347 if (tevent_req_is_nterror(req
, &status
)) {
1348 tevent_req_received(req
);
1352 tevent_req_received(req
);
1353 return NT_STATUS_OK
;
1356 static bool irpc_bh_ref_alloc(struct dcerpc_binding_handle
*h
)
1361 static const struct dcerpc_binding_handle_ops irpc_bh_ops
= {
1363 .is_connected
= irpc_bh_is_connected
,
1364 .set_timeout
= irpc_bh_set_timeout
,
1365 .raw_call_send
= irpc_bh_raw_call_send
,
1366 .raw_call_recv
= irpc_bh_raw_call_recv
,
1367 .disconnect_send
= irpc_bh_disconnect_send
,
1368 .disconnect_recv
= irpc_bh_disconnect_recv
,
1370 .ref_alloc
= irpc_bh_ref_alloc
,
1373 /* initialise a irpc binding handle */
1374 struct dcerpc_binding_handle
*irpc_binding_handle(TALLOC_CTX
*mem_ctx
,
1375 struct imessaging_context
*msg_ctx
,
1376 struct server_id server_id
,
1377 const struct ndr_interface_table
*table
)
1379 struct dcerpc_binding_handle
*h
;
1380 struct irpc_bh_state
*hs
;
1382 h
= dcerpc_binding_handle_create(mem_ctx
,
1387 struct irpc_bh_state
,
1392 hs
->msg_ctx
= msg_ctx
;
1393 hs
->server_id
= server_id
;
1395 hs
->timeout
= IRPC_CALL_TIMEOUT
;
1397 dcerpc_binding_handle_set_sync_ev(h
, msg_ctx
->event
.ev
);
1402 struct dcerpc_binding_handle
*irpc_binding_handle_by_name(TALLOC_CTX
*mem_ctx
,
1403 struct imessaging_context
*msg_ctx
,
1404 const char *dest_task
,
1405 const struct ndr_interface_table
*table
)
1407 struct dcerpc_binding_handle
*h
;
1408 struct server_id
*sids
;
1409 struct server_id sid
;
1411 /* find the server task */
1412 sids
= irpc_servers_byname(msg_ctx
, mem_ctx
, dest_task
);
1414 errno
= EADDRNOTAVAIL
;
1417 if (sids
[0].pid
== 0) {
1419 errno
= EADDRNOTAVAIL
;
1425 h
= irpc_binding_handle(mem_ctx
, msg_ctx
,
1434 void irpc_binding_handle_add_security_token(struct dcerpc_binding_handle
*h
,
1435 struct security_token
*token
)
1437 struct irpc_bh_state
*hs
=
1438 dcerpc_binding_handle_data(h
,
1439 struct irpc_bh_state
);