diff options
Diffstat (limited to 'source4/lib/messaging')
| -rw-r--r-- | source4/lib/messaging/irpc.h | 86 | ||||
| -rw-r--r-- | source4/lib/messaging/messaging.c | 1485 | ||||
| -rw-r--r-- | source4/lib/messaging/messaging.h | 70 | ||||
| -rw-r--r-- | source4/lib/messaging/messaging_handlers.c | 135 | ||||
| -rw-r--r-- | source4/lib/messaging/messaging_internal.h | 50 | ||||
| -rw-r--r-- | source4/lib/messaging/messaging_send.c | 115 | ||||
| -rw-r--r-- | source4/lib/messaging/pymessaging.c | 551 | ||||
| -rw-r--r-- | source4/lib/messaging/tests/irpc.c | 308 | ||||
| -rw-r--r-- | source4/lib/messaging/tests/messaging.c | 694 | ||||
| -rw-r--r-- | source4/lib/messaging/wscript_build | 33 |
10 files changed, 3527 insertions, 0 deletions
diff --git a/source4/lib/messaging/irpc.h b/source4/lib/messaging/irpc.h new file mode 100644 index 0000000..ebf30af --- /dev/null +++ b/source4/lib/messaging/irpc.h @@ -0,0 +1,86 @@ +/* + Unix SMB/CIFS implementation. + + Samba internal rpc code - header + + Copyright (C) Andrew Tridgell 2005 + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +#ifndef IRPC_H +#define IRPC_H + +#include "lib/messaging/messaging.h" +#include "librpc/gen_ndr/irpc.h" + +/* + an incoming irpc message +*/ +struct irpc_message { + struct server_id from; + void *private_data; + struct irpc_header header; + struct ndr_pull *ndr; + bool defer_reply; + bool no_reply; + struct imessaging_context *msg_ctx; + struct irpc_list *irpc; + void *data; +}; + +/* don't allow calls to take too long */ +#define IRPC_CALL_TIMEOUT 10 +/* wait for the calls as long as it takes */ +#define IRPC_CALL_TIMEOUT_INF 0 + + +/* the server function type */ +typedef NTSTATUS (*irpc_function_t)(struct irpc_message *, void *r); + +/* register a server function with the irpc messaging system */ +#define IRPC_REGISTER(msg_ctx, pipename, funcname, function, private_data) \ + irpc_register(msg_ctx, &ndr_table_ ## pipename, \ + NDR_ ## funcname, \ + (irpc_function_t)function, private_data) + +struct ndr_interface_table; + +NTSTATUS irpc_register(struct imessaging_context *msg_ctx, + const struct ndr_interface_table *table, + int call, irpc_function_t fn, void *private_data); + +struct dcerpc_binding_handle *irpc_binding_handle(TALLOC_CTX *mem_ctx, + struct imessaging_context *msg_ctx, + struct server_id server_id, + const struct ndr_interface_table *table); +struct dcerpc_binding_handle *irpc_binding_handle_by_name(TALLOC_CTX *mem_ctx, + struct imessaging_context *msg_ctx, + const char *dest_task, + const struct ndr_interface_table *table); +void irpc_binding_handle_add_security_token(struct dcerpc_binding_handle *h, + struct security_token *token); + +NTSTATUS irpc_add_name(struct imessaging_context *msg_ctx, const char *name); +NTSTATUS irpc_servers_byname(struct imessaging_context *msg_ctx, + TALLOC_CTX *mem_ctx, const char *name, + unsigned *num_servers, + struct server_id **servers); +struct irpc_name_records *irpc_all_servers(struct imessaging_context *msg_ctx, + TALLOC_CTX *mem_ctx); +void irpc_remove_name(struct imessaging_context *msg_ctx, const char *name); +NTSTATUS irpc_send_reply(struct irpc_message *m, NTSTATUS status); + +#endif + diff --git a/source4/lib/messaging/messaging.c b/source4/lib/messaging/messaging.c new file mode 100644 index 0000000..67b0366 --- /dev/null +++ b/source4/lib/messaging/messaging.c @@ -0,0 +1,1485 @@ +/* + Unix SMB/CIFS implementation. + + Samba internal messaging functions + + Copyright (C) Andrew Tridgell 2004 + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +#include "includes.h" +#include "lib/events/events.h" +#include "lib/util/server_id.h" +#include "system/filesys.h" +#include "messaging/messaging.h" +#include "messaging/messaging_internal.h" +#include "../lib/util/dlinklist.h" +#include "lib/socket/socket.h" +#include "librpc/gen_ndr/ndr_irpc.h" +#include "lib/messaging/irpc.h" +#include "../lib/util/unix_privs.h" +#include "librpc/rpc/dcerpc.h" +#include "cluster/cluster.h" +#include "../lib/util/tevent_ntstatus.h" +#include "lib/param/param.h" +#include "lib/util/server_id_db.h" +#include "lib/util/talloc_report_printf.h" +#include "lib/messaging/messages_dgm.h" +#include "lib/messaging/messages_dgm_ref.h" +#include "../source3/lib/messages_util.h" +#include <tdb.h> + +/* change the message version with any incompatible changes in the protocol */ +#define IMESSAGING_VERSION 1 + +/* + a pending irpc call +*/ +struct irpc_request { + struct irpc_request *prev, *next; + struct imessaging_context *msg_ctx; + int callid; + struct { + void (*handler)(struct irpc_request *irpc, struct irpc_message *m); + void *private_data; + } incoming; +}; + +/* we have a linked list of dispatch handlers for each msg_type that + this messaging server can deal with */ +struct dispatch_fn { + struct dispatch_fn *next, *prev; + uint32_t msg_type; + void *private_data; + msg_callback_t fn; +}; + +/* an individual message */ + +static void irpc_handler(struct imessaging_context *, + void *, + uint32_t, + struct server_id, + size_t, + int *, + DATA_BLOB *); + + +/* + A useful function for testing the message system. +*/ +static void ping_message(struct imessaging_context *msg, + void *private_data, + uint32_t msg_type, + struct server_id src, + size_t num_fds, + int *fds, + DATA_BLOB *data) +{ + struct server_id_buf idbuf; + + if (num_fds != 0) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + DEBUG(1,("INFO: Received PING message from server %s [%.*s]\n", + server_id_str_buf(src, &idbuf), (int)data->length, + data->data?(const char *)data->data:"")); + imessaging_send(msg, src, MSG_PONG, data); +} + +static void pool_message(struct imessaging_context *msg, + void *private_data, + uint32_t msg_type, + struct server_id src, + size_t num_fds, + int *fds, + DATA_BLOB *data) +{ + FILE *f = NULL; + + if (num_fds != 1) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + f = fdopen(fds[0], "w"); + if (f == NULL) { + DBG_DEBUG("fopen failed: %s\n", strerror(errno)); + return; + } + + talloc_full_report_printf(NULL, f); + fclose(f); +} + +static void ringbuf_log_msg(struct imessaging_context *msg, + void *private_data, + uint32_t msg_type, + struct server_id src, + size_t num_fds, + int *fds, + DATA_BLOB *data) +{ + char *log = debug_get_ringbuf(); + size_t logsize = debug_get_ringbuf_size(); + DATA_BLOB blob; + + if (num_fds != 0) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + if (log == NULL) { + log = discard_const_p(char, "*disabled*\n"); + logsize = strlen(log) + 1; + } + + blob.data = (uint8_t *)log; + blob.length = logsize; + + imessaging_send(msg, src, MSG_RINGBUF_LOG, &blob); +} + +/**************************************************************************** + Receive a "set debug level" message. +****************************************************************************/ + +static void debug_imessage(struct imessaging_context *msg_ctx, + void *private_data, + uint32_t msg_type, + struct server_id src, + size_t num_fds, + int *fds, + DATA_BLOB *data) +{ + const char *params_str = (const char *)data->data; + struct server_id_buf src_buf; + struct server_id dst = imessaging_get_server_id(msg_ctx); + struct server_id_buf dst_buf; + + if (num_fds != 0) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + /* Check, it's a proper string! */ + if (params_str[(data->length)-1] != '\0') { + DBG_ERR("Invalid debug message from pid %s to pid %s\n", + server_id_str_buf(src, &src_buf), + server_id_str_buf(dst, &dst_buf)); + return; + } + + DBG_ERR("INFO: Remote set of debug to `%s' (pid %s from pid %s)\n", + params_str, + server_id_str_buf(dst, &dst_buf), + server_id_str_buf(src, &src_buf)); + + debug_parse_levels(params_str); +} + +/**************************************************************************** + Return current debug level. +****************************************************************************/ + +static void debuglevel_imessage(struct imessaging_context *msg_ctx, + void *private_data, + uint32_t msg_type, + struct server_id src, + size_t num_fds, + int *fds, + DATA_BLOB *data) +{ + char *message = debug_list_class_names_and_levels(); + DATA_BLOB blob = data_blob_null; + struct server_id_buf src_buf; + struct server_id dst = imessaging_get_server_id(msg_ctx); + struct server_id_buf dst_buf; + + if (num_fds != 0) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + DBG_DEBUG("Received REQ_DEBUGLEVEL message (pid %s from pid %s)\n", + server_id_str_buf(dst, &dst_buf), + server_id_str_buf(src, &src_buf)); + + if (message == NULL) { + DBG_ERR("debug_list_class_names_and_levels returned NULL\n"); + return; + } + + blob = data_blob_string_const_null(message); + imessaging_send(msg_ctx, src, MSG_DEBUGLEVEL, &blob); + + TALLOC_FREE(message); +} + +/* + return uptime of messaging server via irpc +*/ +static NTSTATUS irpc_uptime(struct irpc_message *msg, + struct irpc_uptime *r) +{ + struct imessaging_context *ctx = talloc_get_type(msg->private_data, struct imessaging_context); + *r->out.start_time = timeval_to_nttime(&ctx->start_time); + return NT_STATUS_OK; +} + +static struct dispatch_fn *imessaging_find_dispatch( + struct imessaging_context *msg, uint32_t msg_type) +{ + /* temporary IDs use an idtree, the rest use a array of pointers */ + if (msg_type >= MSG_TMP_BASE) { + return (struct dispatch_fn *)idr_find(msg->dispatch_tree, + msg_type); + } + if (msg_type < msg->num_types) { + return msg->dispatch[msg_type]; + } + return NULL; +} + +/* + Register a dispatch function for a particular message type. +*/ +NTSTATUS imessaging_register(struct imessaging_context *msg, void *private_data, + uint32_t msg_type, msg_callback_t fn) +{ + struct dispatch_fn *d; + + /* possibly expand dispatch array */ + if (msg_type >= msg->num_types) { + struct dispatch_fn **dp; + uint32_t i; + dp = talloc_realloc(msg, msg->dispatch, struct dispatch_fn *, msg_type+1); + NT_STATUS_HAVE_NO_MEMORY(dp); + msg->dispatch = dp; + for (i=msg->num_types;i<=msg_type;i++) { + msg->dispatch[i] = NULL; + } + msg->num_types = msg_type+1; + } + + d = talloc_zero(msg->dispatch, struct dispatch_fn); + NT_STATUS_HAVE_NO_MEMORY(d); + d->msg_type = msg_type; + d->private_data = private_data; + d->fn = fn; + + DLIST_ADD(msg->dispatch[msg_type], d); + + return NT_STATUS_OK; +} + +/* + register a temporary message handler. The msg_type is allocated + above MSG_TMP_BASE +*/ +NTSTATUS imessaging_register_tmp(struct imessaging_context *msg, void *private_data, + msg_callback_t fn, uint32_t *msg_type) +{ + struct dispatch_fn *d; + int id; + + d = talloc_zero(msg->dispatch, struct dispatch_fn); + NT_STATUS_HAVE_NO_MEMORY(d); + d->private_data = private_data; + d->fn = fn; + + id = idr_get_new_above(msg->dispatch_tree, d, MSG_TMP_BASE, UINT16_MAX); + if (id == -1) { + talloc_free(d); + return NT_STATUS_TOO_MANY_CONTEXT_IDS; + } + + d->msg_type = (uint32_t)id; + (*msg_type) = d->msg_type; + + return NT_STATUS_OK; +} + +/* + De-register the function for a particular message type. +*/ +void imessaging_deregister(struct imessaging_context *msg, uint32_t msg_type, void *private_data) +{ + struct dispatch_fn *d, *next; + + if (msg_type >= msg->num_types) { + d = (struct dispatch_fn *)idr_find(msg->dispatch_tree, + msg_type); + if (!d) return; + idr_remove(msg->dispatch_tree, msg_type); + talloc_free(d); + return; + } + + for (d = msg->dispatch[msg_type]; d; d = next) { + next = d->next; + if (d->private_data == private_data) { + DLIST_REMOVE(msg->dispatch[msg_type], d); + talloc_free(d); + } + } +} + +/* +*/ +int imessaging_cleanup(struct imessaging_context *msg) +{ + if (!msg) { + return 0; + } + return 0; +} + +static void imessaging_dgm_recv(struct tevent_context *ev, + const uint8_t *buf, size_t buf_len, + int *fds, size_t num_fds, + void *private_data); + +/* Keep a list of imessaging contexts */ +static struct imessaging_context *msg_ctxs; + +/* + * A process has terminated, clean-up any names it has registered. + */ +NTSTATUS imessaging_process_cleanup( + struct imessaging_context *msg_ctx, + pid_t pid) +{ + struct irpc_name_records *names = NULL; + uint32_t i = 0; + uint32_t j = 0; + TALLOC_CTX *mem_ctx = talloc_new(NULL); + + if (mem_ctx == NULL) { + DBG_ERR("OOM unable to clean up messaging for process (%d)\n", + pid); + return NT_STATUS_NO_MEMORY; + } + + names = irpc_all_servers(msg_ctx, mem_ctx); + if (names == NULL) { + TALLOC_FREE(mem_ctx); + return NT_STATUS_OK; + } + for (i = 0; i < names->num_records; i++) { + for (j = 0; j < names->names[i]->count; j++) { + if (names->names[i]->ids[j].pid == pid) { + int ret = server_id_db_prune_name( + msg_ctx->names, + names->names[i]->name, + names->names[i]->ids[j]); + if (ret != 0 && ret != ENOENT) { + TALLOC_FREE(mem_ctx); + return map_nt_error_from_unix_common( + ret); + } + } + } + } + TALLOC_FREE(mem_ctx); + return NT_STATUS_OK; +} + +static int imessaging_context_destructor(struct imessaging_context *msg) +{ + struct irpc_request *irpc = NULL; + struct irpc_request *next = NULL; + + for (irpc = msg->requests; irpc != NULL; irpc = next) { + next = irpc->next; + + DLIST_REMOVE(msg->requests, irpc); + irpc->callid = -1; + } + + DLIST_REMOVE(msg_ctxs, msg); + TALLOC_FREE(msg->msg_dgm_ref); + return 0; +} + +/* + * Cleanup messaging dgm contexts on a specific event context. + * + * We must make sure to unref all messaging_dgm_ref's *before* the + * tevent context goes away. Only when the last ref is freed, the + * refcounted messaging dgm context will be freed. + */ +void imessaging_dgm_unref_ev(struct tevent_context *ev) +{ + struct imessaging_context *msg = NULL; + + for (msg = msg_ctxs; msg != NULL; msg = msg->next) { + if (msg->ev == ev) { + TALLOC_FREE(msg->msg_dgm_ref); + } + } +} + +static NTSTATUS imessaging_reinit(struct imessaging_context *msg) +{ + int ret = -1; + + TALLOC_FREE(msg->msg_dgm_ref); + + if (msg->discard_incoming) { + msg->num_incoming_listeners = 0; + } else { + msg->num_incoming_listeners = 1; + } + + msg->server_id.pid = getpid(); + + msg->msg_dgm_ref = messaging_dgm_ref(msg, + msg->ev, + &msg->server_id.unique_id, + msg->sock_dir, + msg->lock_dir, + imessaging_dgm_recv, + msg, + &ret); + + if (msg->msg_dgm_ref == NULL) { + DEBUG(2, ("messaging_dgm_ref failed: %s\n", + strerror(ret))); + return map_nt_error_from_unix_common(ret); + } + + server_id_db_reinit(msg->names, msg->server_id); + return NT_STATUS_OK; +} + +/* + * Must be called after a fork. + */ +NTSTATUS imessaging_reinit_all(void) +{ + struct imessaging_context *msg = NULL; + + for (msg = msg_ctxs; msg != NULL; msg = msg->next) { + NTSTATUS status = imessaging_reinit(msg); + if (!NT_STATUS_IS_OK(status)) { + return status; + } + } + return NT_STATUS_OK; +} + +/* + create the listening socket and setup the dispatcher +*/ +static struct imessaging_context *imessaging_init_internal( + TALLOC_CTX *mem_ctx, + bool discard_incoming, + struct loadparm_context *lp_ctx, + struct server_id server_id, + struct tevent_context *ev) +{ + NTSTATUS status; + struct imessaging_context *msg; + bool ok; + int ret; + const char *lock_dir = NULL; + int tdb_flags = TDB_INCOMPATIBLE_HASH | TDB_CLEAR_IF_FIRST; + + if (ev == NULL) { + return NULL; + } + + msg = talloc_zero(mem_ctx, struct imessaging_context); + if (msg == NULL) { + return NULL; + } + msg->ev = ev; + msg->discard_incoming = discard_incoming; + if (msg->discard_incoming) { + msg->num_incoming_listeners = 0; + } else { + msg->num_incoming_listeners = 1; + } + + talloc_set_destructor(msg, imessaging_context_destructor); + + /* create the messaging directory if needed */ + + lock_dir = lpcfg_lock_directory(lp_ctx); + if (lock_dir == NULL) { + goto fail; + } + + msg->sock_dir = lpcfg_private_path(msg, lp_ctx, "msg.sock"); + if (msg->sock_dir == NULL) { + goto fail; + } + ok = directory_create_or_exist_strict(msg->sock_dir, geteuid(), 0700); + if (!ok) { + goto fail; + } + + msg->lock_dir = lpcfg_lock_path(msg, lp_ctx, "msg.lock"); + if (msg->lock_dir == NULL) { + goto fail; + } + ok = directory_create_or_exist_strict(msg->lock_dir, geteuid(), 0755); + if (!ok) { + goto fail; + } + + msg->msg_dgm_ref = messaging_dgm_ref( + msg, ev, &server_id.unique_id, msg->sock_dir, msg->lock_dir, + imessaging_dgm_recv, msg, &ret); + + if (msg->msg_dgm_ref == NULL) { + goto fail; + } + + msg->server_id = server_id; + msg->idr = idr_init(msg); + if (msg->idr == NULL) { + goto fail; + } + + msg->dispatch_tree = idr_init(msg); + if (msg->dispatch_tree == NULL) { + goto fail; + } + + msg->start_time = timeval_current(); + + tdb_flags |= lpcfg_tdb_flags(lp_ctx, 0); + + /* + * This context holds a destructor that cleans up any names + * registered on this context on talloc_free() + */ + msg->names = server_id_db_init(msg, server_id, lock_dir, 0, tdb_flags); + if (msg->names == NULL) { + goto fail; + } + + status = imessaging_register(msg, NULL, MSG_PING, ping_message); + if (!NT_STATUS_IS_OK(status)) { + goto fail; + } + status = imessaging_register(msg, NULL, MSG_REQ_POOL_USAGE, + pool_message); + if (!NT_STATUS_IS_OK(status)) { + goto fail; + } + status = imessaging_register(msg, NULL, MSG_IRPC, irpc_handler); + if (!NT_STATUS_IS_OK(status)) { + goto fail; + } + status = imessaging_register(msg, NULL, MSG_REQ_RINGBUF_LOG, + ringbuf_log_msg); + if (!NT_STATUS_IS_OK(status)) { + goto fail; + } + status = imessaging_register(msg, NULL, MSG_DEBUG, + debug_imessage); + if (!NT_STATUS_IS_OK(status)) { + goto fail; + } + status = imessaging_register(msg, NULL, MSG_REQ_DEBUGLEVEL, + debuglevel_imessage); + if (!NT_STATUS_IS_OK(status)) { + goto fail; + } + status = IRPC_REGISTER(msg, irpc, IRPC_UPTIME, irpc_uptime, msg); + if (!NT_STATUS_IS_OK(status)) { + goto fail; + } +#if defined(DEVELOPER) || defined(ENABLE_SELFTEST) + /* + * Register handlers for messages specific to developer and + * self test builds + */ + status = imessaging_register_extra_handlers(msg); + if (!NT_STATUS_IS_OK(status)) { + goto fail; + } +#endif /* defined(DEVELOPER) || defined(ENABLE_SELFTEST) */ + + DLIST_ADD(msg_ctxs, msg); + + return msg; +fail: + talloc_free(msg); + return NULL; +} + +/* + create the listening socket and setup the dispatcher +*/ +struct imessaging_context *imessaging_init(TALLOC_CTX *mem_ctx, + struct loadparm_context *lp_ctx, + struct server_id server_id, + struct tevent_context *ev) +{ + bool discard_incoming = false; + return imessaging_init_internal(mem_ctx, + discard_incoming, + lp_ctx, + server_id, + ev); +} + +struct imessaging_context *imessaging_init_discard_incoming( + TALLOC_CTX *mem_ctx, + struct loadparm_context *lp_ctx, + struct server_id server_id, + struct tevent_context *ev) +{ + bool discard_incoming = true; + return imessaging_init_internal(mem_ctx, + discard_incoming, + lp_ctx, + server_id, + ev); +} + +struct imessaging_post_state { + struct imessaging_context *msg_ctx; + struct imessaging_post_state **busy_ref; + size_t buf_len; + uint8_t buf[]; +}; + +static int imessaging_post_state_destructor(struct imessaging_post_state *state) +{ + if (state->busy_ref != NULL) { + *state->busy_ref = NULL; + state->busy_ref = NULL; + } + return 0; +} + +static void imessaging_post_handler(struct tevent_context *ev, + struct tevent_immediate *ti, + void *private_data) +{ + struct imessaging_post_state *state = talloc_get_type_abort( + private_data, struct imessaging_post_state); + + if (state == NULL) { + return; + } + + /* + * In usecases like using messaging_client_init() with irpc processing + * we may free the imessaging_context during the messaging handler. + * imessaging_post_state is a child of imessaging_context and + * might be implicitly free'ed before the explicit TALLOC_FREE(state). + * + * The busy_ref pointer makes sure the destructor clears + * the local 'state' variable. + */ + + SMB_ASSERT(state->busy_ref == NULL); + state->busy_ref = &state; + + imessaging_dgm_recv(ev, state->buf, state->buf_len, NULL, 0, + state->msg_ctx); + + state->busy_ref = NULL; + TALLOC_FREE(state); +} + +static int imessaging_post_self(struct imessaging_context *msg, + const uint8_t *buf, size_t buf_len) +{ + struct tevent_immediate *ti; + struct imessaging_post_state *state; + + state = talloc_size( + msg, offsetof(struct imessaging_post_state, buf) + buf_len); + if (state == NULL) { + return ENOMEM; + } + talloc_set_name_const(state, "struct imessaging_post_state"); + + talloc_set_destructor(state, imessaging_post_state_destructor); + + ti = tevent_create_immediate(state); + if (ti == NULL) { + TALLOC_FREE(state); + return ENOMEM; + } + + state->msg_ctx = msg; + state->busy_ref = NULL; + state->buf_len = buf_len; + memcpy(state->buf, buf, buf_len); + + tevent_schedule_immediate(ti, msg->ev, imessaging_post_handler, + state); + + return 0; +} + +static void imessaging_dgm_recv(struct tevent_context *ev, + const uint8_t *buf, size_t buf_len, + int *fds, size_t num_fds, + void *private_data) +{ + struct imessaging_context *msg = talloc_get_type_abort( + private_data, struct imessaging_context); + uint32_t msg_type; + struct server_id src, dst; + struct server_id_buf srcbuf, dstbuf; + DATA_BLOB data; + + if (buf_len < MESSAGE_HDR_LENGTH) { + /* Invalid message, ignore */ + return; + } + + if (msg->num_incoming_listeners == 0) { + struct server_id_buf selfbuf; + + message_hdr_get(&msg_type, &src, &dst, buf); + + DBG_DEBUG("not listening - discarding message from " + "src[%s] to dst[%s] (self[%s]) type=0x%x " + "on %s event context\n", + server_id_str_buf(src, &srcbuf), + server_id_str_buf(dst, &dstbuf), + server_id_str_buf(msg->server_id, &selfbuf), + (unsigned)msg_type, + (ev != msg->ev) ? "different" : "main"); + return; + } + + if (ev != msg->ev) { + int ret; + ret = imessaging_post_self(msg, buf, buf_len); + if (ret != 0) { + DBG_WARNING("imessaging_post_self failed: %s\n", + strerror(ret)); + } + return; + } + + message_hdr_get(&msg_type, &src, &dst, buf); + + data.data = discard_const_p(uint8_t, buf + MESSAGE_HDR_LENGTH); + data.length = buf_len - MESSAGE_HDR_LENGTH; + + if ((cluster_id_equal(&dst, &msg->server_id)) || + ((dst.task_id == 0) && (msg->server_id.pid == 0))) { + struct dispatch_fn *d, *next; + + DEBUG(10, ("%s: dst %s matches my id: %s, type=0x%x\n", + __func__, + server_id_str_buf(dst, &dstbuf), + server_id_str_buf(msg->server_id, &srcbuf), + (unsigned)msg_type)); + + d = imessaging_find_dispatch(msg, msg_type); + + for (; d; d = next) { + next = d->next; + d->fn(msg, + d->private_data, + d->msg_type, + src, + num_fds, + fds, + &data); + } + } else { + DEBUG(10, ("%s: Ignoring type=0x%x dst %s, I am %s, \n", + __func__, (unsigned)msg_type, + server_id_str_buf(dst, &dstbuf), + server_id_str_buf(msg->server_id, &srcbuf))); + } +} + +/* + A hack, for the short term until we get 'client only' messaging in place +*/ +struct imessaging_context *imessaging_client_init(TALLOC_CTX *mem_ctx, + struct loadparm_context *lp_ctx, + struct tevent_context *ev) +{ + struct server_id id; + ZERO_STRUCT(id); + id.pid = getpid(); + id.task_id = generate_random(); + id.vnn = NONCLUSTER_VNN; + + /* This is because we are not in the s3 serverid database */ + id.unique_id = SERVERID_UNIQUE_ID_NOT_TO_VERIFY; + + return imessaging_init_discard_incoming(mem_ctx, lp_ctx, id, ev); +} + +/* + a list of registered irpc server functions +*/ +struct irpc_list { + struct irpc_list *next, *prev; + struct GUID uuid; + const struct ndr_interface_table *table; + int callnum; + irpc_function_t fn; + void *private_data; +}; + + +/* + register a irpc server function +*/ +NTSTATUS irpc_register(struct imessaging_context *msg_ctx, + const struct ndr_interface_table *table, + int callnum, irpc_function_t fn, void *private_data) +{ + struct irpc_list *irpc; + + /* override an existing handler, if any */ + for (irpc=msg_ctx->irpc; irpc; irpc=irpc->next) { + if (irpc->table == table && irpc->callnum == callnum) { + break; + } + } + if (irpc == NULL) { + irpc = talloc(msg_ctx, struct irpc_list); + NT_STATUS_HAVE_NO_MEMORY(irpc); + DLIST_ADD(msg_ctx->irpc, irpc); + } + + irpc->table = table; + irpc->callnum = callnum; + irpc->fn = fn; + irpc->private_data = private_data; + irpc->uuid = irpc->table->syntax_id.uuid; + + return NT_STATUS_OK; +} + + +/* + handle an incoming irpc reply message +*/ +static void irpc_handler_reply(struct imessaging_context *msg_ctx, struct irpc_message *m) +{ + struct irpc_request *irpc; + + irpc = (struct irpc_request *)idr_find(msg_ctx->idr, m->header.callid); + if (irpc == NULL) return; + + irpc->incoming.handler(irpc, m); +} + +/* + send a irpc reply +*/ +NTSTATUS irpc_send_reply(struct irpc_message *m, NTSTATUS status) +{ + struct ndr_push *push; + DATA_BLOB packet; + enum ndr_err_code ndr_err; + + m->header.status = status; + + /* setup the reply */ + push = ndr_push_init_ctx(m->ndr); + if (push == NULL) { + status = NT_STATUS_NO_MEMORY; + goto failed; + } + + m->header.flags |= IRPC_FLAG_REPLY; + m->header.creds.token= NULL; + + /* construct the packet */ + ndr_err = ndr_push_irpc_header(push, NDR_SCALARS|NDR_BUFFERS, &m->header); + if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) { + status = ndr_map_error2ntstatus(ndr_err); + goto failed; + } + + ndr_err = m->irpc->table->calls[m->irpc->callnum].ndr_push(push, NDR_OUT, m->data); + if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) { + status = ndr_map_error2ntstatus(ndr_err); + goto failed; + } + + /* send the reply message */ + packet = ndr_push_blob(push); + status = imessaging_send(m->msg_ctx, m->from, MSG_IRPC, &packet); + if (!NT_STATUS_IS_OK(status)) goto failed; + +failed: + talloc_free(m); + return status; +} + +/* + handle an incoming irpc request message +*/ +static void irpc_handler_request(struct imessaging_context *msg_ctx, + struct irpc_message *m) +{ + struct irpc_list *i; + void *r; + enum ndr_err_code ndr_err; + + for (i=msg_ctx->irpc; i; i=i->next) { + if (GUID_equal(&i->uuid, &m->header.uuid) && + i->table->syntax_id.if_version == m->header.if_version && + i->callnum == m->header.callnum) { + break; + } + } + + if (i == NULL) { + /* no registered handler for this message */ + talloc_free(m); + return; + } + + /* allocate space for the structure */ + r = talloc_zero_size(m->ndr, i->table->calls[m->header.callnum].struct_size); + if (r == NULL) goto failed; + + m->ndr->flags |= LIBNDR_FLAG_REF_ALLOC; + + /* parse the request data */ + ndr_err = i->table->calls[i->callnum].ndr_pull(m->ndr, NDR_IN, r); + if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) goto failed; + + /* make the call */ + m->private_data= i->private_data; + m->defer_reply = false; + m->no_reply = false; + m->msg_ctx = msg_ctx; + m->irpc = i; + m->data = r; + + m->header.status = i->fn(m, r); + + if (m->no_reply) { + /* the server function won't ever be replying to this request */ + talloc_free(m); + return; + } + + if (m->defer_reply) { + /* the server function has asked to defer the reply to later */ + talloc_steal(msg_ctx, m); + return; + } + + irpc_send_reply(m, m->header.status); + return; + +failed: + talloc_free(m); +} + +/* + handle an incoming irpc message +*/ +static void irpc_handler(struct imessaging_context *msg_ctx, + void *private_data, + uint32_t msg_type, + struct server_id src, + size_t num_fds, + int *fds, + DATA_BLOB *packet) +{ + struct irpc_message *m; + enum ndr_err_code ndr_err; + + if (num_fds != 0) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + m = talloc(msg_ctx, struct irpc_message); + if (m == NULL) goto failed; + + m->from = src; + + m->ndr = ndr_pull_init_blob(packet, m); + if (m->ndr == NULL) goto failed; + + m->ndr->flags |= LIBNDR_FLAG_REF_ALLOC; + + ndr_err = ndr_pull_irpc_header(m->ndr, NDR_BUFFERS|NDR_SCALARS, &m->header); + if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) goto failed; + + if (m->header.flags & IRPC_FLAG_REPLY) { + irpc_handler_reply(msg_ctx, m); + } else { + irpc_handler_request(msg_ctx, m); + } + return; + +failed: + talloc_free(m); +} + + +/* + destroy a irpc request +*/ +static int irpc_destructor(struct irpc_request *irpc) +{ + if (irpc->callid != -1) { + DLIST_REMOVE(irpc->msg_ctx->requests, irpc); + idr_remove(irpc->msg_ctx->idr, irpc->callid); + if (irpc->msg_ctx->discard_incoming) { + SMB_ASSERT(irpc->msg_ctx->num_incoming_listeners > 0); + } else { + SMB_ASSERT(irpc->msg_ctx->num_incoming_listeners > 1); + } + irpc->msg_ctx->num_incoming_listeners -= 1; + irpc->callid = -1; + } + + return 0; +} + +/* + add a string name that this irpc server can be called on + + It will be removed from the DB either via irpc_remove_name or on + talloc_free(msg_ctx->names). +*/ +NTSTATUS irpc_add_name(struct imessaging_context *msg_ctx, const char *name) +{ + int ret; + + ret = server_id_db_add(msg_ctx->names, name); + if (ret != 0) { + return map_nt_error_from_unix_common(ret); + } + return NT_STATUS_OK; +} + +static int all_servers_func(const char *name, unsigned num_servers, + const struct server_id *servers, + void *private_data) +{ + struct irpc_name_records *name_records = talloc_get_type( + private_data, struct irpc_name_records); + struct irpc_name_record *name_record; + uint32_t i; + + name_records->names + = talloc_realloc(name_records, name_records->names, + struct irpc_name_record *, name_records->num_records+1); + if (!name_records->names) { + return -1; + } + + name_records->names[name_records->num_records] = name_record + = talloc(name_records->names, + struct irpc_name_record); + if (!name_record) { + return -1; + } + + name_records->num_records++; + + name_record->name = talloc_strdup(name_record, name); + if (!name_record->name) { + return -1; + } + + name_record->count = num_servers; + name_record->ids = talloc_array(name_record, struct server_id, + num_servers); + if (name_record->ids == NULL) { + return -1; + } + for (i=0;i<name_record->count;i++) { + name_record->ids[i] = servers[i]; + } + return 0; +} + +/* + return a list of server ids for a server name +*/ +struct irpc_name_records *irpc_all_servers(struct imessaging_context *msg_ctx, + TALLOC_CTX *mem_ctx) +{ + int ret; + struct irpc_name_records *name_records = talloc_zero(mem_ctx, struct irpc_name_records); + if (name_records == NULL) { + return NULL; + } + + ret = server_id_db_traverse_read(msg_ctx->names, all_servers_func, + name_records); + if (ret == -1) { + TALLOC_FREE(name_records); + return NULL; + } + + return name_records; +} + +/* + remove a name from a messaging context +*/ +void irpc_remove_name(struct imessaging_context *msg_ctx, const char *name) +{ + server_id_db_remove(msg_ctx->names, name); +} + +struct server_id imessaging_get_server_id(struct imessaging_context *msg_ctx) +{ + return msg_ctx->server_id; +} + +struct irpc_bh_state { + struct imessaging_context *msg_ctx; + struct server_id server_id; + const struct ndr_interface_table *table; + uint32_t timeout; + struct security_token *token; +}; + +static bool irpc_bh_is_connected(struct dcerpc_binding_handle *h) +{ + struct irpc_bh_state *hs = dcerpc_binding_handle_data(h, + struct irpc_bh_state); + + if (!hs->msg_ctx) { + return false; + } + + return true; +} + +static uint32_t irpc_bh_set_timeout(struct dcerpc_binding_handle *h, + uint32_t timeout) +{ + struct irpc_bh_state *hs = dcerpc_binding_handle_data(h, + struct irpc_bh_state); + uint32_t old = hs->timeout; + + hs->timeout = timeout; + + return old; +} + +struct irpc_bh_raw_call_state { + struct irpc_request *irpc; + uint32_t opnum; + DATA_BLOB in_data; + DATA_BLOB in_packet; + DATA_BLOB out_data; +}; + +static void irpc_bh_raw_call_incoming_handler(struct irpc_request *irpc, + struct irpc_message *m); + +static struct tevent_req *irpc_bh_raw_call_send(TALLOC_CTX *mem_ctx, + struct tevent_context *ev, + struct dcerpc_binding_handle *h, + const struct GUID *object, + uint32_t opnum, + uint32_t in_flags, + const uint8_t *in_data, + size_t in_length) +{ + struct irpc_bh_state *hs = + dcerpc_binding_handle_data(h, + struct irpc_bh_state); + struct tevent_req *req; + struct irpc_bh_raw_call_state *state; + bool ok; + struct irpc_header header; + struct ndr_push *ndr; + NTSTATUS status; + enum ndr_err_code ndr_err; + + req = tevent_req_create(mem_ctx, &state, + struct irpc_bh_raw_call_state); + if (req == NULL) { + return NULL; + } + state->opnum = opnum; + state->in_data.data = discard_const_p(uint8_t, in_data); + state->in_data.length = in_length; + + ok = irpc_bh_is_connected(h); + if (!ok) { + tevent_req_nterror(req, NT_STATUS_CONNECTION_DISCONNECTED); + return tevent_req_post(req, ev); + } + + state->irpc = talloc_zero(state, struct irpc_request); + if (tevent_req_nomem(state->irpc, req)) { + return tevent_req_post(req, ev); + } + + state->irpc->msg_ctx = hs->msg_ctx; + state->irpc->callid = idr_get_new(hs->msg_ctx->idr, + state->irpc, UINT16_MAX); + if (state->irpc->callid == -1) { + tevent_req_nterror(req, NT_STATUS_INSUFFICIENT_RESOURCES); + return tevent_req_post(req, ev); + } + state->irpc->incoming.handler = irpc_bh_raw_call_incoming_handler; + state->irpc->incoming.private_data = req; + + /* make sure we accept incoming messages */ + SMB_ASSERT(state->irpc->msg_ctx->num_incoming_listeners < UINT64_MAX); + state->irpc->msg_ctx->num_incoming_listeners += 1; + DLIST_ADD_END(state->irpc->msg_ctx->requests, state->irpc); + talloc_set_destructor(state->irpc, irpc_destructor); + + /* setup the header */ + header.uuid = hs->table->syntax_id.uuid; + + header.if_version = hs->table->syntax_id.if_version; + header.callid = state->irpc->callid; + header.callnum = state->opnum; + header.flags = 0; + header.status = NT_STATUS_OK; + header.creds.token= hs->token; + + /* construct the irpc packet */ + ndr = ndr_push_init_ctx(state->irpc); + if (tevent_req_nomem(ndr, req)) { + return tevent_req_post(req, ev); + } + + ndr_err = ndr_push_irpc_header(ndr, NDR_SCALARS|NDR_BUFFERS, &header); + status = ndr_map_error2ntstatus(ndr_err); + if (!NT_STATUS_IS_OK(status)) { + tevent_req_nterror(req, status); + return tevent_req_post(req, ev); + } + + ndr_err = ndr_push_bytes(ndr, in_data, in_length); + status = ndr_map_error2ntstatus(ndr_err); + if (!NT_STATUS_IS_OK(status)) { + tevent_req_nterror(req, status); + return tevent_req_post(req, ev); + } + + /* and send it */ + state->in_packet = ndr_push_blob(ndr); + status = imessaging_send(hs->msg_ctx, hs->server_id, + MSG_IRPC, &state->in_packet); + if (!NT_STATUS_IS_OK(status)) { + tevent_req_nterror(req, status); + return tevent_req_post(req, ev); + } + + if (hs->timeout != IRPC_CALL_TIMEOUT_INF) { + /* set timeout-callback in case caller wants that */ + ok = tevent_req_set_endtime(req, ev, timeval_current_ofs(hs->timeout, 0)); + if (!ok) { + return tevent_req_post(req, ev); + } + } + + return req; +} + +static void irpc_bh_raw_call_incoming_handler(struct irpc_request *irpc, + struct irpc_message *m) +{ + struct tevent_req *req = + talloc_get_type_abort(irpc->incoming.private_data, + struct tevent_req); + struct irpc_bh_raw_call_state *state = + tevent_req_data(req, + struct irpc_bh_raw_call_state); + + talloc_steal(state, m); + + if (!NT_STATUS_IS_OK(m->header.status)) { + tevent_req_nterror(req, m->header.status); + return; + } + + state->out_data = data_blob_talloc(state, + m->ndr->data + m->ndr->offset, + m->ndr->data_size - m->ndr->offset); + if ((m->ndr->data_size - m->ndr->offset) > 0 && !state->out_data.data) { + tevent_req_oom(req); + return; + } + + tevent_req_done(req); +} + +static NTSTATUS irpc_bh_raw_call_recv(struct tevent_req *req, + TALLOC_CTX *mem_ctx, + uint8_t **out_data, + size_t *out_length, + uint32_t *out_flags) +{ + struct irpc_bh_raw_call_state *state = + tevent_req_data(req, + struct irpc_bh_raw_call_state); + NTSTATUS status; + + if (tevent_req_is_nterror(req, &status)) { + tevent_req_received(req); + return status; + } + + *out_data = talloc_move(mem_ctx, &state->out_data.data); + *out_length = state->out_data.length; + *out_flags = 0; + tevent_req_received(req); + return NT_STATUS_OK; +} + +struct irpc_bh_disconnect_state { + uint8_t _dummy; +}; + +static struct tevent_req *irpc_bh_disconnect_send(TALLOC_CTX *mem_ctx, + struct tevent_context *ev, + struct dcerpc_binding_handle *h) +{ + struct irpc_bh_state *hs = dcerpc_binding_handle_data(h, + struct irpc_bh_state); + struct tevent_req *req; + struct irpc_bh_disconnect_state *state; + bool ok; + + req = tevent_req_create(mem_ctx, &state, + struct irpc_bh_disconnect_state); + if (req == NULL) { + return NULL; + } + + ok = irpc_bh_is_connected(h); + if (!ok) { + tevent_req_nterror(req, NT_STATUS_CONNECTION_DISCONNECTED); + return tevent_req_post(req, ev); + } + + hs->msg_ctx = NULL; + + tevent_req_done(req); + return tevent_req_post(req, ev); +} + +static NTSTATUS irpc_bh_disconnect_recv(struct tevent_req *req) +{ + NTSTATUS status; + + if (tevent_req_is_nterror(req, &status)) { + tevent_req_received(req); + return status; + } + + tevent_req_received(req); + return NT_STATUS_OK; +} + +static bool irpc_bh_ref_alloc(struct dcerpc_binding_handle *h) +{ + return true; +} + +static const struct dcerpc_binding_handle_ops irpc_bh_ops = { + .name = "wbint", + .is_connected = irpc_bh_is_connected, + .set_timeout = irpc_bh_set_timeout, + .raw_call_send = irpc_bh_raw_call_send, + .raw_call_recv = irpc_bh_raw_call_recv, + .disconnect_send = irpc_bh_disconnect_send, + .disconnect_recv = irpc_bh_disconnect_recv, + + .ref_alloc = irpc_bh_ref_alloc, +}; + +/* initialise a irpc binding handle */ +struct dcerpc_binding_handle *irpc_binding_handle(TALLOC_CTX *mem_ctx, + struct imessaging_context *msg_ctx, + struct server_id server_id, + const struct ndr_interface_table *table) +{ + struct dcerpc_binding_handle *h; + struct irpc_bh_state *hs; + + h = dcerpc_binding_handle_create(mem_ctx, + &irpc_bh_ops, + NULL, + table, + &hs, + struct irpc_bh_state, + __location__); + if (h == NULL) { + return NULL; + } + hs->msg_ctx = msg_ctx; + hs->server_id = server_id; + hs->table = table; + hs->timeout = IRPC_CALL_TIMEOUT; + + return h; +} + +struct dcerpc_binding_handle *irpc_binding_handle_by_name(TALLOC_CTX *mem_ctx, + struct imessaging_context *msg_ctx, + const char *dest_task, + const struct ndr_interface_table *table) +{ + struct dcerpc_binding_handle *h; + unsigned num_sids; + struct server_id *sids; + struct server_id sid; + NTSTATUS status; + + /* find the server task */ + + status = irpc_servers_byname(msg_ctx, mem_ctx, dest_task, + &num_sids, &sids); + if (!NT_STATUS_IS_OK(status)) { + errno = EADDRNOTAVAIL; + return NULL; + } + sid = sids[0]; + talloc_free(sids); + + h = irpc_binding_handle(mem_ctx, msg_ctx, + sid, table); + if (h == NULL) { + return NULL; + } + + return h; +} + +void irpc_binding_handle_add_security_token(struct dcerpc_binding_handle *h, + struct security_token *token) +{ + struct irpc_bh_state *hs = + dcerpc_binding_handle_data(h, + struct irpc_bh_state); + + hs->token = token; +} diff --git a/source4/lib/messaging/messaging.h b/source4/lib/messaging/messaging.h new file mode 100644 index 0000000..e7ae9e8 --- /dev/null +++ b/source4/lib/messaging/messaging.h @@ -0,0 +1,70 @@ +/* + Unix SMB/CIFS implementation. + messages.c header + Copyright (C) Andrew Tridgell 2000 + Copyright (C) 2001, 2002 by Martin Pool + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +#ifndef _SOURCE4_LIB_MESSAGING_MESSAGES_H_ +#define _SOURCE4_LIB_MESSAGING_MESSAGES_H_ + +#include "librpc/gen_ndr/server_id.h" +#include "librpc/gen_ndr/messaging.h" + +struct imessaging_context; + +/* taskid for messaging of parent process */ +#define SAMBA_PARENT_TASKID 0 + +typedef void (*msg_callback_t)( + struct imessaging_context *msg, + void *private_data, + uint32_t msg_type, + struct server_id server_id, + size_t num_fds, + int *fds, + DATA_BLOB *data); + +NTSTATUS imessaging_send(struct imessaging_context *msg, struct server_id server, + uint32_t msg_type, const DATA_BLOB *data); +NTSTATUS imessaging_register(struct imessaging_context *msg, void *private_data, + uint32_t msg_type, + msg_callback_t fn); +NTSTATUS imessaging_register_tmp(struct imessaging_context *msg, void *private_data, + msg_callback_t fn, uint32_t *msg_type); +struct imessaging_context *imessaging_init(TALLOC_CTX *mem_ctx, + struct loadparm_context *lp_ctx, + struct server_id server_id, + struct tevent_context *ev); +struct imessaging_context *imessaging_init_discard_incoming( + TALLOC_CTX *mem_ctx, + struct loadparm_context *lp_ctx, + struct server_id server_id, + struct tevent_context *ev); +void imessaging_dgm_unref_ev(struct tevent_context *ev); +NTSTATUS imessaging_reinit_all(void); +int imessaging_cleanup(struct imessaging_context *msg); +struct imessaging_context *imessaging_client_init(TALLOC_CTX *mem_ctx, + struct loadparm_context *lp_ctx, + struct tevent_context *ev); +NTSTATUS imessaging_send_ptr(struct imessaging_context *msg, struct server_id server, + uint32_t msg_type, void *ptr); +void imessaging_deregister(struct imessaging_context *msg, uint32_t msg_type, void *private_data); +struct server_id imessaging_get_server_id(struct imessaging_context *msg_ctx); +NTSTATUS imessaging_process_cleanup(struct imessaging_context *msg_ctx, + pid_t pid); + +#endif diff --git a/source4/lib/messaging/messaging_handlers.c b/source4/lib/messaging/messaging_handlers.c new file mode 100644 index 0000000..57e3e1c --- /dev/null +++ b/source4/lib/messaging/messaging_handlers.c @@ -0,0 +1,135 @@ +/* + Unix SMB/CIFS implementation. + + Handers for non core Samba internal messages + + Handlers for messages that are only included in developer and self test + builds. + + Copyright (C) Andrew Bartlett <abartlet@samba.org> 2018 + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +#include "includes.h" +#include "lib/util/server_id.h" +#include "messaging/messaging.h" +#include "messaging/messaging_internal.h" + +#if defined(DEVELOPER) || defined(ENABLE_SELFTEST) + +/* + * Inject a fault into the currently running process + */ +static void do_inject_fault(struct imessaging_context *msg, + void *private_data, + uint32_t msg_type, + struct server_id src, + size_t num_fds, + int *fds, + DATA_BLOB *data) +{ + int sig; + struct server_id_buf tmp; + + if (num_fds != 0) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + if (data->length != sizeof(sig)) { + DBG_ERR("Process %s sent bogus signal injection request\n", + server_id_str_buf(src, &tmp)); + return; + } + + sig = *(int *)data->data; + if (sig == -1) { + DBG_ERR("Process %s requested an iternal failure, " + "calling exit(1)\n", + server_id_str_buf(src, &tmp)); + exit(1); + } + +#if HAVE_STRSIGNAL + DBG_ERR("Process %s requested injection of signal %d (%s)\n", + server_id_str_buf(src, &tmp), + sig, + strsignal(sig)); +#else + DBG_ERR("Process %s requested injection of signal %d\n", + server_id_str_buf(src, &tmp), + sig); +#endif + + kill(getpid(), sig); +} + +/* + * Cause the current process to sleep for a specified number of seconds + */ +static void do_sleep(struct imessaging_context *msg, + void *private_data, + uint32_t msg_type, + struct server_id src, + size_t num_fds, + int *fds, + DATA_BLOB *data) +{ + unsigned int seconds; + struct server_id_buf tmp; + + if (num_fds != 0) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + if (data->length != sizeof(seconds)) { + DBG_ERR("Process %s sent bogus sleep request\n", + server_id_str_buf(src, &tmp)); + return; + } + + seconds = *(unsigned int *)data->data; + DBG_ERR("Process %s requested a sleep of %u seconds\n", + server_id_str_buf(src, &tmp), + seconds); + sleep(seconds); + DBG_ERR("Restarting after %u second sleep requested by process %s\n", + seconds, + server_id_str_buf(src, &tmp)); +} + +/* + * Register the extra messaging handlers + */ +NTSTATUS imessaging_register_extra_handlers(struct imessaging_context *msg) +{ + NTSTATUS status; + + status = imessaging_register( + msg, NULL, MSG_SMB_INJECT_FAULT, do_inject_fault); + if (!NT_STATUS_IS_OK(status)) { + return status; + } + + status = imessaging_register(msg, NULL, MSG_SMB_SLEEP, do_sleep); + if (!NT_STATUS_IS_OK(status)) { + return status; + } + + return NT_STATUS_OK; +} + +#endif /* defined(DEVELOPER) || defined(ENABLE_SELFTEST) */ diff --git a/source4/lib/messaging/messaging_internal.h b/source4/lib/messaging/messaging_internal.h new file mode 100644 index 0000000..6281bda --- /dev/null +++ b/source4/lib/messaging/messaging_internal.h @@ -0,0 +1,50 @@ +/* + Unix SMB/CIFS implementation. + + Samba internal messaging functions + + Copyright (C) Andrew Tridgell 2004 + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +struct irpc_request; + +struct imessaging_context { + struct imessaging_context *prev, *next; + struct tevent_context *ev; + struct server_id server_id; + const char *sock_dir; + const char *lock_dir; + struct dispatch_fn **dispatch; + uint32_t num_types; + struct idr_context *dispatch_tree; + struct irpc_list *irpc; + struct idr_context *idr; + struct irpc_request *requests; + struct server_id_db *names; + struct timeval start_time; + void *msg_dgm_ref; + /* + * The number of instances waiting for incoming + * messages. By default it's always greater than 0. + * + * If it's 0 we'll discard incoming messages, + * see imessaging_init_discard_imcoming(). + */ + bool discard_incoming; + uint64_t num_incoming_listeners; +}; + +NTSTATUS imessaging_register_extra_handlers(struct imessaging_context *msg); diff --git a/source4/lib/messaging/messaging_send.c b/source4/lib/messaging/messaging_send.c new file mode 100644 index 0000000..24cdce3 --- /dev/null +++ b/source4/lib/messaging/messaging_send.c @@ -0,0 +1,115 @@ +/* + Unix SMB/CIFS implementation. + + Samba internal messaging functions (send). + + Copyright (C) Andrew Tridgell 2004 + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +#include "includes.h" +#include "messaging/messaging.h" +#include "messaging/irpc.h" +#include "lib/messaging/messages_dgm.h" +#include "lib/messaging/messages_dgm_ref.h" +#include "../source3/lib/messages_util.h" +#include "messaging/messaging_internal.h" +#include "lib/util/server_id_db.h" +#include "cluster/cluster.h" +#include "../lib/util/unix_privs.h" + +/* + * This file is for functions that can be called from auth_log without + * depending on all of dcerpc and so cause dep loops. + */ + +/* + return a list of server ids for a server name +*/ +NTSTATUS irpc_servers_byname(struct imessaging_context *msg_ctx, + TALLOC_CTX *mem_ctx, const char *name, + unsigned *num_servers, + struct server_id **servers) +{ + int ret; + + ret = server_id_db_lookup(msg_ctx->names, name, mem_ctx, + num_servers, servers); + if (ret != 0) { + return map_nt_error_from_unix_common(ret); + } + return NT_STATUS_OK; +} + +/* + Send a message to a particular server +*/ +NTSTATUS imessaging_send(struct imessaging_context *msg, struct server_id server, + uint32_t msg_type, const DATA_BLOB *data) +{ + uint8_t hdr[MESSAGE_HDR_LENGTH]; + struct iovec iov[2]; + int num_iov, ret; + pid_t pid; + void *priv; + + if (!cluster_node_equal(&msg->server_id, &server)) { + /* No cluster in source4... */ + return NT_STATUS_OK; + } + + message_hdr_put(hdr, msg_type, msg->server_id, server); + + iov[0] = (struct iovec) { .iov_base = &hdr, .iov_len = sizeof(hdr) }; + num_iov = 1; + + if (data != NULL) { + iov[1] = (struct iovec) { .iov_base = data->data, + .iov_len = data->length }; + num_iov += 1; + } + + pid = server.pid; + if (pid == 0) { + pid = getpid(); + } + + ret = messaging_dgm_send(pid, iov, num_iov, NULL, 0); + + if (ret == EACCES) { + priv = root_privileges(); + ret = messaging_dgm_send(pid, iov, num_iov, NULL, 0); + TALLOC_FREE(priv); + } + + if (ret != 0) { + return map_nt_error_from_unix_common(ret); + } + return NT_STATUS_OK; +} + +/* + Send a message to a particular server, with the message containing a single pointer +*/ +NTSTATUS imessaging_send_ptr(struct imessaging_context *msg, struct server_id server, + uint32_t msg_type, void *ptr) +{ + DATA_BLOB blob; + + blob.data = (uint8_t *)&ptr; + blob.length = sizeof(void *); + + return imessaging_send(msg, server, msg_type, &blob); +} diff --git a/source4/lib/messaging/pymessaging.c b/source4/lib/messaging/pymessaging.c new file mode 100644 index 0000000..148c941 --- /dev/null +++ b/source4/lib/messaging/pymessaging.c @@ -0,0 +1,551 @@ +/* + Unix SMB/CIFS implementation. + Copyright © Jelmer Vernooij <jelmer@samba.org> 2008 + + Based on the equivalent for EJS: + Copyright © Andrew Tridgell <tridge@samba.org> 2005 + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +#include <Python.h> +#include "python/py3compat.h" +#include "includes.h" +#include "python/modules.h" +#include "libcli/util/pyerrors.h" +#include "librpc/rpc/pyrpc_util.h" +#include "librpc/ndr/libndr.h" +#include "lib/messaging/messaging.h" +#include "lib/messaging/irpc.h" +#include "lib/events/events.h" +#include "cluster/cluster.h" +#include "param/param.h" +#include "param/pyparam.h" +#include "librpc/rpc/dcerpc.h" +#include "librpc/gen_ndr/server_id.h" +#include <pytalloc.h> +#include "messaging_internal.h" + + +extern PyTypeObject imessaging_Type; + +static bool server_id_from_py(PyObject *object, struct server_id *server_id) +{ + if (!PyTuple_Check(object)) { + if (!py_check_dcerpc_type(object, "samba.dcerpc.server_id", "server_id")) { + + PyErr_SetString(PyExc_ValueError, "Expected tuple or server_id"); + return false; + } + *server_id = *pytalloc_get_type(object, struct server_id); + return true; + } + if (PyTuple_Size(object) == 3) { + unsigned long long pid; + int task_id, vnn; + + if (!PyArg_ParseTuple(object, "Kii", &pid, &task_id, &vnn)) { + return false; + } + server_id->pid = pid; + server_id->task_id = task_id; + server_id->vnn = vnn; + return true; + } else if (PyTuple_Size(object) == 2) { + unsigned long long pid; + int task_id; + if (!PyArg_ParseTuple(object, "Ki", &pid, &task_id)) + return false; + *server_id = cluster_id(pid, task_id); + return true; + } else { + unsigned long long pid = getpid(); + int task_id; + if (!PyArg_ParseTuple(object, "i", &task_id)) + return false; + *server_id = cluster_id(pid, task_id); + return true; + } +} + +typedef struct { + PyObject_HEAD + TALLOC_CTX *mem_ctx; + struct imessaging_context *msg_ctx; +} imessaging_Object; + +static PyObject *py_imessaging_connect(PyTypeObject *self, PyObject *args, PyObject *kwargs) +{ + struct tevent_context *ev; + const char *kwnames[] = { "own_id", "lp_ctx", NULL }; + PyObject *own_id = Py_None; + PyObject *py_lp_ctx = Py_None; + imessaging_Object *ret; + struct loadparm_context *lp_ctx; + + if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|OO:connect", + discard_const_p(char *, kwnames), &own_id, &py_lp_ctx)) { + return NULL; + } + + ret = PyObject_New(imessaging_Object, &imessaging_Type); + if (ret == NULL) + return NULL; + + ret->mem_ctx = talloc_new(NULL); + + lp_ctx = lpcfg_from_py_object(ret->mem_ctx, py_lp_ctx); + if (lp_ctx == NULL) { + PyErr_SetString(PyExc_RuntimeError, "imessaging_connect unable to interpret loadparm_context"); + talloc_free(ret->mem_ctx); + return NULL; + } + + ev = s4_event_context_init(ret->mem_ctx); + + if (own_id != Py_None) { + struct server_id server_id; + + if (!server_id_from_py(own_id, &server_id)) + return NULL; + + ret->msg_ctx = imessaging_init(ret->mem_ctx, + lp_ctx, + server_id, + ev); + } else { + ret->msg_ctx = imessaging_client_init(ret->mem_ctx, + lp_ctx, + ev); + } + + if (ret->msg_ctx == NULL) { + PyErr_SetString(PyExc_RuntimeError, "imessaging_connect unable to create a messaging context"); + talloc_free(ret->mem_ctx); + return NULL; + } + + return (PyObject *)ret; +} + +static void py_imessaging_dealloc(PyObject *self) +{ + imessaging_Object *iface = (imessaging_Object *)self; + talloc_free(iface->msg_ctx); + self->ob_type->tp_free(self); +} + +static PyObject *py_imessaging_send(PyObject *self, PyObject *args, PyObject *kwargs) +{ + imessaging_Object *iface = (imessaging_Object *)self; + uint32_t msg_type; + DATA_BLOB data; + PyObject *target; + NTSTATUS status; + struct server_id server; + const char *kwnames[] = { "target", "msg_type", "data", NULL }; + Py_ssize_t length; + + if (!PyArg_ParseTupleAndKeywords(args, kwargs, "OIs#:send", + discard_const_p(char *, kwnames), &target, &msg_type, &data.data, &length)) { + + return NULL; + } + + data.length = length; + + if (!server_id_from_py(target, &server)) + return NULL; + + status = imessaging_send(iface->msg_ctx, server, msg_type, &data); + if (NT_STATUS_IS_ERR(status)) { + PyErr_SetNTSTATUS(status); + return NULL; + } + + Py_RETURN_NONE; +} + +static void py_msg_callback_wrapper(struct imessaging_context *msg, + void *private_data, + uint32_t msg_type, + struct server_id server_id, + size_t num_fds, + int *fds, + DATA_BLOB *data) +{ + PyObject *py_server_id, *callback_and_tuple = (PyObject *)private_data; + PyObject *callback, *py_private; + + struct server_id *p_server_id = talloc(NULL, struct server_id); + + if (num_fds != 0) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + if (!p_server_id) { + PyErr_NoMemory(); + return; + } + *p_server_id = server_id; + + if (!PyArg_ParseTuple(callback_and_tuple, "OO", + &callback, + &py_private)) { + return; + } + + py_server_id = py_return_ndr_struct("samba.dcerpc.server_id", "server_id", p_server_id, p_server_id); + talloc_unlink(NULL, p_server_id); + + PyObject_CallFunction(callback, discard_const_p(char, "OiOs#"), + py_private, + msg_type, + py_server_id, + data->data, data->length); +} + +static PyObject *py_imessaging_register(PyObject *self, PyObject *args, PyObject *kwargs) +{ + imessaging_Object *iface = (imessaging_Object *)self; + int msg_type = -1; + PyObject *callback_and_context; + NTSTATUS status; + const char *kwnames[] = { "callback_and_context", "msg_type", NULL }; + + if (!PyArg_ParseTupleAndKeywords(args, kwargs, "O|i:register", + discard_const_p(char *, kwnames), + &callback_and_context, &msg_type)) { + return NULL; + } + if (!PyTuple_Check(callback_and_context) + || PyTuple_Size(callback_and_context) != 2) { + PyErr_SetString(PyExc_ValueError, "Expected of size 2 for callback_and_context"); + return NULL; + } + + Py_INCREF(callback_and_context); + + if (msg_type == -1) { + uint32_t msg_type32 = msg_type; + status = imessaging_register_tmp(iface->msg_ctx, callback_and_context, + py_msg_callback_wrapper, &msg_type32); + msg_type = msg_type32; + } else { + status = imessaging_register(iface->msg_ctx, callback_and_context, + msg_type, py_msg_callback_wrapper); + } + if (NT_STATUS_IS_ERR(status)) { + PyErr_SetNTSTATUS(status); + return NULL; + } + + return PyLong_FromLong(msg_type); +} + +static PyObject *py_imessaging_deregister(PyObject *self, PyObject *args, PyObject *kwargs) +{ + imessaging_Object *iface = (imessaging_Object *)self; + int msg_type = -1; + PyObject *callback; + const char *kwnames[] = { "callback", "msg_type", NULL }; + + if (!PyArg_ParseTupleAndKeywords(args, kwargs, "O|i:deregister", + discard_const_p(char *, kwnames), &callback, &msg_type)) { + return NULL; + } + + imessaging_deregister(iface->msg_ctx, msg_type, callback); + + Py_RETURN_NONE; +} + +static void simple_timer_handler(struct tevent_context *ev, + struct tevent_timer *te, + struct timeval current_time, + void *private_data) +{ + return; +} + +static PyObject *py_imessaging_loop_once(PyObject *self, PyObject *args, PyObject *kwargs) +{ + imessaging_Object *iface = (imessaging_Object *)self; + double offset; + int seconds; + struct timeval next_event; + struct tevent_timer *timer = NULL; + const char *kwnames[] = { "timeout", NULL }; + + TALLOC_CTX *frame = talloc_stackframe(); + + if (!PyArg_ParseTupleAndKeywords(args, kwargs, "d", + discard_const_p(char *, kwnames), &offset)) { + TALLOC_FREE(frame); + return NULL; + } + + if (offset != 0.0) { + seconds = offset; + offset -= seconds; + next_event = tevent_timeval_current_ofs(seconds, (int)(offset*1000000)); + + timer = tevent_add_timer(iface->msg_ctx->ev, frame, next_event, simple_timer_handler, + NULL); + if (timer == NULL) { + PyErr_NoMemory(); + TALLOC_FREE(frame); + return NULL; + } + } + + tevent_loop_once(iface->msg_ctx->ev); + + TALLOC_FREE(frame); + + Py_RETURN_NONE; +} + +static PyObject *py_irpc_add_name(PyObject *self, PyObject *args) +{ + imessaging_Object *iface = (imessaging_Object *)self; + char *server_name; + NTSTATUS status; + + if (!PyArg_ParseTuple(args, "s", &server_name)) { + return NULL; + } + + status = irpc_add_name(iface->msg_ctx, server_name); + if (!NT_STATUS_IS_OK(status)) { + PyErr_SetNTSTATUS(status); + return NULL; + } + + Py_RETURN_NONE; +} + +static PyObject *py_irpc_remove_name(PyObject *self, PyObject *args) +{ + imessaging_Object *iface = (imessaging_Object *)self; + char *server_name; + + if (!PyArg_ParseTuple(args, "s", &server_name)) { + return NULL; + } + + irpc_remove_name(iface->msg_ctx, server_name); + + Py_RETURN_NONE; +} + +static PyObject *py_irpc_servers_byname(PyObject *self, PyObject *args) +{ + imessaging_Object *iface = (imessaging_Object *)self; + char *server_name; + unsigned i, num_ids; + struct server_id *ids; + PyObject *pylist; + TALLOC_CTX *mem_ctx = talloc_new(NULL); + NTSTATUS status; + + if (!mem_ctx) { + PyErr_NoMemory(); + return NULL; + } + + if (!PyArg_ParseTuple(args, "s", &server_name)) { + TALLOC_FREE(mem_ctx); + return NULL; + } + + status = irpc_servers_byname(iface->msg_ctx, mem_ctx, server_name, + &num_ids, &ids); + if (!NT_STATUS_IS_OK(status)) { + TALLOC_FREE(mem_ctx); + PyErr_SetString(PyExc_KeyError, "No such name"); + return NULL; + } + + pylist = PyList_New(num_ids); + if (pylist == NULL) { + TALLOC_FREE(mem_ctx); + PyErr_NoMemory(); + return NULL; + } + for (i = 0; i < num_ids; i++) { + PyObject *py_server_id; + struct server_id *p_server_id = talloc(NULL, struct server_id); + if (!p_server_id) { + PyErr_NoMemory(); + return NULL; + } + *p_server_id = ids[i]; + + py_server_id = py_return_ndr_struct("samba.dcerpc.server_id", "server_id", p_server_id, p_server_id); + if (!py_server_id) { + return NULL; + } + PyList_SetItem(pylist, i, py_server_id); + talloc_unlink(NULL, p_server_id); + } + TALLOC_FREE(mem_ctx); + return pylist; +} + +static PyObject *py_irpc_all_servers(PyObject *self, + PyObject *Py_UNUSED(ignored)) +{ + imessaging_Object *iface = (imessaging_Object *)self; + PyObject *pylist; + int i; + struct irpc_name_records *records; + TALLOC_CTX *mem_ctx = talloc_new(NULL); + if (!mem_ctx) { + PyErr_NoMemory(); + return NULL; + } + + records = irpc_all_servers(iface->msg_ctx, mem_ctx); + if (records == NULL) { + return NULL; + } + + pylist = PyList_New(records->num_records); + if (pylist == NULL) { + TALLOC_FREE(mem_ctx); + PyErr_NoMemory(); + return NULL; + } + for (i = 0; i < records->num_records; i++) { + PyObject *py_name_record + = py_return_ndr_struct("samba.dcerpc.irpc", + "name_record", + records->names[i], + records->names[i]); + if (!py_name_record) { + return NULL; + } + PyList_SetItem(pylist, i, + py_name_record); + } + TALLOC_FREE(mem_ctx); + return pylist; +} + +static PyMethodDef py_imessaging_methods[] = { + { "send", PY_DISCARD_FUNC_SIG(PyCFunction, py_imessaging_send), + METH_VARARGS|METH_KEYWORDS, + "S.send(target, msg_type, data) -> None\nSend a message" }, + { "register", PY_DISCARD_FUNC_SIG(PyCFunction, py_imessaging_register), + METH_VARARGS|METH_KEYWORDS, + "S.register((callback, context), msg_type=None) -> msg_type\nRegister a message handler. " + "The callback and context must be supplied as a two-element tuple." }, + { "deregister", PY_DISCARD_FUNC_SIG(PyCFunction, + py_imessaging_deregister), + METH_VARARGS|METH_KEYWORDS, + "S.deregister((callback, context), msg_type) -> None\nDeregister a message handler " + "The callback and context must be supplied as the exact same two-element tuple " + "as was used as registration time." }, + { "loop_once", PY_DISCARD_FUNC_SIG(PyCFunction, + py_imessaging_loop_once), + METH_VARARGS|METH_KEYWORDS, + "S.loop_once(timeout) -> None\n" + "Loop on the internal event context until we get an event " + "(which might be a message calling the callback), " + "timeout after timeout seconds (if not 0)" }, + { "irpc_add_name", (PyCFunction)py_irpc_add_name, METH_VARARGS, + "S.irpc_add_name(name) -> None\n" + "Add this context to the list of server_id values that " + "are registered for a particular name" }, + { "irpc_remove_name", (PyCFunction)py_irpc_remove_name, METH_VARARGS, + "S.irpc_remove_name(name) -> None\n" + "Remove this context from the list of server_id values that " + "are registered for a particular name" }, + { "irpc_servers_byname", (PyCFunction)py_irpc_servers_byname, METH_VARARGS, + "S.irpc_servers_byname(name) -> list\nGet list of server_id values that are registered for a particular name" }, + { "irpc_all_servers", (PyCFunction)py_irpc_all_servers, METH_NOARGS, + "S.irpc_all_servers() -> list\n" + "Get list of all registered names and the associated server_id values" }, + { NULL, NULL, 0, NULL } +}; + +static PyObject *py_imessaging_server_id(PyObject *obj, void *closure) +{ + imessaging_Object *iface = (imessaging_Object *)obj; + PyObject *py_server_id; + struct server_id server_id = imessaging_get_server_id(iface->msg_ctx); + struct server_id *p_server_id = talloc(NULL, struct server_id); + if (!p_server_id) { + PyErr_NoMemory(); + return NULL; + } + *p_server_id = server_id; + + py_server_id = py_return_ndr_struct("samba.dcerpc.server_id", "server_id", p_server_id, p_server_id); + talloc_unlink(NULL, p_server_id); + + return py_server_id; +} + +static PyGetSetDef py_imessaging_getset[] = { + { + .name = discard_const_p(char, "server_id"), + .get = py_imessaging_server_id, + .doc = discard_const_p(char, "local server id") + }, + { .name = NULL }, +}; + + +PyTypeObject imessaging_Type = { + PyVarObject_HEAD_INIT(NULL, 0) + .tp_name = "messaging.Messaging", + .tp_basicsize = sizeof(imessaging_Object), + .tp_flags = Py_TPFLAGS_DEFAULT|Py_TPFLAGS_BASETYPE, + .tp_new = py_imessaging_connect, + .tp_dealloc = py_imessaging_dealloc, + .tp_methods = py_imessaging_methods, + .tp_getset = py_imessaging_getset, + .tp_doc = "Messaging(own_id=None)\n" \ + "Create a new object that can be used to communicate with the peers in the specified messaging path.\n" +}; + +static struct PyModuleDef moduledef = { + PyModuleDef_HEAD_INIT, + .m_name = "messaging", + .m_doc = "Internal RPC", + .m_size = -1, + .m_methods = NULL, +}; + +MODULE_INIT_FUNC(messaging) +{ + PyObject *mod; + + if (PyType_Ready(&imessaging_Type) < 0) + return NULL; + + mod = PyModule_Create(&moduledef); + if (mod == NULL) + return NULL; + + Py_INCREF((PyObject *)&imessaging_Type); + PyModule_AddObject(mod, "Messaging", (PyObject *)&imessaging_Type); + PyModule_AddObject(mod, "IRPC_CALL_TIMEOUT", PyLong_FromLong(IRPC_CALL_TIMEOUT)); + PyModule_AddObject(mod, "IRPC_CALL_TIMEOUT_INF", PyLong_FromLong(IRPC_CALL_TIMEOUT_INF)); + + return mod; +} diff --git a/source4/lib/messaging/tests/irpc.c b/source4/lib/messaging/tests/irpc.c new file mode 100644 index 0000000..466b47f --- /dev/null +++ b/source4/lib/messaging/tests/irpc.c @@ -0,0 +1,308 @@ +/* + Unix SMB/CIFS implementation. + + local test for irpc code + + Copyright (C) Andrew Tridgell 2004 + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +#include "includes.h" +#include "lib/events/events.h" +#include "lib/messaging/irpc.h" +#include "librpc/gen_ndr/ndr_echo.h" +#include "librpc/gen_ndr/ndr_echo_c.h" +#include "torture/torture.h" +#include "cluster/cluster.h" +#include "param/param.h" +#include "torture/local/proto.h" + +const uint32_t MSG_ID1 = 1, MSG_ID2 = 2; + +static bool test_debug; + +struct irpc_test_data +{ + struct imessaging_context *msg_ctx1, *msg_ctx2; + struct tevent_context *ev; +}; + +/* + serve up AddOne over the irpc system +*/ +static NTSTATUS irpc_AddOne(struct irpc_message *irpc, struct echo_AddOne *r) +{ + *r->out.out_data = r->in.in_data + 1; + if (test_debug) { + printf("irpc_AddOne: in=%u in+1=%u out=%u\n", + r->in.in_data, r->in.in_data+1, *r->out.out_data); + } + return NT_STATUS_OK; +} + +/* + a deferred reply to echodata +*/ +static void deferred_echodata(struct tevent_context *ev, struct tevent_timer *te, + struct timeval t, void *private_data) +{ + struct irpc_message *irpc = talloc_get_type(private_data, struct irpc_message); + struct echo_EchoData *r = (struct echo_EchoData *)irpc->data; + r->out.out_data = (uint8_t *)talloc_memdup(r, r->in.in_data, r->in.len); + if (r->out.out_data == NULL) { + irpc_send_reply(irpc, NT_STATUS_NO_MEMORY); + } + printf("sending deferred reply\n"); + irpc_send_reply(irpc, NT_STATUS_OK); +} + + +/* + serve up EchoData over the irpc system +*/ +static NTSTATUS irpc_EchoData(struct irpc_message *irpc, struct echo_EchoData *r) +{ + struct irpc_test_data *data = talloc_get_type_abort(irpc->private_data, struct irpc_test_data); + irpc->defer_reply = true; + tevent_add_timer(data->ev, irpc, timeval_zero(), deferred_echodata, irpc); + return NT_STATUS_OK; +} + + +/* + test a addone call over the internal messaging system +*/ +static bool test_addone(struct torture_context *test, const void *_data, + const void *_value) +{ + struct echo_AddOne r; + NTSTATUS status; + const struct irpc_test_data *data = (const struct irpc_test_data *)_data; + uint32_t value = *(const uint32_t *)_value; + struct dcerpc_binding_handle *irpc_handle; + + irpc_handle = irpc_binding_handle(test, data->msg_ctx1, + cluster_id(0, MSG_ID2), + &ndr_table_rpcecho); + torture_assert(test, irpc_handle, "no memory"); + + /* make the call */ + r.in.in_data = value; + + test_debug = true; + /* + * Note: this makes use of nested event loops + * as client and server use the same loop. + */ + dcerpc_binding_handle_set_sync_ev(irpc_handle, data->ev); + status = dcerpc_echo_AddOne_r(irpc_handle, test, &r); + test_debug = false; + torture_assert_ntstatus_ok(test, status, "AddOne failed"); + + /* check the answer */ + torture_assert(test, *r.out.out_data == r.in.in_data + 1, + "AddOne wrong answer"); + + torture_comment(test, "%u + 1 = %u\n", r.in.in_data, *r.out.out_data); + return true; +} + +/* + test a echodata call over the internal messaging system +*/ +static bool test_echodata(struct torture_context *tctx, + const void *tcase_data, + const void *test_data) +{ + struct echo_EchoData r; + NTSTATUS status; + const struct irpc_test_data *data = (const struct irpc_test_data *)tcase_data; + TALLOC_CTX *mem_ctx = tctx; + struct dcerpc_binding_handle *irpc_handle; + + irpc_handle = irpc_binding_handle(mem_ctx, data->msg_ctx1, + cluster_id(0, MSG_ID2), + &ndr_table_rpcecho); + torture_assert(tctx, irpc_handle, "no memory"); + + /* make the call */ + r.in.in_data = (unsigned char *)talloc_strdup(mem_ctx, "0123456789"); + r.in.len = strlen((char *)r.in.in_data); + + /* + * Note: this makes use of nested event loops + * as client and server use the same loop. + */ + dcerpc_binding_handle_set_sync_ev(irpc_handle, data->ev); + status = dcerpc_echo_EchoData_r(irpc_handle, mem_ctx, &r); + torture_assert_ntstatus_ok(tctx, status, "EchoData failed"); + + /* check the answer */ + if (memcmp(r.out.out_data, r.in.in_data, r.in.len) != 0) { + NDR_PRINT_OUT_DEBUG(echo_EchoData, &r); + torture_fail(tctx, "EchoData wrong answer"); + } + + torture_comment(tctx, "Echo '%*.*s' -> '%*.*s'\n", + r.in.len, r.in.len, + r.in.in_data, + r.in.len, r.in.len, + r.out.out_data); + return true; +} + +struct irpc_callback_state { + struct echo_AddOne r; + int *pong_count; +}; + +static void irpc_callback(struct tevent_req *subreq) +{ + struct irpc_callback_state *s = + tevent_req_callback_data(subreq, + struct irpc_callback_state); + NTSTATUS status; + + status = dcerpc_echo_AddOne_r_recv(subreq, s); + TALLOC_FREE(subreq); + if (!NT_STATUS_IS_OK(status)) { + printf("irpc call failed - %s\n", nt_errstr(status)); + } + if (*s->r.out.out_data != s->r.in.in_data + 1) { + printf("AddOne wrong answer - %u + 1 = %u should be %u\n", + s->r.in.in_data, *s->r.out.out_data, s->r.in.in_data+1); + } + (*s->pong_count)++; +} + +/* + test echo speed +*/ +static bool test_speed(struct torture_context *tctx, + const void *tcase_data, + const void *test_data) +{ + int ping_count = 0; + int pong_count = 0; + const struct irpc_test_data *data = (const struct irpc_test_data *)tcase_data; + struct timeval tv; + TALLOC_CTX *mem_ctx = tctx; + int timelimit = torture_setting_int(tctx, "timelimit", 10); + struct dcerpc_binding_handle *irpc_handle; + + irpc_handle = irpc_binding_handle(mem_ctx, data->msg_ctx1, + cluster_id(0, MSG_ID2), + &ndr_table_rpcecho); + torture_assert(tctx, irpc_handle, "no memory"); + + tv = timeval_current(); + + torture_comment(tctx, "Sending echo for %d seconds\n", timelimit); + while (timeval_elapsed(&tv) < timelimit) { + struct tevent_req *subreq; + struct irpc_callback_state *s; + + s = talloc_zero(mem_ctx, struct irpc_callback_state); + torture_assert(tctx, s != NULL, "no mem"); + + s->pong_count = &pong_count; + + subreq = dcerpc_echo_AddOne_r_send(mem_ctx, + tctx->ev, + irpc_handle, + &s->r); + torture_assert(tctx, subreq != NULL, "AddOne send failed"); + + tevent_req_set_callback(subreq, irpc_callback, s); + + ping_count++; + + while (ping_count > pong_count + 20) { + tevent_loop_once(data->ev); + } + } + + torture_comment(tctx, "waiting for %d remaining replies (done %d)\n", + ping_count - pong_count, pong_count); + while (timeval_elapsed(&tv) < 30 && pong_count < ping_count) { + tevent_loop_once(data->ev); + } + + torture_assert_int_equal(tctx, ping_count, pong_count, "ping test failed"); + + torture_comment(tctx, "echo rate of %.0f messages/sec\n", + (ping_count+pong_count)/timeval_elapsed(&tv)); + return true; +} + + +static bool irpc_setup(struct torture_context *tctx, void **_data) +{ + struct irpc_test_data *data; + + *_data = data = talloc(tctx, struct irpc_test_data); + + lpcfg_set_cmdline(tctx->lp_ctx, "pid directory", "piddir.tmp"); + + data->ev = tctx->ev; + torture_assert(tctx, data->msg_ctx1 = + imessaging_init(tctx, + tctx->lp_ctx, + cluster_id(0, MSG_ID1), + data->ev), + "Failed to init first messaging context"); + + torture_assert(tctx, data->msg_ctx2 = + imessaging_init(tctx, + tctx->lp_ctx, + cluster_id(0, MSG_ID2), + data->ev), + "Failed to init second messaging context"); + + /* register the server side function */ + IRPC_REGISTER(data->msg_ctx1, rpcecho, ECHO_ADDONE, irpc_AddOne, data); + IRPC_REGISTER(data->msg_ctx2, rpcecho, ECHO_ADDONE, irpc_AddOne, data); + + IRPC_REGISTER(data->msg_ctx1, rpcecho, ECHO_ECHODATA, irpc_EchoData, data); + IRPC_REGISTER(data->msg_ctx2, rpcecho, ECHO_ECHODATA, irpc_EchoData, data); + + return true; +} + +struct torture_suite *torture_local_irpc(TALLOC_CTX *mem_ctx) +{ + struct torture_suite *suite = torture_suite_create(mem_ctx, "irpc"); + struct torture_tcase *tcase = torture_suite_add_tcase(suite, "irpc"); + int i; + uint32_t *values = talloc_array(tcase, uint32_t, 5); + + values[0] = 0; + values[1] = 0x7FFFFFFE; + values[2] = 0xFFFFFFFE; + values[3] = 0xFFFFFFFF; + values[4] = random() & 0xFFFFFFFF; + + tcase->setup = irpc_setup; + + for (i = 0; i < 5; i++) { + torture_tcase_add_test_const(tcase, "addone", test_addone, + (void *)&values[i]); + } + + torture_tcase_add_test_const(tcase, "echodata", test_echodata, NULL); + torture_tcase_add_test_const(tcase, "speed", test_speed, NULL); + + return suite; +} diff --git a/source4/lib/messaging/tests/messaging.c b/source4/lib/messaging/tests/messaging.c new file mode 100644 index 0000000..dcbbc19 --- /dev/null +++ b/source4/lib/messaging/tests/messaging.c @@ -0,0 +1,694 @@ +/* + Unix SMB/CIFS implementation. + + local test for messaging code + + Copyright (C) Andrew Tridgell 2004 + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +#include "includes.h" +#include "lib/events/events.h" +#include "lib/messaging/irpc.h" +#include "torture/torture.h" +#include "cluster/cluster.h" +#include "param/param.h" +#include "torture/local/proto.h" +#include "system/select.h" +#include "system/filesys.h" + +#include <gnutls/gnutls.h> +#include <gnutls/crypto.h> + +static uint32_t msg_pong; + +static void ping_message(struct imessaging_context *msg, + void *private_data, + uint32_t msg_type, + struct server_id src, + size_t num_fds, + int *fds, + DATA_BLOB *data) +{ + NTSTATUS status; + + if (num_fds != 0) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + status = imessaging_send(msg, src, msg_pong, data); + if (!NT_STATUS_IS_OK(status)) { + printf("pong failed - %s\n", nt_errstr(status)); + } +} + +static void pong_message(struct imessaging_context *msg, + void *private_data, + uint32_t msg_type, + struct server_id src, + size_t num_fds, + int *fds, + DATA_BLOB *data) +{ + int *count = (int *)private_data; + + if (num_fds != 0) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + (*count)++; +} + +static void exit_message(struct imessaging_context *msg, + void *private_data, + uint32_t msg_type, + struct server_id src, + size_t num_fds, + int *fds, + DATA_BLOB *data) +{ + if (num_fds != 0) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + talloc_free(private_data); + exit(0); +} + +/* + test ping speed +*/ +static bool test_ping_speed(struct torture_context *tctx) +{ + struct tevent_context *ev; + struct imessaging_context *msg_client_ctx; + struct imessaging_context *msg_server_ctx; + int ping_count = 0; + int pong_count = 0; + struct timeval tv; + int timelimit = torture_setting_int(tctx, "timelimit", 10); + uint32_t msg_ping, msg_exit; + + lpcfg_set_cmdline(tctx->lp_ctx, "pid directory", "piddir.tmp"); + + ev = tctx->ev; + + msg_server_ctx = imessaging_init(tctx, + tctx->lp_ctx, cluster_id(0, 1), + ev); + + torture_assert(tctx, msg_server_ctx != NULL, "Failed to init ping messaging context"); + + imessaging_register_tmp(msg_server_ctx, NULL, ping_message, &msg_ping); + imessaging_register_tmp(msg_server_ctx, tctx, exit_message, &msg_exit); + + msg_client_ctx = imessaging_init(tctx, + tctx->lp_ctx, + cluster_id(0, 2), + ev); + + torture_assert(tctx, msg_client_ctx != NULL, + "msg_client_ctx imessaging_init() failed"); + + imessaging_register_tmp(msg_client_ctx, &pong_count, pong_message, &msg_pong); + + tv = timeval_current(); + + torture_comment(tctx, "Sending pings for %d seconds\n", timelimit); + while (timeval_elapsed(&tv) < timelimit) { + DATA_BLOB data; + NTSTATUS status1, status2; + + data.data = discard_const_p(uint8_t, "testing"); + data.length = strlen((const char *)data.data); + + status1 = imessaging_send(msg_client_ctx, cluster_id(0, 1), msg_ping, &data); + status2 = imessaging_send(msg_client_ctx, cluster_id(0, 1), msg_ping, NULL); + + torture_assert_ntstatus_ok(tctx, status1, "msg1 failed"); + ping_count++; + + torture_assert_ntstatus_ok(tctx, status2, "msg2 failed"); + ping_count++; + + while (ping_count > pong_count + 20) { + tevent_loop_once(ev); + } + } + + torture_comment(tctx, "waiting for %d remaining replies (done %d)\n", + ping_count - pong_count, pong_count); + while (timeval_elapsed(&tv) < 30 && pong_count < ping_count) { + tevent_loop_once(ev); + } + + torture_comment(tctx, "sending exit\n"); + imessaging_send(msg_client_ctx, cluster_id(0, 1), msg_exit, NULL); + + torture_assert_int_equal(tctx, ping_count, pong_count, "ping test failed"); + + torture_comment(tctx, "ping rate of %.0f messages/sec\n", + (ping_count+pong_count)/timeval_elapsed(&tv)); + + talloc_free(msg_client_ctx); + talloc_free(msg_server_ctx); + + return true; +} + +static bool test_messaging_overflow(struct torture_context *tctx) +{ + struct imessaging_context *msg_ctx; + ssize_t nwritten, nread; + pid_t child; + char c = 0; + int up_pipe[2], down_pipe[2]; + int i, ret, child_status; + + ret = pipe(up_pipe); + torture_assert(tctx, ret == 0, "pipe failed"); + ret = pipe(down_pipe); + torture_assert(tctx, ret == 0, "pipe failed"); + + child = fork(); + if (child < 0) { + torture_fail(tctx, "fork failed"); + } + + if (child == 0) { + ret = tevent_re_initialise(tctx->ev); + torture_assert(tctx, ret == 0, "tevent_re_initialise failed"); + + msg_ctx = imessaging_init(tctx, tctx->lp_ctx, + cluster_id(getpid(), 0), + tctx->ev); + torture_assert(tctx, msg_ctx != NULL, + "imessaging_init failed"); + + do { + nwritten = write(up_pipe[1], &c, 1); + } while ((nwritten == -1) && (errno == EINTR)); + + ret = close(down_pipe[1]); + torture_assert(tctx, ret == 0, "close failed"); + + do { + nread = read(down_pipe[0], &c, 1); + } while ((nread == -1) && (errno == EINTR)); + + exit(0); + } + + do { + nread = read(up_pipe[0], &c, 1); + } while ((nread == -1) && (errno == EINTR)); + + msg_ctx = imessaging_init(tctx, tctx->lp_ctx, cluster_id(getpid(), 0), + tctx->ev); + torture_assert(tctx, msg_ctx != NULL, "imessaging_init failed"); + + for (i=0; i<1000; i++) { + NTSTATUS status; + status = imessaging_send(msg_ctx, cluster_id(child, 0), + MSG_PING, NULL); + torture_assert_ntstatus_ok(tctx, status, + "imessaging_send failed"); + } + + tevent_loop_once(tctx->ev); + + talloc_free(msg_ctx); + + ret = close(down_pipe[1]); + torture_assert(tctx, ret == 0, "close failed"); + + ret = waitpid(child, &child_status, 0); + torture_assert(tctx, ret == child, "wrong child exited"); + torture_assert(tctx, child_status == 0, "child failed"); + + poll(NULL, 0, 500); + + return true; +} + +struct overflow_parent_child { + gnutls_hash_hd_t md5_hash_hnd; + bool done; +}; + +static void overflow_md5_child_handler(struct imessaging_context *msg, + void *private_data, + uint32_t msg_type, + struct server_id server_id, + size_t num_fds, + int *fds, + DATA_BLOB *data) +{ + struct overflow_parent_child *state = private_data; + + if (num_fds != 0) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + if (data->length == 0) { + state->done = true; + return; + } + + gnutls_hash(state->md5_hash_hnd, data->data, data->length); +} + +struct overflow_child_parent { + uint8_t final[16]; + bool done; +}; + +static void overflow_md5_parent_handler(struct imessaging_context *msg_ctx, + void *private_data, + uint32_t msg_type, + struct server_id server_id, + size_t num_fds, + int *fds, + DATA_BLOB *data) +{ + struct overflow_child_parent *state = private_data; + + if (num_fds != 0) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + if (data->length != sizeof(state->final)) { + memset(state->final, 0, sizeof(state->final)); + state->done = true; + return; + } + memcpy(state->final, data->data, 16); + state->done = true; +} + +static bool test_messaging_overflow_check(struct torture_context *tctx) +{ + struct imessaging_context *msg_ctx; + ssize_t nwritten, nread; + pid_t child; + char c = 0; + int up_pipe[2], down_pipe[2]; + int i, ret, child_status; + gnutls_hash_hd_t hash_hnd; + uint8_t final[16]; + struct overflow_child_parent child_msg = { .done = false }; + NTSTATUS status; + + ret = pipe(up_pipe); + torture_assert(tctx, ret == 0, "pipe failed"); + ret = pipe(down_pipe); + torture_assert(tctx, ret == 0, "pipe failed"); + + child = fork(); + if (child < 0) { + torture_fail(tctx, "fork failed"); + } + + if (child == 0) { + struct overflow_parent_child child_state = { .done = false }; + DATA_BLOB retblob = { .data = final, .length = sizeof(final) }; + + ret = tevent_re_initialise(tctx->ev); + torture_assert(tctx, ret == 0, "tevent_re_initialise failed"); + + gnutls_hash_init(&child_state.md5_hash_hnd, GNUTLS_DIG_MD5); + + msg_ctx = imessaging_init(tctx, tctx->lp_ctx, + cluster_id(getpid(), 0), + tctx->ev); + torture_assert(tctx, msg_ctx != NULL, + "imessaging_init failed"); + + status = imessaging_register(msg_ctx, &child_state, + MSG_TMP_BASE-1, + overflow_md5_child_handler); + torture_assert(tctx, NT_STATUS_IS_OK(status), + "imessaging_register failed"); + + do { + nwritten = write(up_pipe[1], &c, 1); + } while ((nwritten == -1) && (errno == EINTR)); + + ret = close(down_pipe[1]); + torture_assert(tctx, ret == 0, "close failed"); + + do { + nread = read(down_pipe[0], &c, 1); + } while ((nread == -1) && (errno == EINTR)); + + while (!child_state.done) { + tevent_loop_once(tctx->ev); + } + + gnutls_hash_deinit(child_state.md5_hash_hnd, final); + + status = imessaging_send(msg_ctx, + cluster_id(getppid(), 0), + MSG_TMP_BASE-2, + &retblob); + torture_assert(tctx, NT_STATUS_IS_OK(status), + "imessaging_send failed"); + + exit(0); + } + + do { + nread = read(up_pipe[0], &c, 1); + } while ((nread == -1) && (errno == EINTR)); + + msg_ctx = imessaging_init(tctx, tctx->lp_ctx, cluster_id(getpid(), 0), + tctx->ev); + torture_assert(tctx, msg_ctx != NULL, "imessaging_init failed"); + + status = imessaging_register(msg_ctx, + &child_msg, + MSG_TMP_BASE-2, + overflow_md5_parent_handler); + torture_assert(tctx, + NT_STATUS_IS_OK(status), + "imessaging_register failed"); + + gnutls_hash_init(&hash_hnd, GNUTLS_DIG_MD5); + + for (i=0; i<1000; i++) { + size_t len = ((random() % 100) + 1); + uint8_t buf[len]; + DATA_BLOB blob = { .data = buf, .length = len }; + + generate_random_buffer(buf, len); + + gnutls_hash(hash_hnd, buf, len); + + status = imessaging_send(msg_ctx, cluster_id(child, 0), + MSG_TMP_BASE-1, &blob); + torture_assert_ntstatus_ok(tctx, status, + "imessaging_send failed"); + } + + status = imessaging_send(msg_ctx, cluster_id(child, 0), + MSG_TMP_BASE-1, NULL); + torture_assert_ntstatus_ok(tctx, status, + "imessaging_send failed"); + + gnutls_hash_deinit(hash_hnd, final); + + do { + nwritten = write(down_pipe[1], &c, 1); + } while ((nwritten == -1) && (errno == EINTR)); + + while (!child_msg.done) { + tevent_loop_once(tctx->ev); + } + + ret = close(down_pipe[1]); + torture_assert(tctx, ret == 0, "close failed"); + + talloc_free(msg_ctx); + + ret = waitpid(child, &child_status, 0); + torture_assert(tctx, ret == child, "wrong child exited"); + torture_assert(tctx, child_status == 0, "child failed"); + + if (memcmp(final, child_msg.final, 16) != 0) { + dump_data_file(final, 16, false, stderr); + dump_data_file(child_msg.final, 16, false, stderr); + fflush(stderr); + torture_fail(tctx, "checksum comparison failed"); + } + + return true; +} + +struct test_multi_ctx { + struct torture_context *tctx; + struct imessaging_context *server_ctx; + struct imessaging_context *client_ctx[4]; + size_t num_missing; + bool got_server; + bool got_client_0_1; + bool got_client_2_3; + bool ok; +}; + +static void multi_ctx_server_handler(struct imessaging_context *msg, + void *private_data, + uint32_t msg_type, + struct server_id server_id, + size_t num_fds, + int *fds, + DATA_BLOB *data) +{ + struct test_multi_ctx *state = private_data; + char *str = NULL; + + if (num_fds != 0) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + torture_assert_goto(state->tctx, state->num_missing >= 1, + state->ok, fail, + "num_missing should be at least 1."); + state->num_missing -= 1; + + torture_assert_goto(state->tctx, !state->got_server, + state->ok, fail, + "already got server."); + state->got_server = true; + + /* + * We free the context itself and most likely reuse + * the memory immediately. + */ + TALLOC_FREE(state->server_ctx); + str = generate_random_str(state->tctx, 128); + torture_assert_goto(state->tctx, str != NULL, + state->ok, fail, + "generate_random_str()"); + +fail: + return; +} + +static void multi_ctx_client_0_1_handler(struct imessaging_context *msg, + void *private_data, + uint32_t msg_type, + struct server_id server_id, + size_t num_fds, + int *fds, + DATA_BLOB *data) +{ + struct test_multi_ctx *state = private_data; + char *str = NULL; + + if (num_fds != 0) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + torture_assert_goto(state->tctx, state->num_missing >= 2, + state->ok, fail, + "num_missing should be at least 2."); + state->num_missing -= 2; + + torture_assert_goto(state->tctx, !state->got_client_0_1, + state->ok, fail, + "already got client_0_1."); + state->got_client_0_1 = true; + + /* + * We free two contexts and most likely reuse + * the memory immediately. + */ + TALLOC_FREE(state->client_ctx[0]); + str = generate_random_str(state->tctx, 128); + torture_assert_goto(state->tctx, str != NULL, + state->ok, fail, + "generate_random_str()"); + TALLOC_FREE(state->client_ctx[1]); + str = generate_random_str(state->tctx, 128); + torture_assert_goto(state->tctx, str != NULL, + state->ok, fail, + "generate_random_str()"); + +fail: + return; +} + +static void multi_ctx_client_2_3_handler(struct imessaging_context *msg, + void *private_data, + uint32_t msg_type, + struct server_id server_id, + size_t num_fds, + int *fds, + DATA_BLOB *data) +{ + struct test_multi_ctx *state = private_data; + char *str = NULL; + + if (num_fds != 0) { + DBG_WARNING("Received %zu fds, ignoring message\n", num_fds); + return; + } + + torture_assert_goto(state->tctx, state->num_missing >= 2, + state->ok, fail, + "num_missing should be at least 2."); + state->num_missing -= 2; + + torture_assert_goto(state->tctx, !state->got_client_2_3, + state->ok, fail, + "already got client_2_3."); + state->got_client_2_3 = true; + + /* + * We free two contexts and most likely reuse + * the memory immediately. + */ + TALLOC_FREE(state->client_ctx[2]); + str = generate_random_str(state->tctx, 128); + torture_assert_goto(state->tctx, str != NULL, + state->ok, fail, + "generate_random_str()"); + TALLOC_FREE(state->client_ctx[3]); + str = generate_random_str(state->tctx, 128); + torture_assert_goto(state->tctx, str != NULL, + state->ok, fail, + "generate_random_str()"); + +fail: + return; +} + +static bool test_multi_ctx(struct torture_context *tctx) +{ + struct test_multi_ctx state = { + .tctx = tctx, + .ok = true, + }; + struct timeval tv; + NTSTATUS status; + + lpcfg_set_cmdline(tctx->lp_ctx, "pid directory", "piddir.tmp"); + + /* + * We use cluster_id(0, 0) as that gets for + * all task ids. + */ + state.server_ctx = imessaging_init(tctx, + tctx->lp_ctx, + cluster_id(0, 0), + tctx->ev); + torture_assert(tctx, state.server_ctx != NULL, + "Failed to init messaging context"); + + status = imessaging_register(state.server_ctx, &state, + MSG_TMP_BASE-1, + multi_ctx_server_handler); + torture_assert(tctx, NT_STATUS_IS_OK(status), "imessaging_register failed"); + + state.client_ctx[0] = imessaging_init(tctx, + tctx->lp_ctx, + cluster_id(0, 0), + tctx->ev); + torture_assert(tctx, state.client_ctx[0] != NULL, + "msg_client_ctx imessaging_init() failed"); + status = imessaging_register(state.client_ctx[0], &state, + MSG_TMP_BASE-1, + multi_ctx_client_0_1_handler); + torture_assert(tctx, NT_STATUS_IS_OK(status), "imessaging_register failed"); + state.client_ctx[1] = imessaging_init(tctx, + tctx->lp_ctx, + cluster_id(0, 0), + tctx->ev); + torture_assert(tctx, state.client_ctx[1] != NULL, + "msg_client_ctx imessaging_init() failed"); + status = imessaging_register(state.client_ctx[1], &state, + MSG_TMP_BASE-1, + multi_ctx_client_0_1_handler); + torture_assert(tctx, NT_STATUS_IS_OK(status), "imessaging_register failed"); + state.client_ctx[2] = imessaging_init(tctx, + tctx->lp_ctx, + cluster_id(0, 0), + tctx->ev); + torture_assert(tctx, state.client_ctx[2] != NULL, + "msg_client_ctx imessaging_init() failed"); + status = imessaging_register(state.client_ctx[2], &state, + MSG_TMP_BASE-1, + multi_ctx_client_2_3_handler); + torture_assert(tctx, NT_STATUS_IS_OK(status), "imessaging_register failed"); + state.client_ctx[3] = imessaging_init(tctx, + tctx->lp_ctx, + cluster_id(0, 0), + tctx->ev); + torture_assert(tctx, state.client_ctx[3] != NULL, + "msg_client_ctx imessaging_init() failed"); + status = imessaging_register(state.client_ctx[3], &state, + MSG_TMP_BASE-1, + multi_ctx_client_2_3_handler); + torture_assert(tctx, NT_STATUS_IS_OK(status), "imessaging_register failed"); + + /* + * Send one message that need to arrive on 3 ( 5 - 2 ) handlers. + */ + state.num_missing = 5; + + status = imessaging_send(state.server_ctx, + cluster_id(0, 0), + MSG_TMP_BASE-1, NULL); + torture_assert_ntstatus_ok(tctx, status, "msg failed"); + + tv = timeval_current(); + while (timeval_elapsed(&tv) < 30 && state.num_missing > 0 && state.ok) { + int ret; + + ret = tevent_loop_once(tctx->ev); + torture_assert_int_equal(tctx, ret, 0, "tevent_loop_once()"); + } + + if (!state.ok) { + return false; + } + + torture_assert_int_equal(tctx, state.num_missing, 0, + "wrong message count"); + + torture_assert(tctx, state.got_client_0_1, "got_client_0_1"); + torture_assert(tctx, state.got_client_2_3, "got_client_2_3"); + torture_assert(tctx, state.got_server, "got_server"); + + return true; +} + +struct torture_suite *torture_local_messaging(TALLOC_CTX *mem_ctx) +{ + struct torture_suite *s = torture_suite_create(mem_ctx, "messaging"); + torture_suite_add_simple_test(s, "overflow", test_messaging_overflow); + torture_suite_add_simple_test(s, "overflow_check", + test_messaging_overflow_check); + torture_suite_add_simple_test(s, "ping_speed", test_ping_speed); + torture_suite_add_simple_test(s, "multi_ctx", test_multi_ctx); + return s; +} diff --git a/source4/lib/messaging/wscript_build b/source4/lib/messaging/wscript_build new file mode 100644 index 0000000..3408396 --- /dev/null +++ b/source4/lib/messaging/wscript_build @@ -0,0 +1,33 @@ +#!/usr/bin/env python + + +bld.SAMBA_LIBRARY('MESSAGING_SEND', + source='messaging_send.c', + public_deps='messages_util messages_dgm UNIX_PRIVS cluster server_id_db', + private_library=True + ) + +bld.SAMBA_LIBRARY('MESSAGING', + source='messaging.c messaging_handlers.c', + public_deps=''' + samba-util + NDR_IRPC + UNIX_PRIVS + cluster + ndr + dcerpc + messages_util + server_id_db + talloc_report_printf + ''', + private_library=True + ) + +pyparam_util = bld.pyembed_libname('pyparam_util') +pytalloc_util = bld.pyembed_libname('pytalloc-util') + +bld.SAMBA_PYTHON('python_messaging', + source='pymessaging.c', + deps='MESSAGING events %s %s' % (pyparam_util, pytalloc_util), + realname='samba/messaging.so' + ) |