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 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)
  83 {
  84         struct rpc_pdu *pdu;
  85         struct rpc_msg msg;
  86
  87         assert(rpc->magic == RPC_CONTEXT_MAGIC);
  88
  89         pdu = malloc(sizeof(struct rpc_pdu));
  90         if (pdu == NULL) {
  91                 rpc_set_error(rpc, "Out of memory: Failed to allocate pdu structure");
  92                 return NULL;
  93         }
  94         memset(pdu, 0, sizeof(struct rpc_pdu));
  95         pdu->xid                = rpc->xid++;
  96         pdu->cb                 = cb;
  97         pdu->private_data       = private_data;
  98         pdu->zdr_decode_fn      = zdr_decode_fn;
  99         pdu->zdr_decode_bufsize = zdr_decode_bufsize;
 100
 101         zdrmem_create(&pdu->zdr, rpc->encodebuf, rpc->encodebuflen, ZDR_ENCODE);
 102         if (rpc->is_udp == 0) {
 103                 zdr_setpos(&pdu->zdr, 4); /* skip past the record marker */
 104         }
 105
 106         memset(&msg, 0, sizeof(struct rpc_msg));
 107         msg.xid                = pdu->xid;
 108         msg.direction          = CALL;
 109         msg.body.cbody.rpcvers = RPC_MSG_VERSION;
 110         msg.body.cbody.prog    = program;
 111         msg.body.cbody.vers    = version;
 112         msg.body.cbody.proc    = procedure;
 113         msg.body.cbody.cred    = rpc->auth->ah_cred;
 114         msg.body.cbody.verf    = rpc->auth->ah_verf;
 115
 116         if (zdr_callmsg(rpc, &pdu->zdr, &msg) == 0) {
 117                 rpc_set_error(rpc, "zdr_callmsg failed with %s",
 118                               rpc_get_error(rpc));
 119                 zdr_destroy(&pdu->zdr);
 120                 free(pdu);
 121                 return NULL;
 122         }
 123
 124         return pdu;
 125 }
 126
 127 void rpc_free_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu)
 128 {
 129         assert(rpc->magic == RPC_CONTEXT_MAGIC);
 130
 131         if (pdu->outdata.data != NULL) {
 132                 free(pdu->outdata.data);
 133                 pdu->outdata.data = NULL;
 134         }
 135
 136         if (pdu->zdr_decode_buf != NULL) {
 137                 zdr_free(pdu->zdr_decode_fn, pdu->zdr_decode_buf);
 138                 free(pdu->zdr_decode_buf);
 139                 pdu->zdr_decode_buf = NULL;
 140         }
 141
 142         zdr_destroy(&pdu->zdr);
 143
 144         free(pdu);
 145 }
 146
 147 void rpc_set_next_xid(struct rpc_context *rpc, uint32_t xid)
 148 {
 149         rpc->xid = xid;
 150 }
 151
 152 int rpc_queue_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu)
 153 {
 154         int size, recordmarker;
 155
 156         assert(rpc->magic == RPC_CONTEXT_MAGIC);
 157
 158         size = zdr_getpos(&pdu->zdr);
 159
 160         /* for udp we dont queue, we just send it straight away */
 161         if (rpc->is_udp != 0) {
 162 // XXX add a rpc->udp_dest_sock_size  and get rid of sys/socket.h and netinet/in.h
 163                 if (sendto(rpc->fd, rpc->encodebuf, size, MSG_DONTWAIT, rpc->udp_dest, sizeof(struct sockaddr_in)) < 0) {
 164                         rpc_set_error(rpc, "Sendto failed with errno %s", strerror(errno));
 165                         rpc_free_pdu(rpc, pdu);
 166                         return -1;
 167                 }
 168                 rpc_enqueue(&rpc->waitpdu, pdu);
 169                 return 0;
 170         }
 171
 172         /* write recordmarker */
 173         zdr_setpos(&pdu->zdr, 0);
 174         recordmarker = (size - 4) | 0x80000000;
 175         zdr_int(&pdu->zdr, &recordmarker);
 176
 177         pdu->outdata.size = size;
 178         pdu->outdata.data = malloc(pdu->outdata.size);
 179         if (pdu->outdata.data == NULL) {
 180                 rpc_set_error(rpc, "Out of memory. Failed to allocate buffer for pdu\n");
 181                 rpc_free_pdu(rpc, pdu);
 182                 return -1;
 183         }
 184
 185         memcpy(pdu->outdata.data, rpc->encodebuf, pdu->outdata.size);
 186         rpc_enqueue(&rpc->outqueue, pdu);
 187
 188         return 0;
 189 }
 190
 191 int rpc_get_pdu_size(char *buf)
 192 {
 193         uint32_t size;
 194
 195         size = ntohl(*(uint32_t *)buf);
 196
 197         return (size & 0x7fffffff) + 4;
 198 }
 199
 200 static int rpc_process_reply(struct rpc_context *rpc, struct rpc_pdu *pdu, ZDR *zdr)
 201 {
 202         struct rpc_msg msg;
 203
 204         assert(rpc->magic == RPC_CONTEXT_MAGIC);
 205
 206         memset(&msg, 0, sizeof(struct rpc_msg));
 207         msg.body.rbody.reply.areply.verf = _null_auth;
 208         if (pdu->zdr_decode_bufsize > 0) {
 209                 if (pdu->zdr_decode_buf != NULL) {
 210                         free(pdu->zdr_decode_buf);
 211                 }
 212                 pdu->zdr_decode_buf = malloc(pdu->zdr_decode_bufsize);
 213                 if (pdu->zdr_decode_buf == NULL) {
 214                         rpc_set_error(rpc, "Failed to allocate memory for "
 215                                       "zdr_encode_buf in rpc_process_reply");
 216                         pdu->cb(rpc, RPC_STATUS_ERROR, "Failed to allocate "
 217                                 "buffer for decoding of ZDR reply",
 218                                 pdu->private_data);
 219                         return 0;
 220                 }
 221                 memset(pdu->zdr_decode_buf, 0, pdu->zdr_decode_bufsize);
 222         }
 223         msg.body.rbody.reply.areply.reply_data.results.where = pdu->zdr_decode_buf;
 224         msg.body.rbody.reply.areply.reply_data.results.proc  = pdu->zdr_decode_fn;
 225
 226         if (zdr_replymsg(rpc, zdr, &msg) == 0) {
 227                 rpc_set_error(rpc, "zdr_replymsg failed in rpc_process_reply: "
 228                               "%s", rpc_get_error(rpc));
 229                 pdu->cb(rpc, RPC_STATUS_ERROR, "Message rejected by server",
 230                         pdu->private_data);
 231                 if (pdu->zdr_decode_buf != NULL) {
 232                         free(pdu->zdr_decode_buf);
 233                         pdu->zdr_decode_buf = NULL;
 234                 }
 235                 return 0;
 236         }
 237         if (msg.body.rbody.stat != MSG_ACCEPTED) {
 238                 pdu->cb(rpc, RPC_STATUS_ERROR, "RPC Packet not accepted by the server", pdu->private_data);
 239                 return 0;
 240         }
 241         switch (msg.body.rbody.reply.areply.stat) {
 242         case SUCCESS:
 243                 pdu->cb(rpc, RPC_STATUS_SUCCESS, pdu->zdr_decode_buf, pdu->private_data);
 244                 break;
 245         case PROG_UNAVAIL:
 246                 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: Program not available", pdu->private_data);
 247                 break;
 248         case PROG_MISMATCH:
 249                 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: Program version mismatch", pdu->private_data);
 250                 break;
 251         case PROC_UNAVAIL:
 252                 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: Procedure not available", pdu->private_data);
 253                 break;
 254         case GARBAGE_ARGS:
 255                 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: Garbage arguments", pdu->private_data);
 256                 break;
 257         case SYSTEM_ERR:
 258                 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: System Error", pdu->private_data);
 259                 break;
 260         default:
 261                 pdu->cb(rpc, RPC_STATUS_ERROR, "Unknown rpc response from server", pdu->private_data);
 262                 break;
 263         }
 264
 265         return 0;
 266 }
 267
 268 int rpc_process_pdu(struct rpc_context *rpc, char *buf, int size)
 269 {
 270         struct rpc_pdu *pdu, *prev_pdu;
 271         ZDR zdr;
 272         int pos, recordmarker = 0;
 273         uint32_t xid;
 274         char *reasbuf = NULL;
 275
 276         assert(rpc->magic == RPC_CONTEXT_MAGIC);
 277
 278         memset(&zdr, 0, sizeof(ZDR));
 279
 280         zdrmem_create(&zdr, buf, size, ZDR_DECODE);
 281         if (rpc->is_udp == 0) {
 282                 if (zdr_int(&zdr, &recordmarker) == 0) {
 283                         rpc_set_error(rpc, "zdr_int reading recordmarker failed");
 284                         zdr_destroy(&zdr);
 285                         return -1;
 286                 }
 287                 if (!(recordmarker&0x80000000)) {
 288                         zdr_destroy(&zdr);
 289                         if (rpc_add_fragment(rpc, buf+4, size-4) != 0) {
 290                                 rpc_set_error(rpc, "Failed to queue fragment for reassembly.");
 291                                 return -1;
 292                         }
 293                         return 0;
 294                 }
 295         }
 296
 297         /* reassembly */
 298         if (recordmarker != 0 && rpc->fragments != NULL) {
 299                 struct rpc_fragment *fragment;
 300                 uint32_t total = size - 4;
 301                 char *ptr;
 302
 303                 zdr_destroy(&zdr);
 304                 for (fragment = rpc->fragments; fragment; fragment = fragment->next) {
 305                         total += fragment->size;
 306                 }
 307
 308                 reasbuf = malloc(total);
 309                 if (reasbuf == NULL) {
 310                         rpc_set_error(rpc, "Failed to reassemble PDU");
 311                         rpc_free_all_fragments(rpc);
 312                         return -1;
 313                 }
 314                 ptr = reasbuf;
 315                 for (fragment = rpc->fragments; fragment; fragment = fragment->next) {
 316                         memcpy(ptr, fragment->data, fragment->size);
 317                         ptr += fragment->size;
 318                 }
 319                 memcpy(ptr, buf + 4, size - 4);
 320                 zdrmem_create(&zdr, reasbuf, total, ZDR_DECODE);
 321                 rpc_free_all_fragments(rpc);
 322         }
 323
 324         pos = zdr_getpos(&zdr);
 325         if (zdr_int(&zdr, (int *)&xid) == 0) {
 326                 rpc_set_error(rpc, "zdr_int reading xid failed");
 327                 zdr_destroy(&zdr);
 328                 if (reasbuf != NULL) {
 329                         free(reasbuf);
 330                 }
 331                 return -1;
 332         }
 333         zdr_setpos(&zdr, pos);
 334
 335         /* Linear traverse singly-linked list, but track previous
 336          * entry for optimised removal */
 337         prev_pdu = NULL;
 338         for (pdu=rpc->waitpdu.head; pdu; pdu=pdu->next) {
 339                 if (pdu->xid != xid) {
 340                         prev_pdu = pdu;
 341                         continue;
 342                 }
 343                 if (rpc->is_udp == 0 || rpc->is_broadcast == 0) {
 344                         /* Singly-linked but we track head and tail */
 345                         if (pdu == rpc->waitpdu.head)
 346                                 rpc->waitpdu.head = pdu->next;
 347                         if (pdu == rpc->waitpdu.tail)
 348                                 rpc->waitpdu.tail = prev_pdu;
 349                         if (prev_pdu != NULL)
 350                                 prev_pdu->next = pdu->next;
 351                 }
 352                 if (rpc_process_reply(rpc, pdu, &zdr) != 0) {
 353                         rpc_set_error(rpc, "rpc_procdess_reply failed");
 354                 }
 355                 zdr_destroy(&zdr);
 356                 if (rpc->is_udp == 0 || rpc->is_broadcast == 0) {
 357                         rpc_free_pdu(rpc, pdu);
 358                 }
 359                 if (reasbuf != NULL) {
 360                         free(reasbuf);
 361                 }
 362                 return 0;
 363         }
 364         rpc_set_error(rpc, "No matching pdu found for xid:%d", xid);
 365         zdr_destroy(&zdr);
 366         if (reasbuf != NULL) {
 367                 free(reasbuf);
 368         }
 369         return -1;
 370 }
 371