/* * Copyright (C) 2020-2021 Soleta Networks * * This program is free software: you can redistribute it and/or modify it under * the terms of the GNU Affero General Public License as published by the * Free Software Foundation; either version 3 of the License, or * (at your option) any later version. */ #include "core.h" #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include struct ev_loop *tip_main_loop; int num_clients; static LIST_HEAD(client_list); static LIST_HEAD(client_redirect_list); static void tip_client_activate_pending(void); static void tip_client_release(struct ev_loop *loop, struct tip_client *cli) { syslog(LOG_INFO, "closing connection with %s:%hu", inet_ntoa(cli->addr.sin_addr), htons(cli->addr.sin_port)); list_del(&cli->list); ev_timer_stop(loop, &cli->timer); ev_io_stop(loop, &cli->io); shutdown(cli->io.fd, SHUT_RDWR); close(cli->io.fd); if (cli->fd > 0) close(cli->fd); free((void *)cli->uri); free((void *)cli->path); if (cli->method == TIP_METHOD_GET) { if (tip_client_large_file(cli)) { num_clients--; tip_client_activate_pending(); } } free(cli); } static int tip_client_payload_too_large(struct tip_client *cli) { char buf[] = "HTTP/1.1 413 Payload Too Large\r\n" "Content-Length: 0\r\n\r\n"; send(tip_client_socket(cli), buf, strlen(buf), 0); return -1; } static int tip_client_state_recv_hdr(struct tip_client *cli) { char *ptr; ptr = strstr(cli->buf, "\r\n\r\n"); if (!ptr) return 0; cli->msg_len = ptr - cli->buf + 4; ptr = strstr(cli->buf, "Content-Length: "); if (ptr) { sscanf(ptr, "Content-Length: %i[^\r\n]", &cli->content_length); if (cli->content_length < 0) return -1; cli->msg_len += cli->content_length; } ptr = strstr(cli->buf, "Authorization: "); if (ptr) sscanf(ptr, "Authorization: %63[^\r\n]", cli->auth_token); return 1; } static int tip_client_recv(struct tip_client *cli, int events) { struct ev_io *io = &cli->io; int ret; ret = recv(io->fd, cli->buf + cli->buf_len, sizeof(cli->buf) - cli->buf_len, 0); if (ret < 0) { syslog(LOG_ERR, "error reading from client %s:%hu (%s)\n", inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port), strerror(errno)); } return ret; } static int tip_client_redirect_create(const struct tip_client *cli); static void tip_client_read_cb(struct ev_loop *loop, struct ev_io *io, int events) { struct tip_client *cli; int ret; cli = container_of(io, struct tip_client, io); ret = tip_client_recv(cli, events); if (ret < 0) goto close; if (cli->state != TIP_CLIENT_CLOSE_WAIT) ev_timer_again(loop, &cli->timer); cli->buf_len += ret; if (cli->buf_len >= sizeof(cli->buf)) { syslog(LOG_ERR, "client request from %s:%hu is too long\n", inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); tip_client_payload_too_large(cli); goto close; } switch (cli->state) { case TIP_CLIENT_RECEIVING_HEADER: ret = tip_client_state_recv_hdr(cli); if (ret < 0) goto close; if (!ret) return; cli->state = TIP_CLIENT_RECEIVING_PAYLOAD; /* Fall through. */ case TIP_CLIENT_RECEIVING_PAYLOAD: /* Still not enough data to process request. */ if (cli->buf_len < cli->msg_len) return; cli->state = TIP_CLIENT_PROCESSING_REQUEST; /* fall through. */ case TIP_CLIENT_PROCESSING_REQUEST: ret = tip_client_state_process_payload(cli); if (ret > 0) { syslog(LOG_INFO, "client %s:%hu for %s enters pending state\n", inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port), cli->uri); return; } else if (ret < 0) { syslog(LOG_ERR, "Failed to process HTTP request from %s:%hu\n", inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); goto close; } if (cli->method == TIP_METHOD_GET) { syslog(LOG_INFO, "client %s:%hu starts download for %s\n", inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port), cli->uri); } else { syslog(LOG_INFO, "client %s:%hu checks if %s exists\n", inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port), cli->uri); } ev_io_stop(loop, &cli->io); ev_io_set(&cli->io, tip_client_socket(cli), EV_READ | EV_WRITE); ev_io_start(loop, &cli->io); break; case TIP_CLIENT_CLOSE_WAIT: if (ret == 0) { syslog(LOG_INFO, "client %s:%hu has closed the connection with us\n", inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); goto shutdown; } syslog(LOG_ERR, "unexpected data from client %s:%hu while waiting to close\n", inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); goto close; default: syslog(LOG_ERR, "unknown read state, critical internal error for %s:%hu\n", inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); goto close; } return; shutdown: if (cli->size > FILE_SIZE_THRESHOLD) tip_client_redirect_create(cli); close: tip_client_release(loop, cli); } static void tip_client_redirect_timer_cb(struct ev_loop *loop, ev_timer *timer, int events) { struct tip_client_redirect *redir; redir = container_of(timer, struct tip_client_redirect, timer); syslog(LOG_ERR, "timeout for client redirection to %s:%hu for %s\n", inet_ntoa(redir->addr.sin_addr), ntohs(redir->addr.sin_port), redir->uri); list_del(&redir->list); free((void *)redir->uri); free(redir); } static int tip_client_redirect_create(const struct tip_client *cli) { struct tip_client_redirect *redir; bool found = false; if (!redirect || !cli->allow_redirect) return 0; list_for_each_entry(redir, &client_redirect_list, list) { if (!strcmp(redir->uri, cli->uri) && redir->addr.sin_addr.s_addr == cli->addr.sin_addr.s_addr) { found = true; break; } } if (found) { syslog(LOG_INFO, "client redirection to %s:%hu for %s already exists, skipping", inet_ntoa(cli->addr.sin_addr), htons(cli->addr.sin_port), cli->uri); return 0; } redir = calloc(1, sizeof(struct tip_client_redirect)); if (!redir) return -1; redir->addr = cli->addr; redir->addr.sin_port = htons(9999); redir->uri = strdup(cli->uri); redir->users = max_clients; list_add_tail(&redir->list, &client_redirect_list); ev_timer_init(&redir->timer, tip_client_redirect_timer_cb, 60, 0.); ev_timer_start(tip_main_loop, &redir->timer); syslog(LOG_INFO, "adding client redirection to %s:%hu for %s", inet_ntoa(redir->addr.sin_addr), htons(redir->addr.sin_port), redir->uri); return 0; } static void tip_client_write_cb(struct ev_loop *loop, struct ev_io *io, int events) { struct tip_client *cli; int ret; cli = container_of(io, struct tip_client, io); ev_timer_again(loop, &cli->timer); switch (cli->state) { case TIP_CLIENT_PROCESSING_REQUEST_2: ret = tip_client_state_process_payload_reply(cli); if (ret > 0) { goto close; } else if (ret < 0) { syslog(LOG_ERR, "Failed to process HTTP request from %s:%hu\n", inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); goto close; } break; case TIP_CLIENT_PROCESSING_REQUEST_3: ret = tip_client_state_process_payload_bulk(cli); if (ret > 0) { /* entering TIP_CLIENT_CLOSE_WAIT state. */ ev_io_stop(loop, &cli->io); ev_io_set(&cli->io, tip_client_socket(cli), EV_READ); ev_io_start(loop, &cli->io); } else if (ret < 0) { syslog(LOG_ERR, "Failed to process HTTP request from %s:%hu\n", inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); goto close; } break; default: syslog(LOG_ERR, "unknown write state, critical internal error for %s:%hu\n", inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); goto close; } return; close: tip_client_release(loop, cli); } static void tip_client_cb(struct ev_loop *loop, struct ev_io *io, int events) { if (events & EV_READ) return tip_client_read_cb(loop, io, events); if (events & EV_WRITE) return tip_client_write_cb(loop, io, events); } static void tip_client_timer_cb(struct ev_loop *loop, ev_timer *timer, int events) { struct tip_client *cli; cli = container_of(timer, struct tip_client, timer); syslog(LOG_ERR, "timeout request for client %s:%hu\n", inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); tip_client_release(loop, cli); } /* Shut down connection if there is no data after 15 seconds. */ #define TIP_CLIENT_TIMEOUT 15. static void tip_client_start(struct tip_client *cli) { cli->state = TIP_CLIENT_RECEIVING_HEADER; ev_io_start(tip_main_loop, &cli->io); ev_init(&cli->timer, tip_client_timer_cb); cli->timer.repeat = TIP_CLIENT_TIMEOUT; ev_timer_again(tip_main_loop, &cli->timer); } void tip_client_pending(struct tip_client *cli) { ev_io_stop(tip_main_loop, &cli->io); ev_timer_stop(tip_main_loop, &cli->timer); cli->state = TIP_CLIENT_PENDING; } static void tip_client_activate_pending(void) { struct tip_client *cli, *next; list_for_each_entry_safe(cli, next, &client_list, list) { if (cli->state != TIP_CLIENT_PENDING) continue; if (!tip_client_redirect(cli)) { syslog(LOG_INFO, "sending %s to client %s:%hu", cli->uri, inet_ntoa(cli->addr.sin_addr), htons(cli->addr.sin_port)); } ev_io_set(&cli->io, tip_client_socket(cli), EV_READ | EV_WRITE); ev_io_start(tip_main_loop, &cli->io); ev_timer_again(tip_main_loop, &cli->timer); cli->state = TIP_CLIENT_PROCESSING_REQUEST_2; break; } } bool tip_client_redirect(struct tip_client *cli) { struct tip_client_redirect *redir, *next; char addr[INET_ADDRSTRLEN + 1]; if (!redirect) return false; inet_ntop(AF_INET, &cli->addr.sin_addr, addr, INET_ADDRSTRLEN); list_for_each_entry_safe(redir, next, &client_redirect_list, list) { if (strcmp(redir->uri, cli->uri) || redir->addr.sin_addr.s_addr == cli->addr.sin_addr.s_addr) continue; cli->redirect = true; cli->redirect_addr = redir->addr; syslog(LOG_INFO, "redirecting client %s:%hu to %s:%hu for %s", addr, htons(cli->addr.sin_port), inet_ntoa(redir->addr.sin_addr), htons(redir->addr.sin_port), cli->uri); list_del(&redir->list); if (--redir->users == 0) { free((void *)redir->uri); ev_timer_stop(tip_main_loop, &redir->timer); free(redir); } else { /* reinsert at the end to round-robin redirections. */ list_add_tail(&redir->list, &client_redirect_list); } return true; } syslog(LOG_INFO, "no client redirections are available for %s:%hu", addr, htons(cli->addr.sin_port)); return false; } #define TIP_TCP_KEEPALIVE_IDLE 60 #define TIP_TCP_KEEPALIVE_INTL 30 #define TIP_TCP_KEEPALIVE_CNT 4 void tip_server_accept_cb(struct ev_loop *loop, struct ev_io *io, int events) { int intl = TIP_TCP_KEEPALIVE_INTL, cnt = TIP_TCP_KEEPALIVE_CNT; int on = 1, idle = TIP_TCP_KEEPALIVE_IDLE; struct sockaddr_in client_addr; socklen_t addrlen = sizeof(client_addr); struct tip_client *cli; int client_sd, flags; if (events & EV_ERROR) return; client_sd = accept(io->fd, (struct sockaddr *)&client_addr, &addrlen); if (client_sd < 0) { syslog(LOG_ERR, "cannot accept client connection\n"); return; } flags = fcntl(client_sd, F_GETFL); fcntl(client_sd, F_SETFL, flags | O_NONBLOCK); setsockopt(client_sd, SOL_SOCKET, SO_KEEPALIVE, &on, sizeof(int)); setsockopt(client_sd, IPPROTO_TCP, TCP_KEEPIDLE, &idle, sizeof(int)); setsockopt(client_sd, IPPROTO_TCP, TCP_KEEPINTVL, &intl, sizeof(int)); setsockopt(client_sd, IPPROTO_TCP, TCP_KEEPCNT, &cnt, sizeof(int)); cli = (struct tip_client *)calloc(1, sizeof(struct tip_client)); if (!cli) { close(client_sd); return; } memcpy(&cli->addr, &client_addr, sizeof(client_addr)); cli->fd = -1; syslog(LOG_ERR, "accepting client connection from %s:%hu", inet_ntoa(cli->addr.sin_addr), htons(cli->addr.sin_port)); list_add_tail(&cli->list, &client_list); ev_io_init(&cli->io, tip_client_cb, client_sd, EV_READ); tip_client_start(cli); } int tip_socket_server_init(const char *port) { struct sockaddr_in local; int sd, on = 1; sd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); if (sd < 0) { syslog(LOG_ERR, "cannot create main socket\n"); return -1; } setsockopt(sd, SOL_SOCKET, SO_REUSEPORT, &on, sizeof(int)); local.sin_addr.s_addr = htonl(INADDR_ANY); local.sin_family = AF_INET; local.sin_port = htons(atoi(port)); if (bind(sd, (struct sockaddr *) &local, sizeof(local)) < 0) { close(sd); syslog(LOG_ERR, "cannot bind socket\n"); return -1; } listen(sd, 250); return sd; }