tiptorrent/src/core.c

/*
 * Copyright (C) 2020-2021 Soleta Networks <info@soleta.eu>
 *
 * 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 <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <syslog.h>
#include <sys/ioctl.h>
#include <netinet/in.h>
#include <ifaddrs.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <netinet/tcp.h>
#include <fcntl.h>
#include <time.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <errno.h>

struct ev_loop *tip_main_loop;

int num_clients;
static LIST_HEAD(client_list);
static LIST_HEAD(client_redirect_list);

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--;
			if (!cli->redirect)
				tip_client_activate_pending(false);
		}
	}

	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 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;
		}
		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_PROCESSING_REQUEST_2:
	case TIP_CLIENT_PROCESSING_REQUEST_3:
		if (ret == 0) {
			syslog(LOG_ERR, "client %s:%hu unexpectedly closes connection\n",
			       inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port));
			goto shutdown;
		}
		syslog(LOG_ERR, "unexpected read from client %s:%hu while in write state\n",
		       inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port));
		goto close;
	case TIP_CLIENT_CLOSE_WAIT:
		if (ret == 0) {
			syslog(LOG_INFO, "client %s:%hu finished to download successfully\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 %d, critical internal error for %s:%hu\n",
		       cli->state, 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);
}

#define TIP_CLIENT_REDIRECT_TIMEOUT	3600.

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, TIP_CLIENT_REDIRECT_TIMEOUT, 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;
		}
		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);
		}
		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 %d, critical internal error for %s:%hu\n",
		       cli->state, 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 bool tip_direct_download_in_progress(const struct tip_client *cli)
{
	struct tip_client *this;

	list_for_each_entry(this, &client_list, list) {
		if (cli->addr.sin_addr.s_addr == this->addr.sin_addr.s_addr)
			continue;
		if (cli->state != TIP_CLIENT_PENDING &&
		    !strcmp(cli->path, this->path))
			return true;
	}

	return false;
}

void tip_client_activate_pending(bool redirect_only)
{
	struct tip_client *cli, *next;
	bool redirected;

	list_for_each_entry_safe(cli, next, &client_list, list) {
		if (cli->state != TIP_CLIENT_PENDING)
			continue;
		redirected = tip_client_redirect(cli);
		if (!redirected) {
			if (redirect_only)
				continue;
			/* another client is fetching this file from server, skip. */
			if (tip_direct_download_in_progress(cli))
				continue;
		}
		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;
		/* only activate one direct download from server. */
		if (!redirected)
			redirect_only = true;
	}
}

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;
}