src/libstrongswan/networking/streams/stream.c

   1 /*
   2  * Copyright (C) 2013 Martin Willi
   3  * Copyright (C) 2013 revosec AG
   4  *
   5  * This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License as published by the
   7  * Free Software Foundation; either version 2 of the License, or (at your
   8  * option) any later version.  See <http://www.fsf.org/copyleft/gpl.txt>.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
  12  * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  13  * for more details.
  14  */
  15
  16 #include <library.h>
  17 #include <errno.h>
  18 #include <unistd.h>
  19 #include <limits.h>
  20
  21 typedef struct private_stream_t private_stream_t;
  22
  23 /**
  24  * Private data of an stream_t object.
  25  */
  26 struct private_stream_t {
  27
  28         /**
  29          * Public stream_t interface.
  30          */
  31         stream_t public;
  32
  33         /**
  34          * Underlying socket
  35          */
  36         int fd;
  37
  38         /**
  39          * Callback if data is ready to read
  40          */
  41         stream_cb_t read_cb;
  42
  43         /**
  44          * Data for read-ready callback
  45          */
  46         void *read_data;
  47
  48         /**
  49          * Callback if write is non-blocking
  50          */
  51         stream_cb_t write_cb;
  52
  53         /**
  54          * Data for write-ready callback
  55          */
  56         void *write_data;
  57 };
  58
  59 METHOD(stream_t, read_, ssize_t,
  60         private_stream_t *this, void *buf, size_t len, bool block)
  61 {
  62         while (TRUE)
  63         {
  64                 ssize_t ret;
  65
  66                 if (block)
  67                 {
  68                         ret = read(this->fd, buf, len);
  69                 }
  70                 else
  71                 {
  72                         ret = recv(this->fd, buf, len, MSG_DONTWAIT);
  73                         if (ret == -1 && errno == EAGAIN)
  74                         {
  75                                 /* unify EGAIN and EWOULDBLOCK */
  76                                 errno = EWOULDBLOCK;
  77                         }
  78                 }
  79                 if (ret == -1 && errno == EINTR)
  80                 {       /* interrupted, try again */
  81                         continue;
  82                 }
  83                 return ret;
  84         }
  85 }
  86
  87 METHOD(stream_t, read_all, bool,
  88         private_stream_t *this, void *buf, size_t len)
  89 {
  90         ssize_t ret;
  91
  92         while (len)
  93         {
  94                 ret = read_(this, buf, len, TRUE);
  95                 if (ret < 0)
  96                 {
  97                         return FALSE;
  98                 }
  99                 if (ret == 0)
 100                 {
 101                         errno = ECONNRESET;
 102                         return FALSE;
 103                 }
 104                 len -= ret;
 105                 buf += ret;
 106         }
 107         return TRUE;
 108 }
 109
 110 METHOD(stream_t, write_, ssize_t,
 111         private_stream_t *this, void *buf, size_t len, bool block)
 112 {
 113         ssize_t ret;
 114
 115         while (TRUE)
 116         {
 117                 if (block)
 118                 {
 119                         ret = write(this->fd, buf, len);
 120                 }
 121                 else
 122                 {
 123                         ret = send(this->fd, buf, len, MSG_DONTWAIT);
 124                         if (ret == -1 && errno == EAGAIN)
 125                         {
 126                                 /* unify EGAIN and EWOULDBLOCK */
 127                                 errno = EWOULDBLOCK;
 128                         }
 129                 }
 130                 if (ret == -1 && errno == EINTR)
 131                 {       /* interrupted, try again */
 132                         continue;
 133                 }
 134                 return ret;
 135         }
 136 }
 137
 138 METHOD(stream_t, write_all, bool,
 139         private_stream_t *this, void *buf, size_t len)
 140 {
 141         ssize_t ret;
 142
 143         while (len)
 144         {
 145                 ret = write_(this, buf, len, TRUE);
 146                 if (ret < 0)
 147                 {
 148                         return FALSE;
 149                 }
 150                 if (ret == 0)
 151                 {
 152                         errno = ECONNRESET;
 153                         return FALSE;
 154                 }
 155                 len -= ret;
 156                 buf += ret;
 157         }
 158         return TRUE;
 159 }
 160
 161 /**
 162  * Watcher callback
 163  */
 164 static bool watch(private_stream_t *this, int fd, watcher_event_t event)
 165 {
 166         bool keep = FALSE;
 167         stream_cb_t cb;
 168
 169         switch (event)
 170         {
 171                 case WATCHER_READ:
 172                         cb = this->read_cb;
 173                         this->read_cb = NULL;
 174                         keep = cb(this->read_data, &this->public);
 175                         if (keep)
 176                         {
 177                                 this->read_cb = cb;
 178                         }
 179                         break;
 180                 case WATCHER_WRITE:
 181                         cb = this->write_cb;
 182                         this->write_cb = NULL;
 183                         keep = cb(this->write_data, &this->public);
 184                         if (keep)
 185                         {
 186                                 this->write_cb = cb;
 187                         }
 188                         break;
 189                 case WATCHER_EXCEPT:
 190                         break;
 191         }
 192         return keep;
 193 }
 194
 195 /**
 196  * Register watcher for stream callbacks
 197  */
 198 static void add_watcher(private_stream_t *this)
 199 {
 200         watcher_event_t events = 0;
 201
 202         if (this->read_cb)
 203         {
 204                 events |= WATCHER_READ;
 205         }
 206         if (this->write_cb)
 207         {
 208                 events |= WATCHER_WRITE;
 209         }
 210         if (events)
 211         {
 212                 lib->watcher->add(lib->watcher, this->fd, events,
 213                                                   (watcher_cb_t)watch, this);
 214         }
 215 }
 216
 217 METHOD(stream_t, on_read, void,
 218         private_stream_t *this, stream_cb_t cb, void *data)
 219 {
 220         lib->watcher->remove(lib->watcher, this->fd);
 221
 222         this->read_cb = cb;
 223         this->read_data = data;
 224
 225         add_watcher(this);
 226 }
 227
 228 METHOD(stream_t, on_write, void,
 229         private_stream_t *this, stream_cb_t cb, void *data)
 230 {
 231         lib->watcher->remove(lib->watcher, this->fd);
 232
 233         this->write_cb = cb;
 234         this->write_data = data;
 235
 236         add_watcher(this);
 237 }
 238
 239 METHOD(stream_t, get_file, FILE*,
 240         private_stream_t *this)
 241 {
 242         FILE *file;
 243         int fd;
 244
 245         /* fclose() closes the FD passed to fdopen(), so dup() it */
 246         fd = dup(this->fd);
 247         if (fd == -1)
 248         {
 249                 return NULL;
 250         }
 251         file = fdopen(fd, "w+");
 252         if (!file)
 253         {
 254                 close(fd);
 255         }
 256         return file;
 257 }
 258
 259 METHOD(stream_t, destroy, void,
 260         private_stream_t *this)
 261 {
 262         lib->watcher->remove(lib->watcher, this->fd);
 263         close(this->fd);
 264         free(this);
 265 }
 266
 267 /**
 268  * See header
 269  */
 270 stream_t *stream_create_from_fd(int fd)
 271 {
 272         private_stream_t *this;
 273
 274         INIT(this,
 275                 .public = {
 276                         .read = _read_,
 277                         .read_all = _read_all,
 278                         .on_read = _on_read,
 279                         .write = _write_,
 280                         .write_all = _write_all,
 281                         .on_write = _on_write,
 282                         .get_file = _get_file,
 283                         .destroy = _destroy,
 284                 },
 285                 .fd = fd,
 286         );
 287
 288         return &this->public;
 289 }
 290
 291 /**
 292  * See header
 293  */
 294 int stream_parse_uri_unix(char *uri, struct sockaddr_un *addr)
 295 {
 296         if (!strpfx(uri, "unix://"))
 297         {
 298                 return -1;
 299         }
 300         uri += strlen("unix://");
 301
 302         memset(addr, 0, sizeof(*addr));
 303         addr->sun_family = AF_UNIX;
 304         strncpy(addr->sun_path, uri, sizeof(addr->sun_path));
 305         addr->sun_path[sizeof(addr->sun_path)-1] = '\0';
 306
 307         return offsetof(struct sockaddr_un, sun_path) + strlen(addr->sun_path);
 308 }
 309
 310 /**
 311  * See header
 312  */
 313 stream_t *stream_create_unix(char *uri)
 314 {
 315         struct sockaddr_un addr;
 316         int len, fd;
 317
 318         len = stream_parse_uri_unix(uri, &addr);
 319         if (len == -1)
 320         {
 321                 DBG1(DBG_NET, "invalid stream URI: '%s'", uri);
 322                 return NULL;
 323         }
 324         fd = socket(AF_UNIX, SOCK_STREAM, 0);
 325         if (fd < 0)
 326         {
 327                 DBG1(DBG_NET, "opening socket '%s' failed: %s", uri, strerror(errno));
 328                 return NULL;
 329         }
 330         if (connect(fd, (struct sockaddr*)&addr, len) < 0)
 331         {
 332                 DBG1(DBG_NET, "connecting to '%s' failed: %s", uri, strerror(errno));
 333                 close(fd);
 334                 return NULL;
 335         }
 336         return stream_create_from_fd(fd);
 337 }
 338
 339 /**
 340  * See header.
 341  */
 342 int stream_parse_uri_tcp(char *uri, struct sockaddr *addr)
 343 {
 344         char *pos, buf[128];
 345         host_t *host;
 346         u_long port;
 347         int len;
 348
 349         if (!strpfx(uri, "tcp://"))
 350         {
 351                 return -1;
 352         }
 353         uri += strlen("tcp://");
 354         pos = strrchr(uri, ':');
 355         if (!pos)
 356         {
 357                 return -1;
 358         }
 359         if (*uri == '[' && pos > uri && *(pos - 1) == ']')
 360         {
 361                 /* IPv6 URI */
 362                 snprintf(buf, sizeof(buf), "%.*s", (int)(pos - uri - 2), uri + 1);
 363         }
 364         else
 365         {
 366                 snprintf(buf, sizeof(buf), "%.*s", (int)(pos - uri), uri);
 367         }
 368         port = strtoul(pos + 1, &pos, 10);
 369         if (port == ULONG_MAX || *pos || port > 65535)
 370         {
 371                 return -1;
 372         }
 373         host = host_create_from_dns(buf, AF_UNSPEC, port);
 374         if (!host)
 375         {
 376                 return -1;
 377         }
 378         len = *host->get_sockaddr_len(host);
 379         memcpy(addr, host->get_sockaddr(host), len);
 380         host->destroy(host);
 381         return len;
 382 }
 383
 384 /**
 385  * See header
 386  */
 387 stream_t *stream_create_tcp(char *uri)
 388 {
 389         union {
 390                 struct sockaddr_in in;
 391                 struct sockaddr_in6 in6;
 392                 struct sockaddr sa;
 393         } addr;
 394         int fd, len;
 395
 396         len = stream_parse_uri_tcp(uri, &addr.sa);
 397         if (len == -1)
 398         {
 399                 DBG1(DBG_NET, "invalid stream URI: '%s'", uri);
 400                 return NULL;
 401         }
 402         fd = socket(addr.sa.sa_family, SOCK_STREAM, 0);
 403         if (fd < 0)
 404         {
 405                 DBG1(DBG_NET, "opening socket '%s' failed: %s", uri, strerror(errno));
 406                 return NULL;
 407         }
 408         if (connect(fd, &addr.sa, len))
 409         {
 410                 DBG1(DBG_NET, "connecting to '%s' failed: %s", uri, strerror(errno));
 411                 close(fd);
 412                 return NULL;
 413         }
 414         return stream_create_from_fd(fd);
 415 }