lib/pdu.c

   1 /*
   2    Copyright (C) 2010 by Ronnie Sahlberg <ronniesahlberg@gmail.com>
   3
   4    This program is free software; you can redistribute it and/or modify
   5    it under the terms of the GNU Lesser General Public License as published by
   6    the Free Software Foundation; either version 2.1 of the License, or
   7    (at your option) any later version.
   8
   9    This program is distributed in the hope that it will be useful,
  10    but WITHOUT ANY WARRANTY; without even the implied warranty of
  11    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12    GNU Lesser General Public License for more details.
  13
  14    You should have received a copy of the GNU Lesser General Public License
  15    along with this program; if not, see <http://www.gnu.org/licenses/>.
  16 */
  17 #ifdef HAVE_CONFIG_H
  18 #include "config.h"
  19 #endif
  20
  21 #ifdef AROS
  22 #include "aros_compat.h"
  23 #endif
  24
  25 #ifdef WIN32
  26 #include "win32_compat.h"
  27 #endif
  28
  29 #ifdef HAVE_NETINET_IN_H
  30 #include <netinet/in.h>
  31 #endif
  32
  33 #ifdef HAVE_SYS_SOCKET_H
  34 #include <sys/socket.h>
  35 #endif
  36
  37 #ifdef HAVE_STRINGS_H
  38 #include <strings.h>
  39 #endif
  40
  41 #include <stdio.h>
  42 #include <stdlib.h>
  43 #include <string.h>
  44 #include <assert.h>
  45 #include <errno.h>
  46 #include "slist.h"
  47 #include "libnfs-zdr.h"
  48 #include "libnfs.h"
  49 #include "libnfs-raw.h"
  50 #include "libnfs-private.h"
  51
  52 void rpc_reset_queue(struct rpc_queue *q)
  53 {
  54         q->head = NULL;
  55         q->tail = NULL;
  56 }
  57
  58 /*
  59  * Push to the tail end of the queue
  60  */
  61 void rpc_enqueue(struct rpc_queue *q, struct rpc_pdu *pdu)
  62 {
  63         if (q->head == NULL)
  64                 q->head = pdu;
  65         else
  66                 q->tail->next = pdu;
  67         q->tail = pdu;
  68         pdu->next = NULL;
  69 }
  70
  71 /*
  72  * Push to the front/head of the queue
  73  */
  74 void rpc_return_to_queue(struct rpc_queue *q, struct rpc_pdu *pdu)
  75 {
  76         pdu->next = q->head;
  77         q->head = pdu;
  78         if (q->tail == NULL)
  79                 q->tail = pdu;
  80 }
  81
  82 unsigned int rpc_hash_xid(uint32_t xid)
  83 {
  84         return (xid * 7919) % HASHES;
  85 }
  86
  87 struct rpc_pdu *rpc_allocate_pdu(struct rpc_context *rpc, int program, int version, int procedure, rpc_cb cb, void *private_data, zdrproc_t zdr_decode_fn, int zdr_decode_bufsize)
  88 {
  89         struct rpc_pdu *pdu;
  90         struct rpc_msg msg;
  91
  92         assert(rpc->magic == RPC_CONTEXT_MAGIC);
  93
  94         pdu = malloc(sizeof(struct rpc_pdu));
  95         if (pdu == NULL) {
  96                 rpc_set_error(rpc, "Out of memory: Failed to allocate pdu structure");
  97                 return NULL;
  98         }
  99         memset(pdu, 0, sizeof(struct rpc_pdu));
 100         pdu->xid                = rpc->xid++;
 101         pdu->cb                 = cb;
 102         pdu->private_data       = private_data;
 103         pdu->zdr_decode_fn      = zdr_decode_fn;
 104         pdu->zdr_decode_bufsize = zdr_decode_bufsize;
 105
 106         zdrmem_create(&pdu->zdr, rpc->encodebuf, rpc->encodebuflen, ZDR_ENCODE);
 107         if (rpc->is_udp == 0) {
 108                 zdr_setpos(&pdu->zdr, 4); /* skip past the record marker */
 109         }
 110
 111         memset(&msg, 0, sizeof(struct rpc_msg));
 112         msg.xid                = pdu->xid;
 113         msg.direction          = CALL;
 114         msg.body.cbody.rpcvers = RPC_MSG_VERSION;
 115         msg.body.cbody.prog    = program;
 116         msg.body.cbody.vers    = version;
 117         msg.body.cbody.proc    = procedure;
 118         msg.body.cbody.cred    = rpc->auth->ah_cred;
 119         msg.body.cbody.verf    = rpc->auth->ah_verf;
 120
 121         if (zdr_callmsg(rpc, &pdu->zdr, &msg) == 0) {
 122                 rpc_set_error(rpc, "zdr_callmsg failed with %s",
 123                               rpc_get_error(rpc));
 124                 zdr_destroy(&pdu->zdr);
 125                 free(pdu);
 126                 return NULL;
 127         }
 128
 129         return pdu;
 130 }
 131
 132 void rpc_free_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu)
 133 {
 134         assert(rpc->magic == RPC_CONTEXT_MAGIC);
 135
 136         if (pdu->outdata.data != NULL) {
 137                 free(pdu->outdata.data);
 138                 pdu->outdata.data = NULL;
 139         }
 140
 141         if (pdu->zdr_decode_buf != NULL) {
 142                 zdr_free(pdu->zdr_decode_fn, pdu->zdr_decode_buf);
 143                 free(pdu->zdr_decode_buf);
 144                 pdu->zdr_decode_buf = NULL;
 145         }
 146
 147         zdr_destroy(&pdu->zdr);
 148
 149         free(pdu);
 150 }
 151
 152 void rpc_set_next_xid(struct rpc_context *rpc, uint32_t xid)
 153 {
 154         rpc->xid = xid;
 155 }
 156
 157 int rpc_queue_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu)
 158 {
 159         int size, recordmarker;
 160
 161         assert(rpc->magic == RPC_CONTEXT_MAGIC);
 162
 163         size = zdr_getpos(&pdu->zdr);
 164
 165         /* for udp we dont queue, we just send it straight away */
 166         if (rpc->is_udp != 0) {
 167                 unsigned int hash;
 168
 169 // XXX add a rpc->udp_dest_sock_size  and get rid of sys/socket.h and netinet/in.h
 170                 if (sendto(rpc->fd, rpc->encodebuf, size, MSG_DONTWAIT, rpc->udp_dest, sizeof(struct sockaddr_in)) < 0) {
 171                         rpc_set_error(rpc, "Sendto failed with errno %s", strerror(errno));
 172                         rpc_free_pdu(rpc, pdu);
 173                         return -1;
 174                 }
 175
 176                 hash = rpc_hash_xid(pdu->xid);
 177                 rpc_enqueue(&rpc->waitpdu[hash], pdu);
 178                 return 0;
 179         }
 180
 181         /* write recordmarker */
 182         zdr_setpos(&pdu->zdr, 0);
 183         recordmarker = (size - 4) | 0x80000000;
 184         zdr_int(&pdu->zdr, &recordmarker);
 185
 186         pdu->outdata.size = size;
 187         pdu->outdata.data = malloc(pdu->outdata.size);
 188         if (pdu->outdata.data == NULL) {
 189                 rpc_set_error(rpc, "Out of memory. Failed to allocate buffer for pdu\n");
 190                 rpc_free_pdu(rpc, pdu);
 191                 return -1;
 192         }
 193
 194         memcpy(pdu->outdata.data, rpc->encodebuf, pdu->outdata.size);
 195         rpc_enqueue(&rpc->outqueue, pdu);
 196
 197         return 0;
 198 }
 199
 200 int rpc_get_pdu_size(char *buf)
 201 {
 202         uint32_t size;
 203
 204         size = ntohl(*(uint32_t *)buf);
 205
 206         return (size & 0x7fffffff) + 4;
 207 }
 208
 209 static int rpc_process_reply(struct rpc_context *rpc, struct rpc_pdu *pdu, ZDR *zdr)
 210 {
 211         struct rpc_msg msg;
 212
 213         assert(rpc->magic == RPC_CONTEXT_MAGIC);
 214
 215         memset(&msg, 0, sizeof(struct rpc_msg));
 216         msg.body.rbody.reply.areply.verf = _null_auth;
 217         if (pdu->zdr_decode_bufsize > 0) {
 218                 if (pdu->zdr_decode_buf != NULL) {
 219                         free(pdu->zdr_decode_buf);
 220                 }
 221                 pdu->zdr_decode_buf = malloc(pdu->zdr_decode_bufsize);
 222                 if (pdu->zdr_decode_buf == NULL) {
 223                         rpc_set_error(rpc, "Failed to allocate memory for "
 224                                       "zdr_encode_buf in rpc_process_reply");
 225                         pdu->cb(rpc, RPC_STATUS_ERROR, "Failed to allocate "
 226                                 "buffer for decoding of ZDR reply",
 227                                 pdu->private_data);
 228                         return 0;
 229                 }
 230                 memset(pdu->zdr_decode_buf, 0, pdu->zdr_decode_bufsize);
 231         }
 232         msg.body.rbody.reply.areply.reply_data.results.where = pdu->zdr_decode_buf;
 233         msg.body.rbody.reply.areply.reply_data.results.proc  = pdu->zdr_decode_fn;
 234
 235         if (zdr_replymsg(rpc, zdr, &msg) == 0) {
 236                 rpc_set_error(rpc, "zdr_replymsg failed in rpc_process_reply: "
 237                               "%s", rpc_get_error(rpc));
 238                 pdu->cb(rpc, RPC_STATUS_ERROR, "Message rejected by server",
 239                         pdu->private_data);
 240                 if (pdu->zdr_decode_buf != NULL) {
 241                         free(pdu->zdr_decode_buf);
 242                         pdu->zdr_decode_buf = NULL;
 243                 }
 244                 return 0;
 245         }
 246         if (msg.body.rbody.stat != MSG_ACCEPTED) {
 247                 pdu->cb(rpc, RPC_STATUS_ERROR, "RPC Packet not accepted by the server", pdu->private_data);
 248                 return 0;
 249         }
 250         switch (msg.body.rbody.reply.areply.stat) {
 251         case SUCCESS:
 252                 pdu->cb(rpc, RPC_STATUS_SUCCESS, pdu->zdr_decode_buf, pdu->private_data);
 253                 break;
 254         case PROG_UNAVAIL:
 255                 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: Program not available", pdu->private_data);
 256                 break;
 257         case PROG_MISMATCH:
 258                 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: Program version mismatch", pdu->private_data);
 259                 break;
 260         case PROC_UNAVAIL:
 261                 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: Procedure not available", pdu->private_data);
 262                 break;
 263         case GARBAGE_ARGS:
 264                 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: Garbage arguments", pdu->private_data);
 265                 break;
 266         case SYSTEM_ERR:
 267                 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: System Error", pdu->private_data);
 268                 break;
 269         default:
 270                 pdu->cb(rpc, RPC_STATUS_ERROR, "Unknown rpc response from server", pdu->private_data);
 271                 break;
 272         }
 273
 274         return 0;
 275 }
 276
 277 int rpc_process_pdu(struct rpc_context *rpc, char *buf, int size)
 278 {
 279         struct rpc_pdu *pdu, *prev_pdu;
 280         struct rpc_queue *q;
 281         ZDR zdr;
 282         int pos, recordmarker = 0;
 283         unsigned int hash;
 284         uint32_t xid;
 285         char *reasbuf = NULL;
 286
 287         assert(rpc->magic == RPC_CONTEXT_MAGIC);
 288
 289         memset(&zdr, 0, sizeof(ZDR));
 290
 291         zdrmem_create(&zdr, buf, size, ZDR_DECODE);
 292         if (rpc->is_udp == 0) {
 293                 if (zdr_int(&zdr, &recordmarker) == 0) {
 294                         rpc_set_error(rpc, "zdr_int reading recordmarker failed");
 295                         zdr_destroy(&zdr);
 296                         return -1;
 297                 }
 298                 if (!(recordmarker&0x80000000)) {
 299                         zdr_destroy(&zdr);
 300                         if (rpc_add_fragment(rpc, buf+4, size-4) != 0) {
 301                                 rpc_set_error(rpc, "Failed to queue fragment for reassembly.");
 302                                 return -1;
 303                         }
 304                         return 0;
 305                 }
 306         }
 307
 308         /* reassembly */
 309         if (recordmarker != 0 && rpc->fragments != NULL) {
 310                 struct rpc_fragment *fragment;
 311                 uint32_t total = size - 4;
 312                 char *ptr;
 313
 314                 zdr_destroy(&zdr);
 315                 for (fragment = rpc->fragments; fragment; fragment = fragment->next) {
 316                         total += fragment->size;
 317                 }
 318
 319                 reasbuf = malloc(total);
 320                 if (reasbuf == NULL) {
 321                         rpc_set_error(rpc, "Failed to reassemble PDU");
 322                         rpc_free_all_fragments(rpc);
 323                         return -1;
 324                 }
 325                 ptr = reasbuf;
 326                 for (fragment = rpc->fragments; fragment; fragment = fragment->next) {
 327                         memcpy(ptr, fragment->data, fragment->size);
 328                         ptr += fragment->size;
 329                 }
 330                 memcpy(ptr, buf + 4, size - 4);
 331                 zdrmem_create(&zdr, reasbuf, total, ZDR_DECODE);
 332                 rpc_free_all_fragments(rpc);
 333         }
 334
 335         pos = zdr_getpos(&zdr);
 336         if (zdr_int(&zdr, (int *)&xid) == 0) {
 337                 rpc_set_error(rpc, "zdr_int reading xid failed");
 338                 zdr_destroy(&zdr);
 339                 if (reasbuf != NULL) {
 340                         free(reasbuf);
 341                 }
 342                 return -1;
 343         }
 344         zdr_setpos(&zdr, pos);
 345
 346         /* Look up the transaction in a hash table of our requests */
 347         hash = rpc_hash_xid(xid);
 348         q = &rpc->waitpdu[hash];
 349
 350         /* Follow the hash chain.  Linear traverse singly-linked list,
 351          * but track previous entry for optimised removal */
 352         prev_pdu = NULL;
 353         for (pdu=q->head; pdu; pdu=pdu->next) {
 354                 if (pdu->xid != xid) {
 355                         prev_pdu = pdu;
 356                         continue;
 357                 }
 358                 if (rpc->is_udp == 0 || rpc->is_broadcast == 0) {
 359                         /* Singly-linked but we track head and tail */
 360                         if (pdu == q->head)
 361                                 q->head = pdu->next;
 362                         if (pdu == q->tail)
 363                                 q->tail = prev_pdu;
 364                         if (prev_pdu != NULL)
 365                                 prev_pdu->next = pdu->next;
 366                 }
 367                 if (rpc_process_reply(rpc, pdu, &zdr) != 0) {
 368                         rpc_set_error(rpc, "rpc_procdess_reply failed");
 369                 }
 370                 zdr_destroy(&zdr);
 371                 if (rpc->is_udp == 0 || rpc->is_broadcast == 0) {
 372                         rpc_free_pdu(rpc, pdu);
 373                 }
 374                 if (reasbuf != NULL) {
 375                         free(reasbuf);
 376                 }
 377                 return 0;
 378         }
 379         rpc_set_error(rpc, "No matching pdu found for xid:%d", xid);
 380         zdr_destroy(&zdr);
 381         if (reasbuf != NULL) {
 382                 free(reasbuf);
 383         }
 384         return -1;
 385 }
 386