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/util/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"
38 /* change the message version with any incompatible changes in the protocol */
39 #define IMESSAGING_VERSION 1
45 struct imessaging_context
*msg_ctx
;
48 void (*handler
)(struct irpc_request
*irpc
, struct irpc_message
*m
);
53 struct imessaging_context
{
54 struct server_id server_id
;
55 struct socket_context
*sock
;
56 const char *base_path
;
58 struct dispatch_fn
**dispatch
;
60 struct idr_context
*dispatch_tree
;
61 struct imessaging_rec
*pending
;
62 struct imessaging_rec
*retry_queue
;
63 struct irpc_list
*irpc
;
64 struct idr_context
*idr
;
66 struct timeval start_time
;
67 struct tevent_timer
*retry_te
;
69 struct tevent_context
*ev
;
70 struct tevent_fd
*fde
;
74 /* we have a linked list of dispatch handlers for each msg_type that
75 this messaging server can deal with */
77 struct dispatch_fn
*next
, *prev
;
83 /* an individual message */
84 struct imessaging_rec
{
85 struct imessaging_rec
*next
, *prev
;
86 struct imessaging_context
*msg
;
89 struct imessaging_header
{
92 struct server_id from
;
102 static void irpc_handler(struct imessaging_context
*, void *,
103 uint32_t, struct server_id
, DATA_BLOB
*);
107 A useful function for testing the message system.
109 static void ping_message(struct imessaging_context
*msg
, void *private_data
,
110 uint32_t msg_type
, struct server_id src
, DATA_BLOB
*data
)
112 char *task_id
= server_id_str(NULL
, &src
);
113 DEBUG(1,("INFO: Received PING message from server %s [%.*s]\n",
114 task_id
, (int)data
->length
,
115 data
->data
?(const char *)data
->data
:""));
116 talloc_free(task_id
);
117 imessaging_send(msg
, src
, MSG_PONG
, data
);
121 return uptime of messaging server via irpc
123 static NTSTATUS
irpc_uptime(struct irpc_message
*msg
,
124 struct irpc_uptime
*r
)
126 struct imessaging_context
*ctx
= talloc_get_type(msg
->private_data
, struct imessaging_context
);
127 *r
->out
.start_time
= timeval_to_nttime(&ctx
->start_time
);
132 return the path to a messaging socket
134 static char *imessaging_path(struct imessaging_context
*msg
, struct server_id server_id
)
136 TALLOC_CTX
*tmp_ctx
= talloc_new(msg
);
137 const char *id
= server_id_str(tmp_ctx
, &server_id
);
142 s
= talloc_asprintf(msg
, "%s/msg.%s", msg
->base_path
, id
);
143 talloc_steal(s
, tmp_ctx
);
148 dispatch a fully received message
150 note that this deliberately can match more than one message handler
151 per message. That allows a single messasging context to register
152 (for example) a debug handler for more than one piece of code
154 static void imessaging_dispatch(struct imessaging_context
*msg
, struct imessaging_rec
*rec
)
156 struct dispatch_fn
*d
, *next
;
158 /* temporary IDs use an idtree, the rest use a array of pointers */
159 if (rec
->header
->msg_type
>= MSG_TMP_BASE
) {
160 d
= (struct dispatch_fn
*)idr_find(msg
->dispatch_tree
,
161 rec
->header
->msg_type
);
162 } else if (rec
->header
->msg_type
< msg
->num_types
) {
163 d
= msg
->dispatch
[rec
->header
->msg_type
];
168 for (; d
; d
= next
) {
171 data
.data
= rec
->packet
.data
+ sizeof(*rec
->header
);
172 data
.length
= rec
->header
->length
;
173 d
->fn(msg
, d
->private_data
, d
->msg_type
, rec
->header
->from
, &data
);
175 rec
->header
->length
= 0;
179 handler for messages that arrive from other nodes in the cluster
181 static void cluster_message_handler(struct imessaging_context
*msg
, DATA_BLOB packet
)
183 struct imessaging_rec
*rec
;
185 rec
= talloc(msg
, struct imessaging_rec
);
187 smb_panic("Unable to allocate imessaging_rec");
191 rec
->path
= msg
->path
;
192 rec
->header
= (struct imessaging_header
*)packet
.data
;
193 rec
->packet
= packet
;
196 if (packet
.length
!= sizeof(*rec
->header
) + rec
->header
->length
) {
197 DEBUG(0,("messaging: bad message header size %d should be %d\n",
198 rec
->header
->length
, (int)(packet
.length
- sizeof(*rec
->header
))));
203 imessaging_dispatch(msg
, rec
);
210 try to send the message
212 static NTSTATUS
try_send(struct imessaging_rec
*rec
)
214 struct imessaging_context
*msg
= rec
->msg
;
218 struct socket_address
*path
;
220 /* rec->path is the path of the *other* socket, where we want
222 path
= socket_address_from_strings(msg
, msg
->sock
->backend_name
,
225 return NT_STATUS_NO_MEMORY
;
228 /* we send with privileges so messages work from any context */
229 priv
= root_privileges();
230 status
= socket_sendto(msg
->sock
, &rec
->packet
, &nsent
, path
);
238 retry backed off messages
240 static void msg_retry_timer(struct tevent_context
*ev
, struct tevent_timer
*te
,
241 struct timeval t
, void *private_data
)
243 struct imessaging_context
*msg
= talloc_get_type(private_data
,
244 struct imessaging_context
);
245 msg
->retry_te
= NULL
;
247 /* put the messages back on the main queue */
248 while (msg
->retry_queue
) {
249 struct imessaging_rec
*rec
= msg
->retry_queue
;
250 DLIST_REMOVE(msg
->retry_queue
, rec
);
251 DLIST_ADD_END(msg
->pending
, rec
, struct imessaging_rec
*);
254 EVENT_FD_WRITEABLE(msg
->event
.fde
);
258 handle a socket write event
260 static void imessaging_send_handler(struct imessaging_context
*msg
)
262 while (msg
->pending
) {
263 struct imessaging_rec
*rec
= msg
->pending
;
265 status
= try_send(rec
);
266 if (NT_STATUS_EQUAL(status
, STATUS_MORE_ENTRIES
)) {
268 if (rec
->retries
> 3) {
269 /* we're getting continuous write errors -
270 backoff this record */
271 DLIST_REMOVE(msg
->pending
, rec
);
272 DLIST_ADD_END(msg
->retry_queue
, rec
,
273 struct imessaging_rec
*);
274 if (msg
->retry_te
== NULL
) {
276 event_add_timed(msg
->event
.ev
, msg
,
277 timeval_current_ofs(1, 0),
278 msg_retry_timer
, msg
);
284 if (!NT_STATUS_IS_OK(status
)) {
285 TALLOC_CTX
*tmp_ctx
= talloc_new(msg
);
286 DEBUG(1,("messaging: Lost message from %s to %s of type %u - %s\n",
287 server_id_str(tmp_ctx
, &rec
->header
->from
),
288 server_id_str(tmp_ctx
, &rec
->header
->to
),
289 rec
->header
->msg_type
,
291 talloc_free(tmp_ctx
);
293 DLIST_REMOVE(msg
->pending
, rec
);
296 if (msg
->pending
== NULL
) {
297 EVENT_FD_NOT_WRITEABLE(msg
->event
.fde
);
302 handle a new incoming packet
304 static void imessaging_recv_handler(struct imessaging_context
*msg
)
306 struct imessaging_rec
*rec
;
311 /* see how many bytes are in the next packet */
312 status
= socket_pending(msg
->sock
, &msize
);
313 if (!NT_STATUS_IS_OK(status
)) {
314 DEBUG(0,("socket_pending failed in messaging - %s\n",
319 packet
= data_blob_talloc(msg
, NULL
, msize
);
320 if (packet
.data
== NULL
) {
321 /* assume this is temporary and retry */
325 status
= socket_recv(msg
->sock
, packet
.data
, msize
, &msize
);
326 if (!NT_STATUS_IS_OK(status
)) {
327 data_blob_free(&packet
);
331 if (msize
< sizeof(*rec
->header
)) {
332 DEBUG(0,("messaging: bad message of size %d\n", (int)msize
));
333 data_blob_free(&packet
);
337 rec
= talloc(msg
, struct imessaging_rec
);
339 smb_panic("Unable to allocate imessaging_rec");
342 talloc_steal(rec
, packet
.data
);
344 rec
->path
= msg
->path
;
345 rec
->header
= (struct imessaging_header
*)packet
.data
;
346 rec
->packet
= packet
;
349 if (msize
!= sizeof(*rec
->header
) + rec
->header
->length
) {
350 DEBUG(0,("messaging: bad message header size %d should be %d\n",
351 rec
->header
->length
, (int)(msize
- sizeof(*rec
->header
))));
356 imessaging_dispatch(msg
, rec
);
362 handle a socket event
364 static void imessaging_handler(struct tevent_context
*ev
, struct tevent_fd
*fde
,
365 uint16_t flags
, void *private_data
)
367 struct imessaging_context
*msg
= talloc_get_type(private_data
,
368 struct imessaging_context
);
369 if (flags
& EVENT_FD_WRITE
) {
370 imessaging_send_handler(msg
);
372 if (flags
& EVENT_FD_READ
) {
373 imessaging_recv_handler(msg
);
379 Register a dispatch function for a particular message type.
381 NTSTATUS
imessaging_register(struct imessaging_context
*msg
, void *private_data
,
382 uint32_t msg_type
, msg_callback_t fn
)
384 struct dispatch_fn
*d
;
386 /* possibly expand dispatch array */
387 if (msg_type
>= msg
->num_types
) {
388 struct dispatch_fn
**dp
;
390 dp
= talloc_realloc(msg
, msg
->dispatch
, struct dispatch_fn
*, msg_type
+1);
391 NT_STATUS_HAVE_NO_MEMORY(dp
);
393 for (i
=msg
->num_types
;i
<=msg_type
;i
++) {
394 msg
->dispatch
[i
] = NULL
;
396 msg
->num_types
= msg_type
+1;
399 d
= talloc_zero(msg
->dispatch
, struct dispatch_fn
);
400 NT_STATUS_HAVE_NO_MEMORY(d
);
401 d
->msg_type
= msg_type
;
402 d
->private_data
= private_data
;
405 DLIST_ADD(msg
->dispatch
[msg_type
], d
);
411 register a temporary message handler. The msg_type is allocated
414 NTSTATUS
imessaging_register_tmp(struct imessaging_context
*msg
, void *private_data
,
415 msg_callback_t fn
, uint32_t *msg_type
)
417 struct dispatch_fn
*d
;
420 d
= talloc_zero(msg
->dispatch
, struct dispatch_fn
);
421 NT_STATUS_HAVE_NO_MEMORY(d
);
422 d
->private_data
= private_data
;
425 id
= idr_get_new_above(msg
->dispatch_tree
, d
, MSG_TMP_BASE
, UINT16_MAX
);
428 return NT_STATUS_TOO_MANY_CONTEXT_IDS
;
431 d
->msg_type
= (uint32_t)id
;
432 (*msg_type
) = d
->msg_type
;
438 De-register the function for a particular message type.
440 void imessaging_deregister(struct imessaging_context
*msg
, uint32_t msg_type
, void *private_data
)
442 struct dispatch_fn
*d
, *next
;
444 if (msg_type
>= msg
->num_types
) {
445 d
= (struct dispatch_fn
*)idr_find(msg
->dispatch_tree
,
448 idr_remove(msg
->dispatch_tree
, msg_type
);
453 for (d
= msg
->dispatch
[msg_type
]; d
; d
= next
) {
455 if (d
->private_data
== private_data
) {
456 DLIST_REMOVE(msg
->dispatch
[msg_type
], d
);
463 Send a message to a particular server
465 NTSTATUS
imessaging_send(struct imessaging_context
*msg
, struct server_id server
,
466 uint32_t msg_type
, const DATA_BLOB
*data
)
468 struct imessaging_rec
*rec
;
470 size_t dlength
= data
?data
->length
:0;
472 rec
= talloc(msg
, struct imessaging_rec
);
474 return NT_STATUS_NO_MEMORY
;
477 rec
->packet
= data_blob_talloc(rec
, NULL
, sizeof(*rec
->header
) + dlength
);
478 if (rec
->packet
.data
== NULL
) {
480 return NT_STATUS_NO_MEMORY
;
485 rec
->header
= (struct imessaging_header
*)rec
->packet
.data
;
487 ZERO_STRUCTP(rec
->header
);
488 rec
->header
->version
= IMESSAGING_VERSION
;
489 rec
->header
->msg_type
= msg_type
;
490 rec
->header
->from
= msg
->server_id
;
491 rec
->header
->to
= server
;
492 rec
->header
->length
= dlength
;
494 memcpy(rec
->packet
.data
+ sizeof(*rec
->header
),
495 data
->data
, dlength
);
498 if (!cluster_node_equal(&msg
->server_id
, &server
)) {
499 /* the destination is on another node - dispatch via
501 status
= cluster_message_send(server
, &rec
->packet
);
506 rec
->path
= imessaging_path(msg
, server
);
507 talloc_steal(rec
, rec
->path
);
509 if (msg
->pending
!= NULL
) {
510 status
= STATUS_MORE_ENTRIES
;
512 status
= try_send(rec
);
515 if (NT_STATUS_EQUAL(status
, STATUS_MORE_ENTRIES
)) {
516 if (msg
->pending
== NULL
) {
517 EVENT_FD_WRITEABLE(msg
->event
.fde
);
519 DLIST_ADD_END(msg
->pending
, rec
, struct imessaging_rec
*);
529 Send a message to a particular server, with the message containing a single pointer
531 NTSTATUS
imessaging_send_ptr(struct imessaging_context
*msg
, struct server_id server
,
532 uint32_t msg_type
, void *ptr
)
536 blob
.data
= (uint8_t *)&ptr
;
537 blob
.length
= sizeof(void *);
539 return imessaging_send(msg
, server
, msg_type
, &blob
);
544 destroy the messaging context
546 static int imessaging_destructor(struct imessaging_context
*msg
)
549 while (msg
->names
&& msg
->names
[0]) {
550 irpc_remove_name(msg
, msg
->names
[0]);
556 create the listening socket and setup the dispatcher
558 struct imessaging_context
*imessaging_init(TALLOC_CTX
*mem_ctx
,
560 struct server_id server_id
,
561 struct tevent_context
*ev
)
563 struct imessaging_context
*msg
;
565 struct socket_address
*path
;
571 msg
= talloc_zero(mem_ctx
, struct imessaging_context
);
576 /* setup a handler for messages from other cluster nodes, if appropriate */
577 status
= cluster_message_init(msg
, server_id
, cluster_message_handler
);
578 if (!NT_STATUS_IS_OK(status
)) {
583 /* create the messaging directory if needed */
586 msg
->base_path
= talloc_reference(msg
, dir
);
587 msg
->path
= imessaging_path(msg
, server_id
);
588 msg
->server_id
= server_id
;
589 msg
->idr
= idr_init(msg
);
590 msg
->dispatch_tree
= idr_init(msg
);
591 msg
->start_time
= timeval_current();
593 status
= socket_create("unix", SOCKET_TYPE_DGRAM
, &msg
->sock
, 0);
594 if (!NT_STATUS_IS_OK(status
)) {
599 /* by stealing here we ensure that the socket is cleaned up (and even
601 talloc_steal(msg
, msg
->sock
);
603 path
= socket_address_from_strings(msg
, msg
->sock
->backend_name
,
610 status
= socket_listen(msg
->sock
, path
, 50, 0);
611 if (!NT_STATUS_IS_OK(status
)) {
612 DEBUG(0,("Unable to setup messaging listener for '%s':%s\n", msg
->path
, nt_errstr(status
)));
617 /* it needs to be non blocking for sends */
618 set_blocking(socket_get_fd(msg
->sock
), false);
621 msg
->event
.fde
= event_add_fd(ev
, msg
, socket_get_fd(msg
->sock
),
622 EVENT_FD_READ
, imessaging_handler
, msg
);
623 tevent_fd_set_auto_close(msg
->event
.fde
);
625 talloc_set_destructor(msg
, imessaging_destructor
);
627 imessaging_register(msg
, NULL
, MSG_PING
, ping_message
);
628 imessaging_register(msg
, NULL
, MSG_IRPC
, irpc_handler
);
629 IRPC_REGISTER(msg
, irpc
, IRPC_UPTIME
, irpc_uptime
, msg
);
635 A hack, for the short term until we get 'client only' messaging in place
637 struct imessaging_context
*imessaging_client_init(TALLOC_CTX
*mem_ctx
,
639 struct tevent_context
*ev
)
643 id
.pid
= random() % 0x10000000;
644 return imessaging_init(mem_ctx
, dir
, id
, ev
);
647 a list of registered irpc server functions
650 struct irpc_list
*next
, *prev
;
652 const struct ndr_interface_table
*table
;
660 register a irpc server function
662 NTSTATUS
irpc_register(struct imessaging_context
*msg_ctx
,
663 const struct ndr_interface_table
*table
,
664 int callnum
, irpc_function_t fn
, void *private_data
)
666 struct irpc_list
*irpc
;
668 /* override an existing handler, if any */
669 for (irpc
=msg_ctx
->irpc
; irpc
; irpc
=irpc
->next
) {
670 if (irpc
->table
== table
&& irpc
->callnum
== callnum
) {
675 irpc
= talloc(msg_ctx
, struct irpc_list
);
676 NT_STATUS_HAVE_NO_MEMORY(irpc
);
677 DLIST_ADD(msg_ctx
->irpc
, irpc
);
681 irpc
->callnum
= callnum
;
683 irpc
->private_data
= private_data
;
684 irpc
->uuid
= irpc
->table
->syntax_id
.uuid
;
691 handle an incoming irpc reply message
693 static void irpc_handler_reply(struct imessaging_context
*msg_ctx
, struct irpc_message
*m
)
695 struct irpc_request
*irpc
;
697 irpc
= (struct irpc_request
*)idr_find(msg_ctx
->idr
, m
->header
.callid
);
698 if (irpc
== NULL
) return;
700 irpc
->incoming
.handler(irpc
, m
);
706 NTSTATUS
irpc_send_reply(struct irpc_message
*m
, NTSTATUS status
)
708 struct ndr_push
*push
;
710 enum ndr_err_code ndr_err
;
712 m
->header
.status
= status
;
714 /* setup the reply */
715 push
= ndr_push_init_ctx(m
->ndr
);
717 status
= NT_STATUS_NO_MEMORY
;
721 m
->header
.flags
|= IRPC_FLAG_REPLY
;
722 m
->header
.creds
.token
= NULL
;
724 /* construct the packet */
725 ndr_err
= ndr_push_irpc_header(push
, NDR_SCALARS
|NDR_BUFFERS
, &m
->header
);
726 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err
)) {
727 status
= ndr_map_error2ntstatus(ndr_err
);
731 ndr_err
= m
->irpc
->table
->calls
[m
->irpc
->callnum
].ndr_push(push
, NDR_OUT
, m
->data
);
732 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err
)) {
733 status
= ndr_map_error2ntstatus(ndr_err
);
737 /* send the reply message */
738 packet
= ndr_push_blob(push
);
739 status
= imessaging_send(m
->msg_ctx
, m
->from
, MSG_IRPC
, &packet
);
740 if (!NT_STATUS_IS_OK(status
)) goto failed
;
748 handle an incoming irpc request message
750 static void irpc_handler_request(struct imessaging_context
*msg_ctx
,
751 struct irpc_message
*m
)
755 enum ndr_err_code ndr_err
;
757 for (i
=msg_ctx
->irpc
; i
; i
=i
->next
) {
758 if (GUID_equal(&i
->uuid
, &m
->header
.uuid
) &&
759 i
->table
->syntax_id
.if_version
== m
->header
.if_version
&&
760 i
->callnum
== m
->header
.callnum
) {
766 /* no registered handler for this message */
771 /* allocate space for the structure */
772 r
= talloc_zero_size(m
->ndr
, i
->table
->calls
[m
->header
.callnum
].struct_size
);
773 if (r
== NULL
) goto failed
;
775 m
->ndr
->flags
|= LIBNDR_FLAG_REF_ALLOC
;
777 /* parse the request data */
778 ndr_err
= i
->table
->calls
[i
->callnum
].ndr_pull(m
->ndr
, NDR_IN
, r
);
779 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err
)) goto failed
;
782 m
->private_data
= i
->private_data
;
783 m
->defer_reply
= false;
785 m
->msg_ctx
= msg_ctx
;
788 m
->ev
= msg_ctx
->event
.ev
;
790 m
->header
.status
= i
->fn(m
, r
);
793 /* the server function won't ever be replying to this request */
798 if (m
->defer_reply
) {
799 /* the server function has asked to defer the reply to later */
800 talloc_steal(msg_ctx
, m
);
804 irpc_send_reply(m
, m
->header
.status
);
812 handle an incoming irpc message
814 static void irpc_handler(struct imessaging_context
*msg_ctx
, void *private_data
,
815 uint32_t msg_type
, struct server_id src
, DATA_BLOB
*packet
)
817 struct irpc_message
*m
;
818 enum ndr_err_code ndr_err
;
820 m
= talloc(msg_ctx
, struct irpc_message
);
821 if (m
== NULL
) goto failed
;
825 m
->ndr
= ndr_pull_init_blob(packet
, m
);
826 if (m
->ndr
== NULL
) goto failed
;
828 m
->ndr
->flags
|= LIBNDR_FLAG_REF_ALLOC
;
830 ndr_err
= ndr_pull_irpc_header(m
->ndr
, NDR_BUFFERS
|NDR_SCALARS
, &m
->header
);
831 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err
)) goto failed
;
833 if (m
->header
.flags
& IRPC_FLAG_REPLY
) {
834 irpc_handler_reply(msg_ctx
, m
);
836 irpc_handler_request(msg_ctx
, m
);
846 destroy a irpc request
848 static int irpc_destructor(struct irpc_request
*irpc
)
850 if (irpc
->callid
!= -1) {
851 idr_remove(irpc
->msg_ctx
->idr
, irpc
->callid
);
859 open the naming database
861 static struct tdb_wrap
*irpc_namedb_open(struct imessaging_context
*msg_ctx
)
864 char *path
= talloc_asprintf(msg_ctx
, "%s/names.tdb", msg_ctx
->base_path
);
868 t
= tdb_wrap_open(msg_ctx
, path
, 0, 0, O_RDWR
|O_CREAT
, 0660);
875 add a string name that this irpc server can be called on
877 NTSTATUS
irpc_add_name(struct imessaging_context
*msg_ctx
, const char *name
)
882 NTSTATUS status
= NT_STATUS_OK
;
884 t
= irpc_namedb_open(msg_ctx
);
885 NT_STATUS_HAVE_NO_MEMORY(t
);
887 if (tdb_lock_bystring(t
->tdb
, name
) != 0) {
889 return NT_STATUS_LOCK_NOT_GRANTED
;
891 rec
= tdb_fetch_bystring(t
->tdb
, name
);
892 count
= rec
.dsize
/ sizeof(struct server_id
);
893 rec
.dptr
= (unsigned char *)realloc_p(rec
.dptr
, struct server_id
, count
+1);
894 rec
.dsize
+= sizeof(struct server_id
);
895 if (rec
.dptr
== NULL
) {
896 tdb_unlock_bystring(t
->tdb
, name
);
898 return NT_STATUS_NO_MEMORY
;
900 ((struct server_id
*)rec
.dptr
)[count
] = msg_ctx
->server_id
;
901 if (tdb_store_bystring(t
->tdb
, name
, rec
, 0) != 0) {
902 status
= NT_STATUS_INTERNAL_ERROR
;
905 tdb_unlock_bystring(t
->tdb
, name
);
908 msg_ctx
->names
= str_list_add(msg_ctx
->names
, name
);
909 talloc_steal(msg_ctx
, msg_ctx
->names
);
915 return a list of server ids for a server name
917 struct server_id
*irpc_servers_byname(struct imessaging_context
*msg_ctx
,
924 struct server_id
*ret
;
926 t
= irpc_namedb_open(msg_ctx
);
931 if (tdb_lock_bystring(t
->tdb
, name
) != 0) {
935 rec
= tdb_fetch_bystring(t
->tdb
, name
);
936 if (rec
.dptr
== NULL
) {
937 tdb_unlock_bystring(t
->tdb
, name
);
941 count
= rec
.dsize
/ sizeof(struct server_id
);
942 ret
= talloc_array(mem_ctx
, struct server_id
, count
+1);
944 tdb_unlock_bystring(t
->tdb
, name
);
948 for (i
=0;i
<count
;i
++) {
949 ret
[i
] = ((struct server_id
*)rec
.dptr
)[i
];
951 ret
[i
] = cluster_id(0, 0);
953 tdb_unlock_bystring(t
->tdb
, name
);
960 remove a name from a messaging context
962 void irpc_remove_name(struct imessaging_context
*msg_ctx
, const char *name
)
967 struct server_id
*ids
;
969 str_list_remove(msg_ctx
->names
, name
);
971 t
= irpc_namedb_open(msg_ctx
);
976 if (tdb_lock_bystring(t
->tdb
, name
) != 0) {
980 rec
= tdb_fetch_bystring(t
->tdb
, name
);
981 if (rec
.dptr
== NULL
) {
982 tdb_unlock_bystring(t
->tdb
, name
);
986 count
= rec
.dsize
/ sizeof(struct server_id
);
989 tdb_unlock_bystring(t
->tdb
, name
);
993 ids
= (struct server_id
*)rec
.dptr
;
994 for (i
=0;i
<count
;i
++) {
995 if (cluster_id_equal(&ids
[i
], &msg_ctx
->server_id
)) {
997 memmove(ids
+i
, ids
+i
+1,
998 sizeof(struct server_id
) * (count
-(i
+1)));
1000 rec
.dsize
-= sizeof(struct server_id
);
1004 tdb_store_bystring(t
->tdb
, name
, rec
, 0);
1006 tdb_unlock_bystring(t
->tdb
, name
);
1010 struct server_id
imessaging_get_server_id(struct imessaging_context
*msg_ctx
)
1012 return msg_ctx
->server_id
;
1015 struct irpc_bh_state
{
1016 struct imessaging_context
*msg_ctx
;
1017 struct server_id server_id
;
1018 const struct ndr_interface_table
*table
;
1020 struct security_token
*token
;
1023 static bool irpc_bh_is_connected(struct dcerpc_binding_handle
*h
)
1025 struct irpc_bh_state
*hs
= dcerpc_binding_handle_data(h
,
1026 struct irpc_bh_state
);
1035 static uint32_t irpc_bh_set_timeout(struct dcerpc_binding_handle
*h
,
1038 struct irpc_bh_state
*hs
= dcerpc_binding_handle_data(h
,
1039 struct irpc_bh_state
);
1040 uint32_t old
= hs
->timeout
;
1042 hs
->timeout
= timeout
;
1047 struct irpc_bh_raw_call_state
{
1048 struct irpc_request
*irpc
;
1051 DATA_BLOB in_packet
;
1055 static void irpc_bh_raw_call_incoming_handler(struct irpc_request
*irpc
,
1056 struct irpc_message
*m
);
1058 static struct tevent_req
*irpc_bh_raw_call_send(TALLOC_CTX
*mem_ctx
,
1059 struct tevent_context
*ev
,
1060 struct dcerpc_binding_handle
*h
,
1061 const struct GUID
*object
,
1064 const uint8_t *in_data
,
1067 struct irpc_bh_state
*hs
=
1068 dcerpc_binding_handle_data(h
,
1069 struct irpc_bh_state
);
1070 struct tevent_req
*req
;
1071 struct irpc_bh_raw_call_state
*state
;
1073 struct irpc_header header
;
1074 struct ndr_push
*ndr
;
1076 enum ndr_err_code ndr_err
;
1078 req
= tevent_req_create(mem_ctx
, &state
,
1079 struct irpc_bh_raw_call_state
);
1083 state
->opnum
= opnum
;
1084 state
->in_data
.data
= discard_const_p(uint8_t, in_data
);
1085 state
->in_data
.length
= in_length
;
1087 ok
= irpc_bh_is_connected(h
);
1089 tevent_req_nterror(req
, NT_STATUS_INVALID_CONNECTION
);
1090 return tevent_req_post(req
, ev
);
1093 state
->irpc
= talloc_zero(state
, struct irpc_request
);
1094 if (tevent_req_nomem(state
->irpc
, req
)) {
1095 return tevent_req_post(req
, ev
);
1098 state
->irpc
->msg_ctx
= hs
->msg_ctx
;
1099 state
->irpc
->callid
= idr_get_new(hs
->msg_ctx
->idr
,
1100 state
->irpc
, UINT16_MAX
);
1101 if (state
->irpc
->callid
== -1) {
1102 tevent_req_nterror(req
, NT_STATUS_INSUFFICIENT_RESOURCES
);
1103 return tevent_req_post(req
, ev
);
1105 state
->irpc
->incoming
.handler
= irpc_bh_raw_call_incoming_handler
;
1106 state
->irpc
->incoming
.private_data
= req
;
1108 talloc_set_destructor(state
->irpc
, irpc_destructor
);
1110 /* setup the header */
1111 header
.uuid
= hs
->table
->syntax_id
.uuid
;
1113 header
.if_version
= hs
->table
->syntax_id
.if_version
;
1114 header
.callid
= state
->irpc
->callid
;
1115 header
.callnum
= state
->opnum
;
1117 header
.status
= NT_STATUS_OK
;
1118 header
.creds
.token
= hs
->token
;
1120 /* construct the irpc packet */
1121 ndr
= ndr_push_init_ctx(state
->irpc
);
1122 if (tevent_req_nomem(ndr
, req
)) {
1123 return tevent_req_post(req
, ev
);
1126 ndr_err
= ndr_push_irpc_header(ndr
, NDR_SCALARS
|NDR_BUFFERS
, &header
);
1127 status
= ndr_map_error2ntstatus(ndr_err
);
1128 if (!NT_STATUS_IS_OK(status
)) {
1129 tevent_req_nterror(req
, status
);
1130 return tevent_req_post(req
, ev
);
1133 ndr_err
= ndr_push_bytes(ndr
, in_data
, in_length
);
1134 status
= ndr_map_error2ntstatus(ndr_err
);
1135 if (!NT_STATUS_IS_OK(status
)) {
1136 tevent_req_nterror(req
, status
);
1137 return tevent_req_post(req
, ev
);
1141 state
->in_packet
= ndr_push_blob(ndr
);
1142 status
= imessaging_send(hs
->msg_ctx
, hs
->server_id
,
1143 MSG_IRPC
, &state
->in_packet
);
1144 if (!NT_STATUS_IS_OK(status
)) {
1145 tevent_req_nterror(req
, status
);
1146 return tevent_req_post(req
, ev
);
1149 if (hs
->timeout
!= IRPC_CALL_TIMEOUT_INF
) {
1150 /* set timeout-callback in case caller wants that */
1151 ok
= tevent_req_set_endtime(req
, ev
, timeval_current_ofs(hs
->timeout
, 0));
1153 return tevent_req_post(req
, ev
);
1160 static void irpc_bh_raw_call_incoming_handler(struct irpc_request
*irpc
,
1161 struct irpc_message
*m
)
1163 struct tevent_req
*req
=
1164 talloc_get_type_abort(irpc
->incoming
.private_data
,
1166 struct irpc_bh_raw_call_state
*state
=
1167 tevent_req_data(req
,
1168 struct irpc_bh_raw_call_state
);
1170 talloc_steal(state
, m
);
1172 if (!NT_STATUS_IS_OK(m
->header
.status
)) {
1173 tevent_req_nterror(req
, m
->header
.status
);
1177 state
->out_data
= data_blob_talloc(state
,
1178 m
->ndr
->data
+ m
->ndr
->offset
,
1179 m
->ndr
->data_size
- m
->ndr
->offset
);
1180 if ((m
->ndr
->data_size
- m
->ndr
->offset
) > 0 && !state
->out_data
.data
) {
1181 tevent_req_nomem(NULL
, req
);
1185 tevent_req_done(req
);
1188 static NTSTATUS
irpc_bh_raw_call_recv(struct tevent_req
*req
,
1189 TALLOC_CTX
*mem_ctx
,
1192 uint32_t *out_flags
)
1194 struct irpc_bh_raw_call_state
*state
=
1195 tevent_req_data(req
,
1196 struct irpc_bh_raw_call_state
);
1199 if (tevent_req_is_nterror(req
, &status
)) {
1200 tevent_req_received(req
);
1204 *out_data
= talloc_move(mem_ctx
, &state
->out_data
.data
);
1205 *out_length
= state
->out_data
.length
;
1207 tevent_req_received(req
);
1208 return NT_STATUS_OK
;
1211 struct irpc_bh_disconnect_state
{
1215 static struct tevent_req
*irpc_bh_disconnect_send(TALLOC_CTX
*mem_ctx
,
1216 struct tevent_context
*ev
,
1217 struct dcerpc_binding_handle
*h
)
1219 struct irpc_bh_state
*hs
= dcerpc_binding_handle_data(h
,
1220 struct irpc_bh_state
);
1221 struct tevent_req
*req
;
1222 struct irpc_bh_disconnect_state
*state
;
1225 req
= tevent_req_create(mem_ctx
, &state
,
1226 struct irpc_bh_disconnect_state
);
1231 ok
= irpc_bh_is_connected(h
);
1233 tevent_req_nterror(req
, NT_STATUS_INVALID_CONNECTION
);
1234 return tevent_req_post(req
, ev
);
1239 tevent_req_done(req
);
1240 return tevent_req_post(req
, ev
);
1243 static NTSTATUS
irpc_bh_disconnect_recv(struct tevent_req
*req
)
1247 if (tevent_req_is_nterror(req
, &status
)) {
1248 tevent_req_received(req
);
1252 tevent_req_received(req
);
1253 return NT_STATUS_OK
;
1256 static bool irpc_bh_ref_alloc(struct dcerpc_binding_handle
*h
)
1261 static const struct dcerpc_binding_handle_ops irpc_bh_ops
= {
1263 .is_connected
= irpc_bh_is_connected
,
1264 .set_timeout
= irpc_bh_set_timeout
,
1265 .raw_call_send
= irpc_bh_raw_call_send
,
1266 .raw_call_recv
= irpc_bh_raw_call_recv
,
1267 .disconnect_send
= irpc_bh_disconnect_send
,
1268 .disconnect_recv
= irpc_bh_disconnect_recv
,
1270 .ref_alloc
= irpc_bh_ref_alloc
,
1273 /* initialise a irpc binding handle */
1274 struct dcerpc_binding_handle
*irpc_binding_handle(TALLOC_CTX
*mem_ctx
,
1275 struct imessaging_context
*msg_ctx
,
1276 struct server_id server_id
,
1277 const struct ndr_interface_table
*table
)
1279 struct dcerpc_binding_handle
*h
;
1280 struct irpc_bh_state
*hs
;
1282 h
= dcerpc_binding_handle_create(mem_ctx
,
1287 struct irpc_bh_state
,
1292 hs
->msg_ctx
= msg_ctx
;
1293 hs
->server_id
= server_id
;
1295 hs
->timeout
= IRPC_CALL_TIMEOUT
;
1297 dcerpc_binding_handle_set_sync_ev(h
, msg_ctx
->event
.ev
);
1302 struct dcerpc_binding_handle
*irpc_binding_handle_by_name(TALLOC_CTX
*mem_ctx
,
1303 struct imessaging_context
*msg_ctx
,
1304 const char *dest_task
,
1305 const struct ndr_interface_table
*table
)
1307 struct dcerpc_binding_handle
*h
;
1308 struct server_id
*sids
;
1309 struct server_id sid
;
1311 /* find the server task */
1312 sids
= irpc_servers_byname(msg_ctx
, mem_ctx
, dest_task
);
1314 errno
= EADDRNOTAVAIL
;
1317 if (sids
[0].pid
== 0) {
1319 errno
= EADDRNOTAVAIL
;
1325 h
= irpc_binding_handle(mem_ctx
, msg_ctx
,
1334 void irpc_binding_handle_add_security_token(struct dcerpc_binding_handle
*h
,
1335 struct security_token
*token
)
1337 struct irpc_bh_state
*hs
=
1338 dcerpc_binding_handle_data(h
,
1339 struct irpc_bh_state
);