/* SPDX-License-Identifier: GPL-2.0 */ /** \file http_send.c paraslash's http sender */ #include #include #include #include #include #include #include #include "server.lsg.h" #include "para.h" #include "error.h" #include "string.h" #include "afh.h" #include "net.h" #include "list.h" #include "server.h" #include "http.h" #include "sched.h" #include "send.h" #include "vss.h" #include "fd.h" /* Describes one entry in the blacklist/whitelist of the http sender. */ struct access_info { /* The address to be black/whitelisted. */ struct in_addr addr; /* The netmask for this entry. */ unsigned netmask; /* The position of this entry in the acl. */ struct list_head node; }; /* * Pretty-print a host/port pair. * * This function calls parse_url() to check the syntax of the input string * and returns "?" on errors. Otherwise, if the url string contains a port, * a copy of this string is returned. If no port was supplied, a colon and * the default port are appended. * * In all cases the returned string is a allocated with malloc(3) and has * to be freed by the caller. */ static __malloc char *format_url(const char *url, int default_port) { char host[MAX_HOSTLEN]; int url_port; assert(url); if (!parse_url(url, host, sizeof(host), &url_port)) return make_message("?"); if (url_port < 0) return make_message("%s:%d", url, default_port); else return para_strdup(url); } /* Return whether addr_1 matches addr_2 in the first netmask bits. */ static bool v4_addr_match(uint32_t addr_1, uint32_t addr_2, uint8_t netmask) { uint32_t mask = ~0U; if (netmask == 0) /* avoid 32-bit shift, which is undefined in C. */ return true; if (netmask < 32) mask <<= (32 - netmask); return (htonl(addr_1) & mask) == (htonl(addr_2) & mask); } /* * Check whether the peer name of a given fd is allowed by an acl. * Returns positive if the peer is permitted, -E_ACL_PERM otherwise. */ static int acl_check_access(int fd, struct list_head *acl, bool default_deny) { struct access_info *ai, *tmp; struct sockaddr_storage ss; socklen_t sslen = sizeof(ss); struct in_addr v4_addr; if (getpeername(fd, (struct sockaddr *)&ss, &sslen) < 0) { PARA_ERROR_LOG("Can not determine peer address: %s\n", strerror(errno)); goto no_match; } extract_v4_addr(&ss, &v4_addr); if (!v4_addr.s_addr) goto no_match; list_for_each_entry_safe(ai, tmp, acl, node) if (v4_addr_match(v4_addr.s_addr, ai->addr.s_addr, ai->netmask)) return default_deny? 1 : -E_ACL_PERM; no_match: return default_deny? -E_ACL_PERM : 1; } static void acl_add_entry(struct list_head *acl, char *addr, int netmask) { struct access_info *ai = alloc(sizeof(struct access_info)); inet_pton(AF_INET, addr, &ai->addr); ai->netmask = netmask; PARA_INFO_LOG("adding %s/%u to access list\n", addr, ai->netmask); para_list_add(&ai->node, acl); } static void acl_del_entry(struct list_head *acl, char *addr, unsigned netmask) { struct access_info *ai, *tmp; struct in_addr to_delete; PARA_INFO_LOG("removing entries matching %s/%u\n", addr, netmask); inet_pton(AF_INET, addr, &to_delete); list_for_each_entry_safe(ai, tmp, acl, node) { if (v4_addr_match(to_delete.s_addr, ai->addr.s_addr, PARA_MIN(netmask, ai->netmask))) { char dst[INET_ADDRSTRLEN + 1]; const char *p = inet_ntop(AF_INET, &ai->addr.s_addr, dst, sizeof(dst)); if (p) PARA_DEBUG_LOG("removing %s/%u\n", p, ai->netmask); list_del(&ai->node); free(ai); } } } /* * Compute a string containing the contents of an acl. Returns a dynamically * allocated string or NULL if the acl is empty. */ static char *acl_get_contents(struct list_head *acl) { struct access_info *ai, *tmp_ai; char *ret = NULL; list_for_each_entry_safe(ai, tmp_ai, acl, node) { char *tmp = make_message("%s%s/%u ", ret? ret : "", inet_ntoa(ai->addr), ai->netmask); free(ret); ret = tmp; } return ret; } /* Permit access for a range of IP addresses. */ static void acl_allow(char *addr, int netmask, struct list_head *acl, bool default_deny) { if (default_deny) acl_add_entry(acl, addr, netmask); else acl_del_entry(acl, addr, netmask); } /* Deny access for a range of IP addresses. */ static void acl_deny(char *addr, int netmask, struct list_head *acl, bool default_deny) { acl_allow(addr, netmask, acl, !default_deny); } /* * The http sender uses chunk queues to deal with laggy connections. If * a chunk can not be sent without blocking, it is put into a queue. Chunk * queues are cheap because only a reference to the audio file data is stored * in the queue. */ struct chunk_queue { /* The list of pending chunks for this client. */ struct list_head q; /* The number of pending bytes for this client. */ unsigned long num_pending; /* More than that many bytes in the queue is considered an error. */ unsigned long max_pending; }; struct queued_chunk { /* Pointer to the data to be queued. */ const char *buf; /* The number of bytes of this chunk. */ size_t num_bytes; /* Position of the chunk in the chunk queue. */ struct list_head node; }; /* Add a chunk to the given queue. */ static int cq_enqueue(struct chunk_queue *cq, const char *buf, size_t num_bytes) { struct queued_chunk *qc; if (cq->num_pending + num_bytes > cq->max_pending) return -E_QUEUE; qc = alloc(sizeof(struct queued_chunk)); cq->num_pending += num_bytes; qc->buf = buf; qc->num_bytes = num_bytes; list_add_tail(&qc->node, &cq->q); PARA_DEBUG_LOG("%lu bytes queued for %p\n", cq->num_pending, &cq->q); return 1; } /* * Lookup the next chunk in the queue. Returns the next queued chunk, * or NULL if there is no chunk available. */ static struct queued_chunk *cq_peek(struct chunk_queue *cq) { if (list_empty(&cq->q)) return NULL; return list_entry(cq->q.next, struct queued_chunk, node); } /* Remove the current chunk from the queue. */ static void cq_dequeue(struct chunk_queue *cq) { struct queued_chunk *qc = cq_peek(cq); assert(qc); assert(cq->num_pending >= qc->num_bytes); cq->num_pending -= qc->num_bytes; list_del(&qc->node); free(qc); } /* Update the number of bytes sent for the current queued chunk. */ static void cq_update(struct chunk_queue *cq, size_t sent) { struct queued_chunk *qc = cq_peek(cq); assert(qc); qc->num_bytes -= sent; qc->buf += sent; cq->num_pending -= sent; } /* Get a pointer to the next queued chunk. */ static void cq_get(struct queued_chunk *qc, const char **buf, size_t *num_bytes) { *buf = qc->buf; *num_bytes = qc->num_bytes; } /* Allocate and initialize a chunk queue. */ static struct chunk_queue *cq_new(size_t max_pending) { struct chunk_queue *cq = alloc(sizeof(*cq)); init_list_head(&cq->q); cq->max_pending = max_pending; cq->num_pending = 0; return cq; } /* Deallocate all resources of a queue. */ static void cq_destroy(struct chunk_queue *cq) { struct queued_chunk *qc, *tmp; list_for_each_entry_safe(qc, tmp, &cq->q, node) { list_del(&qc->node); free(qc); } free(cq); } /* Clients will be kicked if there are more than that many bytes pending. */ #define MAX_CQ_BYTES 40000 /* Describes the current status of the http sender. */ struct sender_status { /* Number of sockets to listen on, size of the two arrays below. */ unsigned num_listen_fds; /* Derived from --http-listen-address. */ char **listen_addresses; /* Default TCP port number for addresses w/o port. */ int default_port; /* The socket fd(s) this sender is listening on. */ int *listen_fds; /* The current number of simultaneous connections. */ int num_clients; /* The maximal number of simultaneous connections. */ int max_clients; /* Whether the access control list is a whitelist. */ bool default_deny; /* The whitelist/blacklist. */ struct list_head acl; /* The list of connected clients. */ struct list_head client_list; }; /* The possible states of a client from the server's POV. */ enum http_client_status { /* We accepted the connection on the tcp socket. */ HTTP_CONNECTED, /* Successfully received the get request. */ HTTP_GOT_GET_REQUEST, /* Connection is ready for sending audio data. */ HTTP_STREAMING, /* We didn't receive a valid get request. */ HTTP_INVALID_GET_REQUEST }; /* For each connected client, a structure of this type is maintained. */ struct private_http_sender_data { /* The current state of this client. */ enum http_client_status status; /* True if audio file header has been sent. */ bool header_sent; /* The list of pending chunks for this client. */ struct chunk_queue *cq; }; static struct sender_status http_sender_status, *hss = &http_sender_status; /* Iterate over all listening addresses of the http sender. */ #define FOR_EACH_LISTEN_FD(_n) for (_n = 0; _n < (hss)->num_listen_fds; _n++) /* * Shut down a connected client of the http sender. * * Close the file descriptor, remove it from the close-on-fork list, * destroy the chunk queue of this client, delete the client from the list * of connected clients and free the sender_client struct. */ static void shutdown_client(struct sender_client *sc) { struct private_http_sender_data *phsd = sc->private_data; if (!process_is_command_handler()) { PARA_INFO_LOG("shutting down %s on fd %d\n", sc->name, sc->fd); close(sc->fd); del_close_on_fork_list(sc->fd); } free(sc->name); cq_destroy(phsd->cq); list_del(&sc->node); free(sc->private_data); free(sc); hss->num_clients--; } /* Loop over all connected clients and call shutdown_client() for each. */ static void http_shutdown_clients(void) { struct sender_client *sc, *tmp; list_for_each_entry_safe(sc, tmp, &hss->client_list, node) shutdown_client(sc); } /* * Try to empty the chunk queue for this fd. Returns negative on errors, * zero if not everything was sent, one otherwise. */ static int send_queued_chunks(int fd, struct chunk_queue *cq) { struct queued_chunk *qc; while ((qc = cq_peek(cq))) { const char *buf; size_t len; int ret; cq_get(qc, &buf, &len); ret = xwrite(fd, buf, len); if (ret < 0) return ret; cq_update(cq, ret); if (ret != len) return 0; cq_dequeue(cq); } return 1; } /* * Return a string containing the current status of a sender. The returned * string is printed by the "sender http status" command. */ static __malloc char *http_status(void) { char *clnts = NULL, *ret, *addr = NULL; struct sender_client *sc, *tmp_sc; unsigned n; char *acl_contents = acl_get_contents(&hss->acl); list_for_each_entry_safe(sc, tmp_sc, &hss->client_list, node) { char *tmp = make_message("%s%s ", clnts? clnts : "", sc->name); free(clnts); clnts = tmp; } FOR_EACH_LISTEN_FD(n) { char *url = format_url(hss->listen_addresses[n], hss->default_port); char *tmp = make_message("%s%s%s (fd %d)", addr? addr : "", addr? ", " : "", url, hss->listen_fds[n]); free(url); free(addr); addr = tmp; } ret = make_message( "listening address(es): %s\n" "default port: %s\n" "number of connected clients: %d\n" "maximal number of clients: %d%s\n" "connected clients: %s\n" "access %s list: %s\n", addr, stringify_port(hss->default_port, "tcp"), hss->num_clients, hss->max_clients, hss->max_clients > 0? "" : " (unlimited)", clnts? clnts : "(none)", hss->default_deny? "allow" : "deny", acl_contents? acl_contents : "(empty)" ); free(acl_contents); free(clnts); return ret; } /* Allow connections from the given range of IP addresses. */ static int http_com_allow(struct sender_command_data *scd) { acl_allow(scd->host, scd->netmask, &hss->acl, hss->default_deny); return 1; } /* Deny connections from the given range of IP addresses. */ static int http_com_deny(struct sender_command_data *scd) { acl_deny(scd->host, scd->netmask, &hss->acl, hss->default_deny); return 1; } /* * This opens a passive TCP socket, sets the resulting file descriptor to * nonblocking mode and adds it to the close on fork list. Errors are logged * but otherwise ignored. */ static void start_listening(void) { int ret; unsigned n; FOR_EACH_LISTEN_FD(n) { if (hss->listen_fds[n] >= 0) continue; ret = para_listen(IPPROTO_TCP, hss->listen_addresses[n], hss->default_port); if (ret < 0) { char *url = format_url(hss->listen_addresses[n], hss->default_port); PARA_ERROR_LOG("could not listen on TCP %s: %s\n", url, para_strerror(-ret)); free(url); continue; } hss->listen_fds[n] = ret; ret = mark_fd_nonblocking(hss->listen_fds[n]); if (ret < 0) { char *url = format_url(hss->listen_addresses[n], hss->default_port); PARA_ERROR_LOG("could not set TCP socket fd for %s to " "nonblocking mode: %s\n", url, para_strerror(-ret)); free(url); close(hss->listen_fds[n]); hss->listen_fds[n] = -1; continue; } add_close_on_fork_list(hss->listen_fds[n]); } } /* Shutdown all connected clients and stop listening on the TCP socket. */ static int http_com_off(__a_unused struct sender_command_data *scd) { unsigned n; FOR_EACH_LISTEN_FD(n) { if (hss->listen_fds[n] < 0) return 1; close(hss->listen_fds[n]); del_close_on_fork_list(hss->listen_fds[n]); http_shutdown_clients(); hss->listen_fds[n] = -1; } return 1; } /* * This accepts incoming connections on any of the listening sockets of the * server. If there is a connection pending, the function * * - Checks whether the maximal number of connections are exceeded. * - Sets fd to nonblocking mode. * - Checks the acl of the sender to find out whether connections * are allowed from the IP of the connecting peer. * - Increases the number of connections for this sender. * - Creates and initializes a new chunk queue for queuing network * packets that can not be sent immediately. * - Allocates a new struct sender_client and fills in its fd, cq * and name members. * - Adds fd to the list of connected clients for this sender. * - Adds fd to the list of file descriptors that should be closed * in the child process when the server calls fork(). * * Returns a pointer to the allocated sender_client structure on success, * NULL on errors. */ static struct sender_client *accept_sender_client(void) { struct sender_client *sc; int fd, ret; unsigned n; FOR_EACH_LISTEN_FD(n) { struct private_http_sender_data *phsd; if (hss->listen_fds[n] < 0) continue; ret = para_accept(hss->listen_fds[n], NULL, 0, &fd); if (ret < 0) goto warn; if (ret == 0) continue; ret = -E_MAX_CLIENTS; if (hss->max_clients > 0 && hss->num_clients >= hss->max_clients) goto close_fd_and_warn; ret = mark_fd_nonblocking(fd); if (ret < 0) goto close_fd_and_warn; ret = acl_check_access(fd, &hss->acl, hss->default_deny); if (ret < 0) goto close_fd_and_warn; hss->num_clients++; sc = zalloc(sizeof(*sc)); sc->fd = fd; sc->name = para_strdup(remote_name(fd)); phsd = alloc(sizeof(*phsd)); phsd->status = HTTP_CONNECTED; phsd->header_sent = false; phsd->cq = cq_new(MAX_CQ_BYTES); sc->private_data = phsd; para_list_add(&sc->node, &hss->client_list); add_close_on_fork_list(fd); PARA_INFO_LOG("accepted client #%d: %s (fd %d)\n", hss->num_clients, sc->name, fd); return sc; close_fd_and_warn: close(fd); warn: PARA_WARNING_LOG("%s\n", para_strerror(-ret)); } return NULL; } /* Returns a dynamically allocated string */ static __malloc char *http_sender_help(void) { return make_message( "usage: {on|off}\n" "usage: {allow|deny} IP[/netmask]\n" " where mask defaults to 32\n" "example: allow 192.168.0.1/24\n" ); } static int send_msg(struct sender_client *sc, const char *msg) { int ret = write_buffer(sc->fd, msg); if (ret < 0) shutdown_client(sc); return ret; } static void http_shutdown(void) { int i; http_shutdown_clients(); /* * Since default_deny is false, the ACL is considered a blacklist. A * netmask of zero matches any IP address, so this call empties the ACL. */ acl_allow("0.0.0.0", 0 /* netmask */, &hss->acl, false /* default_deny */); free(hss->listen_fds); FOR_EACH_LISTEN_FD(i) free(hss->listen_addresses[i]); free(hss->listen_addresses); } static int queue_chunk_or_shutdown(struct sender_client *sc, const char *buf, size_t num_bytes) { struct private_http_sender_data *phsd = sc->private_data; int ret = cq_enqueue(phsd->cq, buf, num_bytes); if (ret < 0) shutdown_client(sc); return ret; } /* * Send one chunk of audio data to a connected client. On errors, the * client is shut down. If only a part of the buffer could be written, * the remainder is put into the chunk queue for that client. */ static void http_send_chunk(struct sender_client *sc, long unsigned current_chunk, const char *buf, size_t len, const char *header_buf, size_t header_len) { int ret; struct private_http_sender_data *phsd = sc->private_data; if (!phsd->header_sent && current_chunk) { if (header_buf && header_len > 0) { ret = queue_chunk_or_shutdown(sc, header_buf, header_len); if (ret < 0) goto out; } } phsd->header_sent = true; ret = send_queued_chunks(sc->fd, phsd->cq); if (ret < 0) { shutdown_client(sc); goto out; } if (!len) goto out; if (!ret) { /* still data left in the queue */ ret = queue_chunk_or_shutdown(sc, buf, len); goto out; } ret = xwrite(sc->fd, buf, len); if (ret < 0) { shutdown_client(sc); goto out; } if (ret != len) ret = queue_chunk_or_shutdown(sc, buf + ret, len - ret); out: if (ret < 0) PARA_NOTICE_LOG("%s\n", para_strerror(-ret)); } static void http_send(long unsigned current_chunk, const char *buf, size_t len, const char *header_buf, size_t header_len) { struct sender_client *sc, *tmp; list_for_each_entry_safe(sc, tmp, &hss->client_list, node) { struct private_http_sender_data *phsd = sc->private_data; if (phsd->status == HTTP_STREAMING) http_send_chunk(sc, current_chunk, buf, len, header_buf, header_len); } } static void http_post_monitor(void) { struct sender_client *sc, *tmp; struct private_http_sender_data *phsd; int ret; list_for_each_entry_safe(sc, tmp, &hss->client_list, node) { phsd = sc->private_data; switch (phsd->status) { case HTTP_STREAMING: /* nothing to do */ break; case HTTP_CONNECTED: /* need to recv get request */ ret = read_and_compare(sc->fd, HTTP_GET_MSG); if (ret < 0) phsd->status = HTTP_INVALID_GET_REQUEST; else if (ret > 0) { phsd->status = HTTP_GOT_GET_REQUEST; PARA_INFO_LOG("received get request\n"); } break; case HTTP_GOT_GET_REQUEST: phsd->status = HTTP_STREAMING; PARA_INFO_LOG("sending http ok to fd %d\n", sc->fd); send_msg(sc, HTTP_OK_MSG); /* ignore retval */ break; case HTTP_INVALID_GET_REQUEST: PARA_NOTICE_LOG("bad request on fd %d\n", sc->fd); ret = send_msg(sc, "HTTP/1.0 400 Bad Request\n"); if (ret >= 0) shutdown_client(sc); break; } } accept_sender_client(); } static void http_pre_monitor(struct sched *s) { struct sender_client *sc, *tmp; unsigned n; FOR_EACH_LISTEN_FD(n) { if (hss->listen_fds[n] < 0) continue; sched_monitor_readfd(hss->listen_fds[n], s); } list_for_each_entry_safe(sc, tmp, &hss->client_list, node) { struct private_http_sender_data *phsd = sc->private_data; if (phsd->status == HTTP_CONNECTED) /* need to recv get request */ sched_monitor_readfd(sc->fd, s); if (phsd->status == HTTP_GOT_GET_REQUEST || phsd->status == HTTP_INVALID_GET_REQUEST) sched_monitor_writefd(sc->fd, s); } } static int http_com_on(__a_unused struct sender_command_data *scd) { start_listening(); return 1; } /* * Initialize the client list and the access control list, and optionally * listen on the tcp port. */ static void http_send_init(void) { int i; unsigned n = lls_opt_given(OPT_RESULT(HTTP_LISTEN_ADDRESS)); if (n == 0) { hss->num_listen_fds = 1; hss->listen_addresses = alloc(sizeof(char *)); hss->listen_addresses[0] = NULL; hss->listen_fds = alloc(sizeof(int)); hss->listen_fds[0] = -1; } else { hss->num_listen_fds = n; hss->listen_addresses = alloc(n * sizeof(char *)); hss->listen_fds = alloc(n * sizeof(int)); FOR_EACH_LISTEN_FD(i) { hss->listen_addresses[i] = para_strdup(lls_string_val(i, OPT_RESULT(HTTP_LISTEN_ADDRESS))); hss->listen_fds[i] = -1; } } hss->default_port = OPT_UINT32_VAL(HTTP_PORT); init_list_head(&hss->client_list); /* Initialize an access control list */ init_list_head(&hss->acl); for (i = 0; i < lls_opt_given(OPT_RESULT(HTTP_ACCESS)); i++) { const char *arg = lls_string_val(i, OPT_RESULT(HTTP_ACCESS)); char addr[16]; int mask; if (!parse_cidr(arg, addr, sizeof(addr), &mask)) PARA_WARNING_LOG("ACL syntax error: %s, ignoring\n", arg); else acl_add_entry(&hss->acl, addr, mask); } hss->num_clients = 0; hss->max_clients = OPT_UINT32_VAL(HTTP_MAX_CLIENTS); hss->default_deny = OPT_GIVEN(HTTP_DEFAULT_DENY); if (OPT_GIVEN(HTTP_NO_AUTOSTART)) return; start_listening(); } /** * The HTTP sender. * * This sender does not FEC-encode the stream because HTTP sits on top of TCP, * a reliable transport which retransmits lost packets automatically. The * sender employs per-client queues which queue chunks of audio data if they * can not be sent immediately because the write operation would block. */ const struct sender http_sender = { .name = "http", .init = http_send_init, .shutdown = http_shutdown, .pre_monitor = http_pre_monitor, .post_monitor = http_post_monitor, .send = http_send, .shutdown_clients = http_shutdown_clients, .client_cmds = { [SENDER_on] = http_com_on, [SENDER_off] = http_com_off, [SENDER_deny] = http_com_deny, [SENDER_allow] = http_com_allow, }, .help = http_sender_help, .status = http_status, };