source4/lib/messaging/messaging.c

   1 /*
   2    Unix SMB/CIFS implementation.
   3
   4    Samba internal messaging functions
   5
   6    Copyright (C) Andrew Tridgell 2004
   7
   8    This program is free software; you can redistribute it and/or modify
   9    it under the terms of the GNU General Public License as published by
  10    the Free Software Foundation; either version 2 of the License, or
  11    (at your option) any later version.
  12
  13    This program is distributed in the hope that it will be useful,
  14    but WITHOUT ANY WARRANTY; without even the implied warranty of
  15    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16    GNU General Public License for more details.
  17
  18    You should have received a copy of the GNU General Public License
  19    along with this program; if not, write to the Free Software
  20    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  21 */
  22
  23 #include "includes.h"
  24 #include "lib/events/events.h"
  25 #include "system/dir.h"
  26 #include "system/filesys.h"
  27 #include "system/time.h"
  28 #include "messages.h"
  29 #include "dlinklist.h"
  30 #include "lib/socket/socket.h"
  31 #include "librpc/gen_ndr/ndr_irpc.h"
  32 #include "lib/messaging/irpc.h"
  33
  34 /* change the message version with any incompatible changes in the protocol */
  35 #define MESSAGING_VERSION 1
  36
  37 struct messaging_context {
  38         uint32_t server_id;
  39         struct socket_context *sock;
  40         const char *base_path;
  41         const char *path;
  42         struct dispatch_fn *dispatch;
  43         struct messaging_rec *pending;
  44         struct irpc_list *irpc;
  45         struct idr_context *idr;
  46
  47         struct {
  48                 struct event_context *ev;
  49                 struct fd_event *fde;
  50         } event;
  51 };
  52
  53 /* we have a linked list of dispatch handlers that this messaging
  54    server can deal with */
  55 struct dispatch_fn {
  56         struct dispatch_fn *next, *prev;
  57         uint32_t msg_type;
  58         void *private;
  59         void (*fn)(struct messaging_context *msg, void *private,
  60                    uint32_t msg_type, uint32_t server_id, DATA_BLOB *data);
  61 };
  62
  63 /* an individual message */
  64 struct messaging_rec {
  65         struct messaging_rec *next, *prev;
  66         struct messaging_context *msg;
  67         const char *path;
  68
  69         struct messaging_header {
  70                 uint32_t version;
  71                 uint32_t msg_type;
  72                 uint32_t from;
  73                 uint32_t to;
  74                 uint32_t length;
  75         } *header;
  76
  77         DATA_BLOB packet;
  78 };
  79
  80
  81 static void irpc_handler(struct messaging_context *, void *,
  82                          uint32_t, uint32_t, DATA_BLOB *);
  83
  84
  85 /*
  86  A useful function for testing the message system.
  87 */
  88 static void ping_message(struct messaging_context *msg, void *private,
  89                          uint32_t msg_type, uint32_t src, DATA_BLOB *data)
  90 {
  91         DEBUG(1,("INFO: Received PING message from server %u [%.*s]\n",
  92                  (uint_t)src, data->length, data->data?(const char *)data->data:""));
  93         messaging_send(msg, src, MSG_PONG, data);
  94 }
  95
  96 /*
  97    return the path to a messaging socket
  98 */
  99 static char *messaging_path(struct messaging_context *msg, uint32_t server_id)
 100 {
 101         return talloc_asprintf(msg, "%s/msg.%u", msg->base_path, (unsigned)server_id);
 102 }
 103
 104 /*
 105   dispatch a fully received message
 106 */
 107 static void messaging_dispatch(struct messaging_context *msg, struct messaging_rec *rec)
 108 {
 109         struct dispatch_fn *d, *next;
 110         for (d=msg->dispatch;d;d=next) {
 111                 next = d->next;
 112                 if (d->msg_type == rec->header->msg_type) {
 113                         DATA_BLOB data;
 114                         data.data = rec->packet.data + sizeof(*rec->header);
 115                         data.length = rec->header->length;
 116                         d->fn(msg, d->private, d->msg_type, rec->header->from, &data);
 117                 }
 118         }
 119         rec->header->length = 0;
 120 }
 121
 122
 123 /*
 124   try to send the message
 125 */
 126 static NTSTATUS try_send(struct messaging_rec *rec)
 127 {
 128         struct messaging_context *msg = rec->msg;
 129         size_t nsent;
 130         void *priv;
 131         NTSTATUS status;
 132
 133         /* we send with privileges so messages work from any context */
 134         priv = root_privileges();
 135         status = socket_sendto(msg->sock, &rec->packet, &nsent, 0, rec->path, 0);
 136         talloc_free(priv);
 137
 138         return status;
 139 }
 140
 141 /*
 142   handle a socket write event
 143 */
 144 static void messaging_send_handler(struct messaging_context *msg)
 145 {
 146         while (msg->pending) {
 147                 struct messaging_rec *rec = msg->pending;
 148                 NTSTATUS status;
 149                 status = try_send(rec);
 150                 if (NT_STATUS_EQUAL(status, STATUS_MORE_ENTRIES)) {
 151                         break;
 152                 }
 153                 if (!NT_STATUS_IS_OK(status)) {
 154                         DEBUG(1,("messaging: Lost message from %u to %u of type %u - %s\n",
 155                                  rec->header->from, rec->header->to, rec->header->msg_type,
 156                                  nt_errstr(status)));
 157                 }
 158                 DLIST_REMOVE(msg->pending, rec);
 159                 talloc_free(rec);
 160         }
 161         if (msg->pending == NULL) {
 162                 EVENT_FD_NOT_WRITEABLE(msg->event.fde);
 163         }
 164 }
 165
 166 /*
 167   handle a new incoming packet
 168 */
 169 static void messaging_recv_handler(struct messaging_context *msg)
 170 {
 171         struct messaging_rec *rec;
 172         NTSTATUS status;
 173         DATA_BLOB packet;
 174         size_t msize;
 175
 176         /* see how many bytes are in the next packet */
 177         status = socket_pending(msg->sock, &msize);
 178         if (!NT_STATUS_IS_OK(status)) {
 179                 DEBUG(0,("socket_pending failed in messaging - %s\n",
 180                          nt_errstr(status)));
 181                 return;
 182         }
 183
 184         packet = data_blob_talloc(msg, NULL, msize);
 185         if (packet.data == NULL) {
 186                 /* assume this is temporary and retry */
 187                 return;
 188         }
 189
 190         status = socket_recv(msg->sock, packet.data, msize, &msize, 0);
 191         if (!NT_STATUS_IS_OK(status)) {
 192                 data_blob_free(&packet);
 193                 return;
 194         }
 195
 196         if (msize < sizeof(*rec->header)) {
 197                 DEBUG(0,("messaging: bad message of size %d\n", msize));
 198                 data_blob_free(&packet);
 199                 return;
 200         }
 201
 202         rec = talloc(msg, struct messaging_rec);
 203         if (rec == NULL) {
 204                 smb_panic("Unable to allocate messaging_rec");
 205         }
 206
 207         talloc_steal(rec, packet.data);
 208         rec->msg           = msg;
 209         rec->path          = msg->path;
 210         rec->header        = (struct messaging_header *)packet.data;
 211         rec->packet        = packet;
 212
 213         if (msize != sizeof(*rec->header) + rec->header->length) {
 214                 DEBUG(0,("messaging: bad message header size %d should be %d\n",
 215                          rec->header->length, msize - sizeof(*rec->header)));
 216                 talloc_free(rec);
 217                 return;
 218         }
 219
 220         messaging_dispatch(msg, rec);
 221         talloc_free(rec);
 222 }
 223
 224
 225 /*
 226   handle a socket event
 227 */
 228 static void messaging_handler(struct event_context *ev, struct fd_event *fde,
 229                               uint16_t flags, void *private)
 230 {
 231         struct messaging_context *msg = talloc_get_type(private,
 232                                                         struct messaging_context);
 233         if (flags & EVENT_FD_WRITE) {
 234                 messaging_send_handler(msg);
 235         }
 236         if (flags & EVENT_FD_READ) {
 237                 messaging_recv_handler(msg);
 238         }
 239 }
 240
 241
 242 /*
 243   Register a dispatch function for a particular message type.
 244 */
 245 void messaging_register(struct messaging_context *msg, void *private,
 246                         uint32_t msg_type,
 247                         void (*fn)(struct messaging_context *, void *, uint32_t, uint32_t, DATA_BLOB *))
 248 {
 249         struct dispatch_fn *d;
 250
 251         d = talloc(msg, struct dispatch_fn);
 252         d->msg_type = msg_type;
 253         d->private = private;
 254         d->fn = fn;
 255         DLIST_ADD(msg->dispatch, d);
 256 }
 257
 258 /*
 259   De-register the function for a particular message type.
 260 */
 261 void messaging_deregister(struct messaging_context *msg, uint32_t msg_type, void *private)
 262 {
 263         struct dispatch_fn *d, *next;
 264
 265         for (d = msg->dispatch; d; d = next) {
 266                 next = d->next;
 267                 if (d->msg_type == msg_type &&
 268                     d->private == private) {
 269                         DLIST_REMOVE(msg->dispatch, d);
 270                         talloc_free(d);
 271                 }
 272         }
 273 }
 274
 275
 276 /*
 277   Send a message to a particular server
 278 */
 279 NTSTATUS messaging_send(struct messaging_context *msg, uint32_t server,
 280                         uint32_t msg_type, DATA_BLOB *data)
 281 {
 282         struct messaging_rec *rec;
 283         NTSTATUS status;
 284         size_t dlength = data?data->length:0;
 285
 286         rec = talloc(msg, struct messaging_rec);
 287         if (rec == NULL) {
 288                 return NT_STATUS_NO_MEMORY;
 289         }
 290
 291         rec->packet = data_blob_talloc(rec, NULL, sizeof(*rec->header) + dlength);
 292         if (rec->packet.data == NULL) {
 293                 talloc_free(rec);
 294                 return NT_STATUS_NO_MEMORY;
 295         }
 296
 297         rec->msg              = msg;
 298         rec->header           = (struct messaging_header *)rec->packet.data;
 299         rec->header->version  = MESSAGING_VERSION;
 300         rec->header->msg_type = msg_type;
 301         rec->header->from     = msg->server_id;
 302         rec->header->to       = server;
 303         rec->header->length   = dlength;
 304         if (dlength != 0) {
 305                 memcpy(rec->packet.data + sizeof(*rec->header),
 306                        data->data, dlength);
 307         }
 308
 309         rec->path = messaging_path(msg, server);
 310         talloc_steal(rec, rec->path);
 311
 312         if (msg->pending != NULL) {
 313                 status = STATUS_MORE_ENTRIES;
 314         } else {
 315                 status = try_send(rec);
 316         }
 317
 318         if (NT_STATUS_EQUAL(status, STATUS_MORE_ENTRIES)) {
 319                 if (msg->pending == NULL) {
 320                         EVENT_FD_WRITEABLE(msg->event.fde);
 321                 }
 322                 DLIST_ADD_END(msg->pending, rec, struct messaging_rec *);
 323                 return NT_STATUS_OK;
 324         }
 325
 326         talloc_free(rec);
 327
 328         return status;
 329 }
 330
 331 /*
 332   Send a message to a particular server, with the message containing a single pointer
 333 */
 334 NTSTATUS messaging_send_ptr(struct messaging_context *msg, uint32_t server,
 335                             uint32_t msg_type, void *ptr)
 336 {
 337         DATA_BLOB blob;
 338
 339         blob.data = (void *)&ptr;
 340         blob.length = sizeof(void *);
 341
 342         return messaging_send(msg, server, msg_type, &blob);
 343 }
 344
 345
 346 /*
 347   destroy the messaging context
 348 */
 349 static int messaging_destructor(void *ptr)
 350 {
 351         struct messaging_context *msg = ptr;
 352         unlink(msg->path);
 353         return 0;
 354 }
 355
 356 /*
 357   create the listening socket and setup the dispatcher
 358 */
 359 struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx, uint32_t server_id,
 360                                          struct event_context *ev)
 361 {
 362         struct messaging_context *msg;
 363         NTSTATUS status;
 364         char *path;
 365
 366         msg = talloc(mem_ctx, struct messaging_context);
 367         if (msg == NULL) {
 368                 return NULL;
 369         }
 370
 371         /* create the messaging directory if needed */
 372         path = smbd_tmp_path(msg, "messaging");
 373         mkdir(path, 0700);
 374         talloc_free(path);
 375
 376         msg->base_path = smbd_tmp_path(msg, "messaging");
 377         msg->path      = messaging_path(msg, server_id);
 378         msg->server_id = server_id;
 379         msg->dispatch  = NULL;
 380         msg->pending   = NULL;
 381         msg->idr       = idr_init(msg);
 382         msg->irpc      = NULL;
 383
 384         status = socket_create("unix", SOCKET_TYPE_DGRAM, &msg->sock, 0);
 385         if (!NT_STATUS_IS_OK(status)) {
 386                 talloc_free(msg);
 387                 return NULL;
 388         }
 389
 390         /* by stealing here we ensure that the socket is cleaned up (and even
 391            deleted) on exit */
 392         talloc_steal(msg, msg->sock);
 393
 394         status = socket_listen(msg->sock, msg->path, 0, 50, 0);
 395         if (!NT_STATUS_IS_OK(status)) {
 396                 DEBUG(0,("Unable to setup messaging listener for '%s'\n", msg->path));
 397                 talloc_free(msg);
 398                 return NULL;
 399         }
 400
 401         /* it needs to be non blocking for sends */
 402         set_blocking(socket_get_fd(msg->sock), False);
 403
 404         msg->event.ev   = talloc_reference(msg, ev);
 405         msg->event.fde  = event_add_fd(ev, msg, socket_get_fd(msg->sock),
 406                                        EVENT_FD_READ, messaging_handler, msg);
 407
 408         talloc_set_destructor(msg, messaging_destructor);
 409
 410         messaging_register(msg, NULL, MSG_PING, ping_message);
 411         messaging_register(msg, NULL, MSG_IRPC, irpc_handler);
 412
 413         return msg;
 414 }
 415
 416
 417 /*
 418   a list of registered irpc server functions
 419 */
 420 struct irpc_list {
 421         struct irpc_list *next, *prev;
 422         struct GUID uuid;
 423         const struct dcerpc_interface_table *table;
 424         int callnum;
 425         irpc_function_t fn;
 426         void *private;
 427 };
 428
 429
 430 /*
 431   register a irpc server function
 432 */
 433 NTSTATUS irpc_register(struct messaging_context *msg_ctx,
 434                        const struct dcerpc_interface_table *table,
 435                        int callnum, irpc_function_t fn, void *private)
 436 {
 437         struct irpc_list *irpc;
 438
 439         /* override an existing handler, if any */
 440         for (irpc=msg_ctx->irpc; irpc; irpc=irpc->next) {
 441                 if (irpc->table == table && irpc->callnum == callnum) {
 442                         break;
 443                 }
 444         }
 445         if (irpc == NULL) {
 446                 irpc = talloc(msg_ctx, struct irpc_list);
 447                 NT_STATUS_HAVE_NO_MEMORY(irpc);
 448                 DLIST_ADD(msg_ctx->irpc, irpc);
 449         }
 450
 451         irpc->table   = table;
 452         irpc->callnum = callnum;
 453         irpc->fn      = fn;
 454         irpc->private = private;
 455         GUID_from_string(irpc->table->uuid, &irpc->uuid);
 456
 457         return NT_STATUS_OK;
 458 }
 459
 460
 461 /*
 462   handle an incoming irpc reply message
 463 */
 464 static void irpc_handler_reply(struct messaging_context *msg_ctx,
 465                                struct ndr_pull *ndr, struct irpc_header *header)
 466 {
 467         struct irpc_request *irpc;
 468
 469         irpc = idr_find(msg_ctx->idr, header->callid);
 470         if (irpc == NULL) return;
 471
 472         /* parse the reply data */
 473         irpc->status = irpc->table->calls[irpc->callnum].ndr_pull(ndr, NDR_OUT, irpc->r);
 474         if (NT_STATUS_IS_OK(irpc->status)) {
 475                 irpc->status = header->status;
 476         }
 477         irpc->done = True;
 478         if (irpc->async.fn) {
 479                 irpc->async.fn(irpc);
 480         }
 481 }
 482
 483
 484 /*
 485   handle an incoming irpc request message
 486 */
 487 static void irpc_handler_request(struct messaging_context *msg_ctx,
 488                                  struct ndr_pull *ndr, struct irpc_header *header,
 489                                  uint32_t src)
 490 {
 491         struct irpc_list *i;
 492         void *r;
 493         NTSTATUS status;
 494         struct irpc_message m;
 495         struct ndr_push *push;
 496         DATA_BLOB packet;
 497
 498         for (i=msg_ctx->irpc; i; i=i->next) {
 499                 if (GUID_equal(&i->uuid, &header->uuid) &&
 500                     i->table->if_version == header->if_version &&
 501                     i->callnum == header->callnum) {
 502                         break;
 503                 }
 504         }
 505
 506         if (i == NULL) {
 507                 /* no registered handler for this message */
 508                 return;
 509         }
 510
 511         /* allocate space for the structure */
 512         r = talloc_zero_size(ndr, i->table->calls[header->callnum].struct_size);
 513         if (r == NULL) goto failed;
 514
 515         /* parse the request data */
 516         status = i->table->calls[i->callnum].ndr_pull(ndr, NDR_IN, r);
 517         if (!NT_STATUS_IS_OK(status)) goto failed;
 518
 519         /* make the call */
 520         m.from    = src;
 521         m.private = i->private;
 522         header->status = i->fn(&m, r);
 523
 524         /* setup the reply */
 525         push = ndr_push_init_ctx(ndr);
 526         if (push == NULL) goto failed;
 527
 528         header->flags |= IRPC_FLAG_REPLY;
 529
 530         /* construct the packet */
 531         status = ndr_push_irpc_header(push, NDR_SCALARS|NDR_BUFFERS, header);
 532         if (!NT_STATUS_IS_OK(status)) goto failed;
 533
 534         status = i->table->calls[i->callnum].ndr_push(push, NDR_OUT, r);
 535         if (!NT_STATUS_IS_OK(status)) goto failed;
 536
 537         /* send the reply message */
 538         packet = ndr_push_blob(push);
 539         status = messaging_send(msg_ctx, src, MSG_IRPC, &packet);
 540         if (!NT_STATUS_IS_OK(status)) goto failed;
 541
 542 failed:
 543         /* nothing to clean up */
 544         return;
 545 }
 546
 547 /*
 548   handle an incoming irpc message
 549 */
 550 static void irpc_handler(struct messaging_context *msg_ctx, void *private,
 551                          uint32_t msg_type, uint32_t src, DATA_BLOB *packet)
 552 {
 553         struct irpc_header header;
 554         struct ndr_pull *ndr;
 555         NTSTATUS status;
 556
 557         ndr = ndr_pull_init_blob(packet, msg_ctx);
 558         if (ndr == NULL) goto failed;
 559
 560         status = ndr_pull_irpc_header(ndr, NDR_BUFFERS|NDR_SCALARS, &header);
 561         if (!NT_STATUS_IS_OK(status)) goto failed;
 562
 563         if (header.flags & IRPC_FLAG_REPLY) {
 564                 irpc_handler_reply(msg_ctx, ndr, &header);
 565         } else {
 566                 irpc_handler_request(msg_ctx, ndr, &header, src);
 567         }
 568
 569 failed:
 570         talloc_free(ndr);
 571 }
 572
 573
 574 /*
 575   destroy a irpc request
 576 */
 577 static int irpc_destructor(void *ptr)
 578 {
 579         struct irpc_request *irpc = talloc_get_type(ptr, struct irpc_request);
 580         idr_remove(irpc->msg_ctx->idr, irpc->callid);
 581         return 0;
 582 }
 583
 584 /*
 585   timeout a irpc request
 586 */
 587 static void irpc_timeout(struct event_context *ev, struct timed_event *te,
 588                          struct timeval t, void *private)
 589 {
 590         struct irpc_request *irpc = talloc_get_type(private, struct irpc_request);
 591         irpc->status = NT_STATUS_IO_TIMEOUT;
 592         irpc->done = True;
 593         if (irpc->async.fn) {
 594                 irpc->async.fn(irpc);
 595         }
 596 }
 597
 598
 599 /*
 600   make a irpc call - async send
 601 */
 602 struct irpc_request *irpc_call_send(struct messaging_context *msg_ctx,
 603                                     uint32_t server_id,
 604                                     const struct dcerpc_interface_table *table,
 605                                     int callnum, void *r)
 606 {
 607         struct irpc_header header;
 608         struct ndr_push *ndr;
 609         NTSTATUS status;
 610         DATA_BLOB packet;
 611         struct irpc_request *irpc;
 612
 613         irpc = talloc(msg_ctx, struct irpc_request);
 614         if (irpc == NULL) goto failed;
 615
 616         irpc->msg_ctx  = msg_ctx;
 617         irpc->table    = table;
 618         irpc->callnum  = callnum;
 619         irpc->callid   = idr_get_new(msg_ctx->idr, irpc, UINT16_MAX);
 620         if (irpc->callid == -1) goto failed;
 621         irpc->r        = r;
 622         irpc->done     = False;
 623         irpc->async.fn = NULL;
 624
 625         talloc_set_destructor(irpc, irpc_destructor);
 626
 627         /* setup the header */
 628         status = GUID_from_string(table->uuid, &header.uuid);
 629         if (!NT_STATUS_IS_OK(status)) goto failed;
 630
 631         header.if_version = table->if_version;
 632         header.callid     = irpc->callid;
 633         header.callnum    = callnum;
 634         header.flags      = 0;
 635         header.status     = NT_STATUS_OK;
 636
 637         /* construct the irpc packet */
 638         ndr = ndr_push_init_ctx(irpc);
 639         if (ndr == NULL) goto failed;
 640
 641         status = ndr_push_irpc_header(ndr, NDR_SCALARS|NDR_BUFFERS, &header);
 642         if (!NT_STATUS_IS_OK(status)) goto failed;
 643
 644         status = table->calls[callnum].ndr_push(ndr, NDR_IN, r);
 645         if (!NT_STATUS_IS_OK(status)) goto failed;
 646
 647         /* and send it */
 648         packet = ndr_push_blob(ndr);
 649         status = messaging_send(msg_ctx, server_id, MSG_IRPC, &packet);
 650         if (!NT_STATUS_IS_OK(status)) goto failed;
 651
 652         event_add_timed(msg_ctx->event.ev, irpc,
 653                         timeval_current_ofs(IRPC_CALL_TIMEOUT, 0),
 654                         irpc_timeout, irpc);
 655
 656         talloc_free(ndr);
 657         return irpc;
 658
 659 failed:
 660         talloc_free(irpc);
 661         return NULL;
 662 }
 663
 664 /*
 665   wait for a irpc reply
 666 */
 667 NTSTATUS irpc_call_recv(struct irpc_request *irpc)
 668 {
 669         NTSTATUS status;
 670         NT_STATUS_HAVE_NO_MEMORY(irpc);
 671         while (!irpc->done) {
 672                 if (event_loop_once(irpc->msg_ctx->event.ev) != 0) {
 673                         return NT_STATUS_CONNECTION_DISCONNECTED;
 674                 }
 675         }
 676         status = irpc->status;
 677         talloc_free(irpc);
 678         return status;
 679 }
 680
 681 /*
 682   perform a synchronous irpc request
 683 */
 684 NTSTATUS irpc_call(struct messaging_context *msg_ctx,
 685                    uint32_t server_id,
 686                    const struct dcerpc_interface_table *table,
 687                    int callnum, void *r)
 688 {
 689         struct irpc_request *irpc = irpc_call_send(msg_ctx, server_id,
 690                                                    table, callnum, r);
 691         return irpc_call_recv(irpc);
 692 }