send_common.c

   1 /* Copyright (C) 2005 Andre Noll <maan@tuebingen.mpg.de>, see file COPYING. */
   2
   3 /** \file send_common.c Functions used by more than one paraslash sender. */
   4
   5 #include <netinet/in.h>
   6 #include <sys/socket.h>
   7 #include <regex.h>
   8 #include <osl.h>
   9 #include <arpa/inet.h>
  10 #include <sys/un.h>
  11 #include <netdb.h>
  12 #include <lopsub.h>
  13
  14 #include "para.h"
  15 #include "error.h"
  16 #include "string.h"
  17 #include "fd.h"
  18 #include "net.h"
  19 #include "list.h"
  20 #include "afh.h"
  21 #include "afs.h"
  22 #include "server.h"
  23 #include "acl.h"
  24 #include "send.h"
  25 #include "close_on_fork.h"
  26 #include "chunk_queue.h"
  27 #include "sched.h"
  28 #include "vss.h"
  29
  30 /** Clients will be kicked if there are more than that many bytes pending. */
  31 #define MAX_CQ_BYTES 40000
  32
  33 /**
  34  * Shut down a client connected to a paraslash sender.
  35  *
  36  * \param sc The client to shut down.
  37  * \param ss The sender whose clients are to be shut down.
  38  *
  39  * Close the file descriptor given by \a sc, remove it from the close-on-fork
  40  * list, destroy the chunk queue of this client, delete the client from the
  41  * list of connected clients and free the sender_client struct.
  42  *
  43  * \sa \ref shutdown_clients().
  44  */
  45 void shutdown_client(struct sender_client *sc, struct sender_status *ss)
  46 {
  47         PARA_INFO_LOG("shutting down %s on fd %d\n", sc->name, sc->fd);
  48         free(sc->name);
  49         close(sc->fd);
  50         del_close_on_fork_list(sc->fd);
  51         cq_destroy(sc->cq);
  52         list_del(&sc->node);
  53         free(sc->private_data);
  54         free(sc);
  55         ss->num_clients--;
  56 }
  57
  58 /**
  59  * Shut down all clients connected to a paraslash sender.
  60  *
  61  * \param ss The sender whose clients are to be shut down.
  62  *
  63  * This just loops over all connected clients and calls shutdown_client()
  64  * for each client.
  65  */
  66 void shutdown_clients(struct sender_status *ss)
  67 {
  68         struct sender_client *sc, *tmp;
  69         list_for_each_entry_safe(sc, tmp, &ss->client_list, node)
  70                 shutdown_client(sc, ss);
  71 }
  72
  73 /**
  74  * Try to empty the chunk queue for this fd.
  75  *
  76  * \param fd The file descriptor.
  77  * \param cq The list of queued chunks.
  78  *
  79  * \return Negative on errors, zero if not everything was sent, one otherwise.
  80  */
  81 int send_queued_chunks(int fd, struct chunk_queue *cq)
  82 {
  83         struct queued_chunk *qc;
  84         while ((qc = cq_peek(cq))) {
  85                 const char *buf;
  86                 size_t len;
  87                 int ret;
  88
  89                 cq_get(qc, &buf, &len);
  90                 ret = xwrite(fd, buf, len);
  91                 if (ret < 0)
  92                         return ret;
  93                 cq_update(cq, ret);
  94                 if (ret != len)
  95                         return 0;
  96                 cq_dequeue(cq);
  97         }
  98         return 1;
  99 }
 100
 101 /**
 102  * Initialize a struct sender status.
 103  *
 104  * \param ss The struct to initialize.
 105  * \param acl_opt_result Contains array of --{http|dccp}-access arguments.
 106  * \param port The tcp or dccp port to listen on.
 107  * \param max_clients The maximal number of simultaneous connections.
 108  * \param default_deny Whether a blacklist should be used for access control.
 109  */
 110 void init_sender_status(struct sender_status *ss,
 111                 const struct lls_opt_result *acl_opt_result, int port,
 112                 int max_clients, int default_deny)
 113 {
 114         int i;
 115
 116         ss->listen_fd = -1;
 117         INIT_LIST_HEAD(&ss->client_list);
 118         ss->port = port;
 119
 120         /* Initialize an access control list */
 121         INIT_LIST_HEAD(&ss->acl);
 122         for (i = 0; i < lls_opt_given(acl_opt_result); i++) {
 123                 const char *arg = lls_string_val(i, acl_opt_result);
 124                 char addr[16];
 125                 int mask;
 126                 if (!parse_cidr(arg, addr, sizeof(addr), &mask))
 127                         PARA_WARNING_LOG("ACL syntax error: %s, ignoring\n",
 128                                 arg);
 129                 else
 130                         acl_add_entry(&ss->acl, addr, mask);
 131         }
 132         ss->num_clients = 0;
 133         ss->max_clients = max_clients;
 134         ss->default_deny = default_deny;
 135 }
 136
 137 /**
 138  * Return a string containing the current status of a sender.
 139  *
 140  * \param ss The sender.
 141  * \param name Used for printing the header line.
 142  *
 143  * \return The string printed in the "si" command.
 144  */
 145 char *generic_sender_status(struct sender_status *ss, const char *name)
 146 {
 147         char *clnts = NULL, *ret;
 148         struct sender_client *sc, *tmp_sc;
 149
 150         char *acl_contents = acl_get_contents(&ss->acl);
 151         list_for_each_entry_safe(sc, tmp_sc, &ss->client_list, node) {
 152                 char *tmp = make_message("%s%s ", clnts? clnts : "", sc->name);
 153                 free(clnts);
 154                 clnts = tmp;
 155         }
 156         ret = make_message(
 157                 "status: %s\n"
 158                 "port: %s\n"
 159                 "number of connected clients: %d\n"
 160                 "maximal number of clients: %d%s\n"
 161                 "connected clients: %s\n"
 162                 "access %s list: %s\n",
 163                 (ss->listen_fd >= 0)? "on" : "off",
 164                 stringify_port(ss->port, strcmp(name, "http") ? "dccp" : "tcp"),
 165                 ss->num_clients,
 166                 ss->max_clients,
 167                 ss->max_clients > 0? "" : " (unlimited)",
 168                 clnts? clnts : "(none)",
 169                 ss->default_deny? "allow" : "deny",
 170                 acl_contents? acl_contents : "(empty)"
 171         );
 172         free(acl_contents);
 173         free(clnts);
 174         return ret;
 175 }
 176
 177 /**
 178  * Allow connections from the given range of IP addresses.
 179  *
 180  * \param scd Contains the IP and the netmask.
 181  * \param ss The sender.
 182  *
 183  * \sa \ref generic_com_deny().
 184  */
 185 void generic_com_allow(struct sender_command_data *scd,
 186                 struct sender_status *ss)
 187 {
 188         acl_allow(scd->host, scd->netmask, &ss->acl, ss->default_deny);
 189 }
 190
 191 /**
 192  * Deny connections from the given range of IP addresses.
 193  *
 194  * \param scd see \ref generic_com_allow().
 195  * \param ss see \ref generic_com_allow().
 196  *
 197  * \sa \ref generic_com_allow().
 198  */
 199 void generic_com_deny(struct sender_command_data *scd,
 200                 struct sender_status *ss)
 201 {
 202         acl_deny(scd->host, scd->netmask, &ss->acl, ss->default_deny);
 203 }
 204
 205 /**
 206  * Activate a paraslash sender.
 207  *
 208  * \param ss The sender to activate.
 209  * \param protocol The symbolic name of the transport-layer protocol.
 210  *
 211  * \return Standard.
 212  */
 213 int generic_com_on(struct sender_status *ss, unsigned protocol)
 214 {
 215         int fd, ret;
 216
 217         if (ss->listen_fd >= 0)
 218                 return 1;
 219         ret = para_listen_simple(protocol, ss->port);
 220         if (ret < 0)
 221                 return ret;
 222         fd = ret;
 223         ret = mark_fd_nonblocking(fd);
 224         if (ret < 0) {
 225                 close(fd);
 226                 return ret;
 227         }
 228         add_close_on_fork_list(fd);
 229         ss->listen_fd = ret;
 230         return 1;
 231 }
 232
 233 /**
 234  * Deactivate a paraslash sender.
 235  *
 236  * Shutdown all connected clients and stop listening on the TCP/DCCP socket.
 237  *
 238  * \param ss The sender to deactivate.
 239  *
 240  * \sa \ref del_close_on_fork_list(), \ref shutdown_clients().
 241  */
 242 void generic_com_off(struct sender_status *ss)
 243 {
 244         if (ss->listen_fd < 0)
 245                 return;
 246         PARA_NOTICE_LOG("closing port %d\n", ss->port);
 247         close(ss->listen_fd);
 248         del_close_on_fork_list(ss->listen_fd);
 249         shutdown_clients(ss);
 250         ss->listen_fd = -1;
 251 }
 252
 253 /**
 254  * Accept a connection on the socket this server is listening on.
 255  *
 256  * \param ss The sender whose listening fd is ready for reading.
 257  * \param rfds Passed to para_accept(),
 258  *
 259  * This calls para_accept() and performs the following actions on the resulting
 260  * file descriptor fd:
 261  *
 262  *      - Checks whether the maximal number of connections are exceeded.
 263  *      - Sets \a fd to nonblocking mode.
 264  *      - Checks the acl of the sender to find out whether connections
 265  *        are allowed from the IP of the connecting peer.
 266  *      - Increases the number of connections for this sender.
 267  *      - Creates and initializes a new chunk queue for queuing network
 268  *        packets that can not be sent immediately.
 269  *      - Allocates a new struct sender_client and fills in its \a fd, \a cq
 270  *        and \a name members.
 271  *      - Adds \a fd to the list of connected clients for this sender.
 272  *      - Adds \a fd to the list of file descriptors that should be closed
 273  *        in the child process when the server calls fork().
 274  *
 275  * \return A pointer to the allocated sender_client structure on success, \p
 276  * NULL on errors.
 277  *
 278  * \sa \ref para_accept(), \ref mark_fd_nonblocking(), \ref acl_check_access(),
 279  * \ref cq_new(), \ref add_close_on_fork_list().
 280  */
 281 struct sender_client *accept_sender_client(struct sender_status *ss, fd_set *rfds)
 282 {
 283         struct sender_client *sc;
 284         int fd, ret;
 285
 286         if (ss->listen_fd < 0)
 287                 return NULL;
 288         ret = para_accept(ss->listen_fd, rfds, NULL, 0, &fd);
 289         if (ret < 0)
 290                 PARA_ERROR_LOG("%s\n", para_strerror(-ret));
 291         if (ret <= 0)
 292                 return NULL;
 293         ret = -E_MAX_CLIENTS;
 294         if (ss->max_clients > 0 && ss->num_clients >= ss->max_clients)
 295                 goto err_out;
 296         ret = mark_fd_nonblocking(fd);
 297         if (ret < 0)
 298                 goto err_out;
 299         ret = acl_check_access(fd, &ss->acl, ss->default_deny);
 300         if (ret < 0)
 301                 goto err_out;
 302         ss->num_clients++;
 303         sc = para_calloc(sizeof(*sc));
 304         sc->fd = fd;
 305         sc->name = para_strdup(remote_name(fd));
 306         sc->cq = cq_new(MAX_CQ_BYTES);
 307         para_list_add(&sc->node, &ss->client_list);
 308         add_close_on_fork_list(fd);
 309         PARA_INFO_LOG("accepted client #%d: %s (fd %d)\n", ss->num_clients,
 310                 sc->name, fd);
 311         return sc;
 312 err_out:
 313         PARA_WARNING_LOG("%s\n", para_strerror(-ret));
 314         close(fd);
 315         return NULL;
 316 }
 317
 318 /**
 319  * Get the generic help text.
 320  *
 321  * \return A dynamically allocated string containing the help text for
 322  * a paraslash sender.
 323  */
 324 char *generic_sender_help(void)
 325 {
 326         return make_message(
 327                 "usage: {on|off}\n"
 328                 "usage: {allow|deny} IP[/netmask]\n"
 329                 "       where mask defaults to 32\n"
 330                 "example: allow 192.168.0.1/24\n"
 331         );
 332 }
 333
 334 static int parse_fec_parms(const char *arg, struct sender_command_data *scd)
 335 {
 336         int32_t val;
 337         char *a = para_strdup(arg),
 338              *b = strchr(a, ':'),
 339              *c = strrchr(a, ':');
 340         int ret = -E_COMMAND_SYNTAX;
 341
 342         if (!b || !c)
 343                 goto out;
 344         *b = *c = '\0';
 345
 346         ret = para_atoi32(a, &val);
 347         if (ret < 0)
 348                 goto out;
 349
 350         /* optional max_slice_bytes (0 means "use MTU") */
 351         if (b == c) {
 352                 scd->max_slice_bytes = 0;
 353         } else {
 354                 if (val < 0 || val > 65535)
 355                         goto fec_einval;
 356                 scd->max_slice_bytes = val;
 357
 358                 ret = para_atoi32(b + 1, &val);
 359                 if (ret < 0)
 360                         goto out;
 361         }
 362
 363         /* k = data_slices_per_group */
 364         if (val < 0 || val > 255)
 365                 goto fec_einval;
 366         scd->data_slices_per_group = val;
 367
 368         /* n = slices_per_group */
 369         ret = para_atoi32(c + 1, &val);
 370         if (ret < 0)
 371                 goto out;
 372         if (val < 0 || val < scd->data_slices_per_group)
 373                 goto fec_einval;
 374         scd->slices_per_group = val;
 375         ret = 0;
 376 out:
 377         free(a);
 378         return ret;
 379 fec_einval:
 380         ret = -ERRNO_TO_PARA_ERROR(EINVAL);
 381         goto out;
 382 }
 383
 384 /**
 385  * Parse a FEC URL string.
 386  *
 387  * \param arg the URL string to parse.
 388  * \param scd The structure containing host, port and the FEC parameters.
 389  *
 390  * \return Standard.
 391  *
 392  * A FEC URL consists of an ordinary URL string according to RFC 3986,
 393  * optionally followed by a slash and the three FEC parameters slice_size,
 394  * data_slices_per_group and slices_per_group. The three FEC parameters are
 395  * separated by colons.
 396  *
 397  * \sa \ref parse_url().
 398  */
 399 int parse_fec_url(const char *arg, struct sender_command_data *scd)
 400 {
 401         char *a = para_strdup(arg), *p = strchr(a, '/');
 402         int ret = 0;
 403
 404         /* default fec parameters */
 405         scd->max_slice_bytes       = 0;
 406         scd->data_slices_per_group = 14;
 407         scd->slices_per_group      = 16;
 408
 409         if (p) {
 410                 *p = '\0';
 411                 ret = parse_fec_parms(p + 1, scd);
 412                 if (ret < 0)
 413                         goto out;
 414         }
 415         if (!parse_url(a, scd->host, sizeof(scd->host), &scd->port))
 416                 ret = -ERRNO_TO_PARA_ERROR(EINVAL);
 417 out:
 418         free(a);
 419         return ret;
 420 }