Changeset 745 for trunk/server/source4/lib/messaging/messaging.c

Timestamp:

Nov 27, 2012, 4:43:17 PM (13 years ago)

Author:

Silvan Scherrer

Message:

Samba Server: updated trunk to 3.6.0

Location:

trunk/server

Files:

• . (modified) (1 prop)
• source4/lib/messaging/messaging.c (modified) (20 diffs)

trunk/server/source4/lib/messaging/messaging.c

@@ -28 +28 @@
 #include "librpc/gen_ndr/ndr_irpc.h"
 #include "lib/messaging/irpc.h"
-#include "tdb_wrap.h"
+#include "lib/util/tdb_wrap.h"
 #include "../lib/util/unix_privs.h"
 #include "librpc/rpc/dcerpc.h"
-#include "../tdb/include/tdb.h"
+#include <tdb.h>
 #include "../lib/util/util_tdb.h"
 #include "cluster/cluster.h"
+#include "../lib/util/tevent_ntstatus.h"
 
 /* change the message version with any incompatible changes in the protocol */
 #define MESSAGING_VERSION 1
+
+/*
+  a pending irpc call
+*/
+struct irpc_request {
+        struct messaging_context *msg_ctx;
+        int callid;
+        struct {
+                void (*handler)(struct irpc_request *irpc, struct irpc_message *m);
+                void *private_data;
+        } incoming;
+};
 
 struct messaging_context {
@@ -48 +61 @@
         struct messaging_rec *pending;
         struct messaging_rec *retry_queue;
-        struct smb_iconv_convenience *iconv_convenience;
         struct irpc_list *irpc;
         struct idr_context *idr;
@@ -99 +111 @@
 {
         DEBUG(1,("INFO: Received PING message from server %u.%u [%.*s]\n",
-                 (uint_t)src.node, (uint_t)src.id, (int)data->length, 
+                 (unsigned int)src.node, (unsigned int)src.id, (int)data->length,
                  data->data?(const char *)data->data:""));
         messaging_send(msg, src, MSG_PONG, data);
@@ -120 +132 @@
 static char *messaging_path(struct messaging_context *msg, struct server_id server_id)
 {
-        return talloc_asprintf(msg, "%s/msg.%s", msg->base_path, 
-                               cluster_id_string(msg, server_id));
+        TALLOC_CTX *tmp_ctx = talloc_new(msg);
+        const char *id = cluster_id_string(tmp_ctx, server_id);
+        char *s;
+        if (id == NULL) {
+                return NULL;
+        }
+        s = talloc_asprintf(msg, "%s/msg.%s", msg->base_path, id);
+        talloc_steal(s, tmp_ctx);
+        return s;
 }
 
@@ -262 +281 @@
                 rec->retries = 0;
                 if (!NT_STATUS_IS_OK(status)) {
+                        TALLOC_CTX *tmp_ctx = talloc_new(msg);
                         DEBUG(1,("messaging: Lost message from %s to %s of type %u - %s\n", 
-                                 cluster_id_string(debug_ctx(), rec->header->from), 
-                                 cluster_id_string(debug_ctx(), rec->header->to), 
+                                 cluster_id_string(tmp_ctx, rec->header->from),
+                                 cluster_id_string(tmp_ctx, rec->header->to),
                                  rec->header->msg_type, 
                                  nt_errstr(status)));
+                        talloc_free(tmp_ctx);
                 }
                 DLIST_REMOVE(msg->pending, rec);
@@ -441 +462 @@
 */
 NTSTATUS messaging_send(struct messaging_context *msg, struct server_id server, 
-                        uint32_t msg_type, DATA_BLOB *data)
+                        uint32_t msg_type, const DATA_BLOB *data)
 {
         struct messaging_rec *rec;
@@ -536 +557 @@
                                          const char *dir,
                                          struct server_id server_id, 
-                                         struct smb_iconv_convenience *iconv_convenience,
                                          struct tevent_context *ev)
 {
@@ -565 +585 @@
         msg->path          = messaging_path(msg, server_id);
         msg->server_id     = server_id;
-        msg->iconv_convenience = iconv_convenience;
         msg->idr           = idr_init(msg);
         msg->dispatch_tree = idr_init(msg);
@@ -600 +619 @@
         msg->event.fde  = event_add_fd(ev, msg, socket_get_fd(msg->sock), 
                                        EVENT_FD_READ, messaging_handler, msg);
+        tevent_fd_set_auto_close(msg->event.fde);
 
         talloc_set_destructor(msg, messaging_destructor);
@@ -615 +635 @@
 struct messaging_context *messaging_client_init(TALLOC_CTX *mem_ctx, 
                                                 const char *dir,
-                                                struct smb_iconv_convenience *iconv_convenience,
                                                 struct tevent_context *ev)
 {
@@ -621 +640 @@
         ZERO_STRUCT(id);
         id.id = random() % 0x10000000;
-        return messaging_init(mem_ctx, dir, id, iconv_convenience, ev);
+        return messaging_init(mem_ctx, dir, id, ev);
 }
 /*
@@ -673 +692 @@
 {
         struct irpc_request *irpc;
-        enum ndr_err_code ndr_err;
 
         irpc = (struct irpc_request *)idr_find(msg_ctx->idr, m->header.callid);
         if (irpc == NULL) return;
 
-        /* parse the reply data */
-        ndr_err = irpc->table->calls[irpc->callnum].ndr_pull(m->ndr, NDR_OUT, irpc->r);
-        if (NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
-                irpc->status = m->header.status;
-                talloc_steal(irpc->mem_ctx, m);
-        } else {
-                irpc->status = ndr_map_error2ntstatus(ndr_err);
-                talloc_steal(irpc, m);
-        }
-        irpc->done = true;
-        if (irpc->async.fn) {
-                irpc->async.fn(irpc);
-        }
+        irpc->incoming.handler(irpc, m);
 }
 
@@ -705 +711 @@
 
         /* setup the reply */
-        push = ndr_push_init_ctx(m->ndr, m->msg_ctx->iconv_convenience);
+        push = ndr_push_init_ctx(m->ndr);
         if (push == NULL) {
                 status = NT_STATUS_NO_MEMORY;
@@ -712 +718 @@
 
         m->header.flags |= IRPC_FLAG_REPLY;
+        m->header.creds.token= NULL;
 
         /* construct the packet */
@@ -764 +771 @@
         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);
@@ -771 +780 @@
         m->private_data= i->private_data;
         m->defer_reply = false;
+        m->no_reply    = false;
         m->msg_ctx     = msg_ctx;
         m->irpc        = i;
@@ -778 +788 @@
         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 */
@@ -805 +821 @@
         m->from = src;
 
-        m->ndr = ndr_pull_init_blob(packet, m, msg_ctx->iconv_convenience);
+        m->ndr = ndr_pull_init_blob(packet, m);
         if (m->ndr == NULL) goto failed;
 
@@ -835 +851 @@
         }
 
-        if (irpc->reject_free) {
-                return -1;
-        }
         return 0;
-}
-
-/*
-  timeout a irpc request
-*/
-static void irpc_timeout(struct tevent_context *ev, struct tevent_timer *te, 
-                         struct timeval t, void *private_data)
-{
-        struct irpc_request *irpc = talloc_get_type(private_data, struct irpc_request);
-        irpc->status = NT_STATUS_IO_TIMEOUT;
-        irpc->done = true;
-        if (irpc->async.fn) {
-                irpc->async.fn(irpc);
-        }
-}
-
-
-/*
-  make a irpc call - async send
-*/
-struct irpc_request *irpc_call_send(struct messaging_context *msg_ctx, 
-                                    struct server_id server_id, 
-                                    const struct ndr_interface_table *table, 
-                                    int callnum, void *r, TALLOC_CTX *ctx)
-{
-        struct irpc_header header;
-        struct ndr_push *ndr;
-        NTSTATUS status;
-        DATA_BLOB packet;
-        struct irpc_request *irpc;
-        enum ndr_err_code ndr_err;
-
-        irpc = talloc(msg_ctx, struct irpc_request);
-        if (irpc == NULL) goto failed;
-
-        irpc->msg_ctx  = msg_ctx;
-        irpc->table    = table;
-        irpc->callnum  = callnum;
-        irpc->callid   = idr_get_new(msg_ctx->idr, irpc, UINT16_MAX);
-        if (irpc->callid == -1) goto failed;
-        irpc->r        = r;
-        irpc->done     = false;
-        irpc->async.fn = NULL;
-        irpc->mem_ctx  = ctx;
-        irpc->reject_free = false;
-
-        talloc_set_destructor(irpc, irpc_destructor);
-
-        /* setup the header */
-        header.uuid = table->syntax_id.uuid;
-
-        header.if_version = table->syntax_id.if_version;
-        header.callid     = irpc->callid;
-        header.callnum    = callnum;
-        header.flags      = 0;
-        header.status     = NT_STATUS_OK;
-
-        /* construct the irpc packet */
-        ndr = ndr_push_init_ctx(irpc, msg_ctx->iconv_convenience);
-        if (ndr == NULL) goto failed;
-
-        ndr_err = ndr_push_irpc_header(ndr, NDR_SCALARS|NDR_BUFFERS, &header);
-        if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) goto failed;
-
-        ndr_err = table->calls[callnum].ndr_push(ndr, NDR_IN, r);
-        if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) goto failed;
-
-        /* and send it */
-        packet = ndr_push_blob(ndr);
-        status = messaging_send(msg_ctx, server_id, MSG_IRPC, &packet);
-        if (!NT_STATUS_IS_OK(status)) goto failed;
-
-        event_add_timed(msg_ctx->event.ev, irpc, 
-                        timeval_current_ofs(IRPC_CALL_TIMEOUT, 0), 
-                        irpc_timeout, irpc);
-
-        talloc_free(ndr);
-        return irpc;
-
-failed:
-        talloc_free(irpc);
-        return NULL;
-}
-
-/*
-  wait for a irpc reply
-*/
-NTSTATUS irpc_call_recv(struct irpc_request *irpc)
-{
-        NTSTATUS status;
-
-        NT_STATUS_HAVE_NO_MEMORY(irpc);
-
-        irpc->reject_free = true;
-
-        while (!irpc->done) {
-                if (event_loop_once(irpc->msg_ctx->event.ev) != 0) {
-                        return NT_STATUS_CONNECTION_DISCONNECTED;
-                }
-        }
-
-        irpc->reject_free = false;
-
-        status = irpc->status;
-        talloc_free(irpc);
-        return status;
-}
-
-/*
-  perform a synchronous irpc request
-*/
-NTSTATUS irpc_call(struct messaging_context *msg_ctx, 
-                   struct server_id server_id, 
-                   const struct ndr_interface_table *table, 
-                   int callnum, void *r,
-                   TALLOC_CTX *mem_ctx)
-{
-        struct irpc_request *irpc = irpc_call_send(msg_ctx, server_id, 
-                                                   table, callnum, r, mem_ctx);
-        return irpc_call_recv(irpc);
 }
 
@@ -1117 +1010 @@
         return msg_ctx->server_id;
 }
+
+struct irpc_bh_state {
+        struct messaging_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_INVALID_CONNECTION);
+                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;
+
+        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 = messaging_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_nomem(NULL, 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_INVALID_CONNECTION);
+                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 messaging_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;
+
+        dcerpc_binding_handle_set_sync_ev(h, msg_ctx->event.ev);
+
+        return h;
+}
+
+struct dcerpc_binding_handle *irpc_binding_handle_by_name(TALLOC_CTX *mem_ctx,
+                                        struct messaging_context *msg_ctx,
+                                        const char *dest_task,
+                                        const struct ndr_interface_table *table)
+{
+        struct dcerpc_binding_handle *h;
+        struct server_id *sids;
+        struct server_id sid;
+
+        /* find the server task */
+        sids = irpc_servers_byname(msg_ctx, mem_ctx, dest_task);
+        if (sids == NULL) {
+                errno = EADDRNOTAVAIL;
+                return NULL;
+        }
+        if (sids[0].id == 0) {
+                talloc_free(sids);
+                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;
+}