dany
/
ant-http
mirror of https://github.com/lxsang/ant-http
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
ant-http/server.c

578 lines
17 KiB
C
Raw Permalink Blame History

#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include <sys/socket.h>
#include <poll.h>
#include <sys/stat.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>
//#include <limits.h>
//#include <stdlib.h>
#ifdef USE_OPENSSL
#include <openssl/ssl.h>
#include <openssl/err.h>
#endif
#include "server.h"
#include "lib/handle.h"
#include "plugin_manager.h"
#include "lib/scheduler.h"
#include "lib/utils.h"
#include "decode.h"
#include "config.h"
static pthread_mutex_t server_mux = PTHREAD_MUTEX_INITIALIZER;
config_t g_server_config;
void *execute_plugin(void *data, const char *pname);
void *serve_file(void *data);
void *accept_request(void *data)
{
char buf[BUFFLEN];
char *token = NULL;
char *line = NULL;
antd_task_t *task;
antd_request_t *rq = (antd_request_t *)data;
task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io);
antd_task_bind_event(task, rq->client->sock, 0, TASK_EVT_ON_WRITABLE | TASK_EVT_ON_READABLE);
// first verify if the socket is ready
antd_client_t *client = (antd_client_t *)rq->client;
struct pollfd pfd[1];
pfd[0].fd = client->sock;
pfd[0].events = POLLIN | POLLOUT;
int sel = poll(pfd, 1, POLL_EVENT_TO);
if (sel == -1)
{
antd_error(rq->client, 400, "Bad request");
return task;
}
if (pfd[0].revents & POLLERR || pfd[0].revents & POLLHUP)
{
antd_error(rq->client, 400, "Bad request");
return task;
}
if (sel == 0 || (!(pfd[0].revents & POLLIN) && !(pfd[0].revents & POLLOUT)))
{
task->handle = accept_request;
return task;
}
// perform the ssl handshake if enabled
#ifdef USE_OPENSSL
int ret = -1, stat;
if (client->ssl && client->state == ANTD_CLIENT_ACCEPT)
{
// LOG("Atttempt %d\n", client->attempt);
if (SSL_accept((SSL *)client->ssl) == -1)
{
stat = SSL_get_error((SSL *)client->ssl, ret);
switch (stat)
{
case SSL_ERROR_WANT_READ:
case SSL_ERROR_WANT_WRITE:
case SSL_ERROR_NONE:
task->handle = accept_request;
return task;
default:
ERROR("Error performing SSL handshake %d %d %s", stat, ret, ERR_error_string(ERR_get_error(), NULL));
antd_error(rq->client, 400, "Invalid SSL request");
// server_config.connection++;
ERR_print_errors_fp(stderr);
return task;
}
}
client->state = ANTD_CLIENT_HANDSHAKE;
task->handle = accept_request;
// LOG("Handshake finish for %d\n", client->sock);
return task;
}
else
{
if (!((pfd[0].revents & POLLIN)))
{
task->handle = accept_request;
return task;
}
}
#endif
// LOG("Ready for reading %d\n", client->sock);
// server_config.connection++;
client->state = ANTD_CLIENT_PROTO_CHECK;
read_buf(rq->client, buf, sizeof(buf));
line = buf;
LOG("Request (%d): %s", rq->client->sock, line);
// get the method string
token = strsep(&line, " ");
if (!line)
{
// LOG("No method found");
antd_error(rq->client, 405, "No method found");
return task;
}
trim(token, ' ');
trim(line, ' ');
dput(rq->request, "METHOD", strdup(token));
// get the request
token = strsep(&line, " ");
if (!line)
{
// LOG("No request found");
antd_error(rq->client, 400, "Bad request");
return task;
}
trim(token, ' ');
trim(line, ' ');
trim(line, '\n');
trim(line, '\r');
dput(rq->request, "PROTOCOL", strdup(line));
dput(rq->request, "REQUEST_QUERY", strdup(token));
line = token;
token = strsep(&line, "?");
dput(rq->request, "REQUEST_PATH", url_decode(token));
// decode request
// now return the task
task->handle = decode_request_header;
return task;
}
void *resolve_request(void *data)
{
struct stat st;
char path[2 * BUFFLEN];
antd_request_t *rq = (antd_request_t *)data;
antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io);
char *url = (char *)dvalue(rq->request, "REQUEST_URI");
char *newurl = NULL;
char *rqp = NULL;
char *oldrqp = NULL;
rq->client->state = ANTD_CLIENT_RESOLVE_REQUEST;
char * root = (char *)dvalue(rq->request, "SERVER_WWW_ROOT");
snprintf(path, sizeof(path), "%s/%s", root, url);
LOG("URL is : %s", url);
LOG("Resource Path is : %s", path);
// if (path[strlen(path) - 1] == '/')
// strcat(path, "index.html");
if (stat(path, &st) == -1)
{
free(task);
rqp = strdup(url);
oldrqp = rqp;
trim(rqp, '/');
newurl = strsep(&rqp, "/");
if (!rqp)
rqp = strdup("/");
else
rqp = strdup(rqp);
dput(rq->request, "REQUEST_URI", rqp);
dput(rq->request, "RESOURCE_PATH", __s("%s/%s", root,rqp));
LOG("Execute plugin %s", newurl);
task = execute_plugin(rq, newurl);
free(oldrqp);
return task;
}
else
{
if (S_ISDIR(st.st_mode))
{
strcat(path, "/index.html");
if (stat(path, &st) == -1)
{
chain_t it;
newurl = NULL;
for_each_assoc(it, g_server_config.handlers)
{
memset(path, 0, sizeof(path));
snprintf(path, sizeof(path), "%s/%s/index.%s", root, url, it->key);
if (stat(path, &st) == 0)
{
i = g_server_config.handlers->cap;
newurl = path;
break;
}
}
if (!newurl)
{
antd_error(rq->client, 404, "Resource Not Found");
return task;
}
}
}
dput(rq->request, "RESOURCE_PATH", strdup(path));
// check if the mime is supported
// if the mime is not supported
// find an handler plugin to process it
// if the plugin is not found, forbidden access to the file should be sent
char *mime_type = mime(path);
dput(rq->request, "RESOURCE_MIME", strdup(mime_type));
if (strcmp(mime_type, "application/octet-stream") == 0)
{
char *ex = ext(path);
char *h = NULL;
if (ex)
{
h = dvalue(g_server_config.handlers, ex);
free(ex);
}
if (h)
{
// sprintf(path,"/%s%s",h,url);
// LOG("WARNING::::Access octetstream via handle %s", h);
// if(execute_plugin(client,buf,method,rq) < 0)
// cannot_execute(client);
free(task);
return execute_plugin(rq, h);
}
else
antd_error(rq->client, 403, "Access forbidden");
}
else
{
// discard all request data
dictionary_t headers = (dictionary_t)dvalue(rq->request, "REQUEST_HEADER");
if (headers)
{
char *sclen = (char *)dvalue(headers, "Content-Length");
unsigned clen = 0;
unsigned read = 0;
int count;
if (sclen)
{
clen = atoi(sclen);
while (read < clen)
{
count = antd_recv(rq->client, path, sizeof(path) < clen ? sizeof(path) : clen);
if (count <= 0)
break;
read += count;
}
}
}
antd_task_bind_event(task, rq->client->sock, 0, TASK_EVT_ON_WRITABLE);
task->handle = serve_file;
}
return task;
}
}
void *finish_request(void *data)
{
if (!data)
return NULL;
destroy_request(data);
g_server_config.connection--;
LOG("Remaining connection %d", g_server_config.connection);
return NULL;
}
void *serve_file(void *data)
{
antd_request_t *rq = (antd_request_t *)data;
antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io);
char *path = (char *)dvalue(rq->request, "RESOURCE_PATH");
char *mime_type = (char *)dvalue(rq->request, "RESOURCE_MIME");
rq->client->state = ANTD_CLIENT_SERVE_FILE;
struct stat st;
int s = stat(path, &st);
if (s == -1)
{
antd_error(rq->client, 404, "File not found");
}
else
{
// check if it is modified
dictionary_t header = (dictionary_t)dvalue(rq->request, "REQUEST_HEADER");
char *last_modif_since = (char *)dvalue(header, "If-Modified-Since");
time_t t = st.st_ctime;
struct tm tm;
if (last_modif_since)
{
strptime(last_modif_since, "%a, %d %b %Y %H:%M:%S GMT", &tm);
t = timegm(&tm);
// t = mktime(localtime(&t));
}
if (last_modif_since && st.st_ctime == t)
{
// return the not changed
antd_error(rq->client, 304, "");
}
else
{
int size = (int)st.st_size;
char ibuf[64];
snprintf(ibuf, sizeof(ibuf), "%d", size);
antd_response_header_t rhd;
rhd.cookie = NULL;
rhd.status = 200;
rhd.header = dict();
dput(rhd.header, "Content-Type", strdup(mime_type));
#ifdef USE_ZLIB
if (!compressable(mime_type) || rq->client->z_level == ANTD_CNONE)
#endif
dput(rhd.header, "Content-Length", strdup(ibuf));
gmtime_r(&st.st_ctime, &tm);
strftime(ibuf, 64, "%a, %d %b %Y %H:%M:%S GMT", &tm);
dput(rhd.header, "Last-Modified", strdup(ibuf));
dput(rhd.header, "Cache-Control", strdup("no-cache"));
antd_send_header(rq->client, &rhd);
__f(rq->client, path);
}
}
return task;
}
static void *proxy_monitor(void *data)
{
antd_request_t *rq = (antd_request_t *)data;
rq->client->state = ANTD_CLIENT_PROXY_MONITOR;
antd_client_t *proxy = (antd_client_t *)dvalue(rq->request, "PROXY_HANDLE");
antd_task_t *task = antd_create_task(NULL, data, NULL, rq->client->last_io);
int pret, ret, sz1 = 0, sz2 = 0;
char *buf = NULL;
buf = (char *)malloc(BUFFLEN);
struct pollfd pfd[1];
memset(pfd, 0, sizeof(pfd));
pfd[0].fd = proxy->sock;
pfd[0].events = POLLIN;
ret = 1;
do
{
sz1 = antd_recv_upto(rq->client, buf, BUFFLEN);
if ((sz1 < 0) || (sz1 > 0 && antd_send(proxy, buf, sz1) != sz1))
{
ret = 0;
break;
}
pret = poll(pfd, 1, 0);
if (pret < 0)
{
(void)close(proxy->sock);
return task;
}
sz2 = 0;
if (pret > 0 && (pfd[0].revents & POLLIN))
{
sz2 = antd_recv_upto(proxy, buf, BUFFLEN);
if (sz2 <= 0 || (sz2 > 0 && antd_send(rq->client, buf, sz2) != sz2))
{
ret = 0;
break;
}
}
if ((pret > 0) && (pfd[0].revents & POLLERR ||
pfd[0].revents & POLLRDHUP ||
pfd[0].revents & POLLHUP ||
pfd[0].revents & POLLNVAL))
{
ret = 0;
break;
}
} while (sz1 > 0 || sz2 > 0);
free(buf);
if (ret == 0)
{
(void)close(proxy->sock);
return task;
}
if (pfd[0].revents & POLLIN)
{
antd_task_bind_event(task, proxy->sock, 0, TASK_EVT_ON_READABLE);
}
else
{
antd_task_bind_event(task, proxy->sock, 50u, TASK_EVT_ON_TIMEOUT);
}
task->handle = proxy_monitor;
task->access_time = rq->client->last_io;
antd_task_bind_event(task, rq->client->sock, 0, TASK_EVT_ON_READABLE);
return task;
}
void *proxify(void *data)
{
int sock_fd, size, ret;
char *str = NULL;
chain_t it;
antd_request_t *rq = (antd_request_t *)data;
antd_client_t *proxy = NULL;
rq->client->state = ANTD_CLIENT_RESOLVE_REQUEST;
char *host = dvalue(rq->request, "PROXY_HOST");
int port = atoi(dvalue(rq->request, "PROXY_PORT"));
char *path = dvalue(rq->request, "PROXY_PATH");
char *query = dvalue(rq->request, "PROXY_QUERY");
char *ptr, *ip;
dictionary_t xheader = dvalue(rq->request, "REQUEST_HEADER");
antd_task_t *task = antd_create_task(NULL, data, NULL, rq->client->last_io);
if (!xheader)
{
antd_error(rq->client, 400, "Badd Request");
return task;
}
pthread_mutex_lock(&server_mux);
ip = NULL;
// ip_from_host is not threadsafe, need to lock it
ptr = ip_from_hostname(host);
if (ptr)
{
ip = strdup(ptr);
}
pthread_mutex_unlock(&server_mux);
if (!ip)
{
antd_error(rq->client, 502, "Badd address");
return task;
}
// TODO support ipv6
sock_fd = antd_request_socket(ip, port);
free(ip);
if (sock_fd == -1)
{
antd_error(rq->client, 503, "Service Unavailable");
return task;
}
set_nonblock(sock_fd);
/*struct timeval timeout;
timeout.tv_sec = 2;
timeout.tv_usec = 0; //POLL_EVENT_TO*1000;
if (setsockopt(sock_fd, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeout, sizeof(timeout)) < 0)
{
ERROR("setsockopt failed:%s", strerror(errno));
antd_error(rq->client, 500, "Internal proxy error");
(void)close(sock_fd);
return task;
}
if (setsockopt(sock_fd, SOL_SOCKET, SO_SNDTIMEO, (char *)&timeout, sizeof(timeout)) < 0)
{
ERROR("setsockopt failed:%s", strerror(errno));
antd_error(rq->client, 500, "Internal proxy error");
(void)close(sock_fd);
return task;
}*/
proxy = (antd_client_t *)malloc(sizeof(antd_client_t));
proxy->sock = sock_fd;
proxy->ssl = NULL;
proxy->zstream = NULL;
proxy->z_level = ANTD_CNONE;
time(&proxy->last_io);
// store content length here
dput(rq->request, "PROXY_HANDLE", proxy);
str = __s("%s %s?%s HTTP/1.1\r\n", (char *)dvalue(rq->request, "METHOD"), path, query);
size = strlen(str);
ret = antd_send(proxy, str, size);
free(str);
if (ret != size)
{
antd_error(rq->client, 500, "");
(void)close(sock_fd);
return task;
}
for_each_assoc(it, xheader)
{
str = __s("%s: %s\r\n", it->key, (char *)it->value);
size = strlen(str);
ret = antd_send(proxy, str, size);
free(str);
if (ret != size)
{
antd_error(rq->client, 500, "");
(void)close(sock_fd);
return task;
}
}
(void)antd_send(proxy, "\r\n", 2);
// now monitor the proxy
task->handle = proxy_monitor;
task->access_time = rq->client->last_io;
// register event
antd_task_bind_event(task, proxy->sock, 0, TASK_EVT_ON_READABLE | TASK_EVT_ON_WRITABLE);
return task;
}
/**
* Execute a plugin based on the http requeset
* First decode the http request header to find the correct plugin
* and the correct function on the plugin
* Second, decode all parameters necessary of the request and pass it
* to the callback function.
* Execute the callback function if sucess
* @param client soket client
* @param path request path
* @param method request method
* @param query_string GET query string
* @return -1 if failure
* 1 if sucess
*/
void *execute_plugin(void *data, const char *pname)
{
char pattern[256];
antd_plugin_ctx_t* ctx;
antd_request_t *rq = (antd_request_t *)data;
antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io);
antd_task_bind_event(task, rq->client->sock, 0, TASK_EVT_ON_WRITABLE | TASK_EVT_ON_READABLE);
snprintf(pattern, sizeof(pattern), "\\b%s\\b", pname);
char *port_s = (char *)dvalue(rq->request, "SERVER_PORT");
port_config_t *pcnf = (port_config_t *)dvalue(g_server_config.ports, port_s);
// check if plugin is enabled on this port
if (!pcnf->plugins || !regex_match(pattern, pcnf->plugins, 0, NULL))
{
LOG("No plugin matched in [%s] using pattern [%s]", pcnf->plugins, pattern);
antd_error(rq->client, 403, "Access forbidden");
return task;
}
// LOG("Plugin name '%s'", pname);
rq->client->state = ANTD_CLIENT_PLUGIN_EXEC;
// load the plugin
pthread_mutex_lock(&server_mux);
ctx = antd_plugin_load(pname);
pthread_mutex_unlock(&server_mux);
if (ctx == NULL)
{
antd_error(rq->client, 503, "Requested service not found");
return task;
}
rq->context = ctx;
// check if we need the raw data or not
if (antd_plugin_is_raw_body(ctx) == 1)
{
task->handle = antd_get_ctx_handle(ctx);
}
else
{
free(task);
task = antd_create_task(decode_post_request, (void *)rq, antd_get_ctx_handle(ctx), rq->client->last_io);
}
return task;
}
Reference in New Issue View Git Blame Copy Permalink