Import new Cytoplasm library based off of code from Telodendria.

Telodendria doesn't use this library yet, but it will soon.

src/HttpServer.c

@@ -0,0 +1,753 @@
+/*
+ * Copyright (C) 2022-2023 Jordan Bancino <@jordan:bancino.net>
+ *
+ * Permission is hereby granted, free of charge, to any person
+ * obtaining a copy of this software and associated documentation files
+ * (the "Software"), to deal in the Software without restriction,
+ * including without limitation the rights to use, copy, modify, merge,
+ * publish, distribute, sublicense, and/or sell copies of the Software,
+ * and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT.  IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
+ * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
+ * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+#include <HttpServer.h>
+#include <Memory.h>
+#include <Queue.h>
+#include <Array.h>
+#include <Util.h>
+#include <Tls.h>
+#include <Log.h>
+#include <Str.h>
+
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <errno.h>
+#include <poll.h>
+#include <string.h>
+#include <ctype.h>
+
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <fcntl.h>
+
+#include <errno.h>
+
+static const char ENABLE = 1;
+
+struct HttpServer
+{
+    HttpServerConfig config;
+    int sd;
+    pthread_t socketThread;
+
+    volatile unsigned int stop:1;
+    volatile unsigned int isRunning:1;
+
+    Queue *connQueue;
+    pthread_mutex_t connQueueMutex;
+
+    Array *threadPool;
+};
+
+struct HttpServerContext
+{
+    HashMap *requestHeaders;
+    HttpRequestMethod requestMethod;
+    char *requestPath;
+    HashMap *requestParams;
+
+    HashMap *responseHeaders;
+    HttpStatus responseStatus;
+
+    Stream *stream;
+};
+
+typedef struct HttpServerWorkerThreadArgs
+{
+    HttpServer *server;
+    int id;
+    pthread_t thread;
+} HttpServerWorkerThreadArgs;
+
+static HttpServerContext *
+HttpServerContextCreate(HttpRequestMethod requestMethod,
+          char *requestPath, HashMap * requestParams, Stream * stream)
+{
+    HttpServerContext *c;
+
+    c = Malloc(sizeof(HttpServerContext));
+    if (!c)
+    {
+        return NULL;
+    }
+
+    c->responseHeaders = HashMapCreate();
+    if (!c->responseHeaders)
+    {
+        Free(c->requestHeaders);
+        Free(c);
+        return NULL;
+    }
+
+    c->requestMethod = requestMethod;
+    c->requestPath = requestPath;
+    c->requestParams = requestParams;
+    c->stream = stream;
+    c->responseStatus = HTTP_OK;
+
+    return c;
+}
+
+static void
+HttpServerContextFree(HttpServerContext * c)
+{
+    char *key;
+    void *val;
+
+    if (!c)
+    {
+        return;
+    }
+
+    while (HashMapIterate(c->requestHeaders, &key, &val))
+    {
+        Free(val);
+    }
+    HashMapFree(c->requestHeaders);
+
+    while (HashMapIterate(c->responseHeaders, &key, &val))
+    {
+        /*
+         * These are generated by code. As such, they may be either
+         * on the heap, or on the stack, depending on how they were
+         * added.
+         *
+         * Basically, if the memory API knows about a pointer, then
+         * it can be freed. If it doesn't know about a pointer, skip
+         * freeing it because it's probably a stack pointer.
+         */
+
+        if (MemoryInfoGet(val))
+        {
+            Free(val);
+        }
+    }
+
+    HashMapFree(c->responseHeaders);
+
+    while (HashMapIterate(c->requestParams, &key, &val))
+    {
+        Free(val);
+    }
+
+    HashMapFree(c->requestParams);
+
+    Free(c->requestPath);
+    StreamClose(c->stream);
+
+    Free(c);
+}
+
+HashMap *
+HttpRequestHeaders(HttpServerContext * c)
+{
+    if (!c)
+    {
+        return NULL;
+    }
+
+    return c->requestHeaders;
+}
+
+HttpRequestMethod
+HttpRequestMethodGet(HttpServerContext * c)
+{
+    if (!c)
+    {
+        return HTTP_METHOD_UNKNOWN;
+    }
+
+    return c->requestMethod;
+}
+
+char *
+HttpRequestPath(HttpServerContext * c)
+{
+    if (!c)
+    {
+        return NULL;
+    }
+
+    return c->requestPath;
+}
+
+HashMap *
+HttpRequestParams(HttpServerContext * c)
+{
+    if (!c)
+    {
+        return NULL;
+    }
+
+    return c->requestParams;
+}
+
+char *
+HttpResponseHeader(HttpServerContext * c, char *key, char *val)
+{
+    if (!c)
+    {
+        return NULL;
+    }
+
+    return HashMapSet(c->responseHeaders, key, val);
+}
+
+void
+HttpResponseStatus(HttpServerContext * c, HttpStatus status)
+{
+    if (!c)
+    {
+        return;
+    }
+
+    c->responseStatus = status;
+}
+
+HttpStatus
+HttpResponseStatusGet(HttpServerContext * c)
+{
+    if (!c)
+    {
+        return HTTP_STATUS_UNKNOWN;
+    }
+
+    return c->responseStatus;
+}
+
+Stream *
+HttpServerStream(HttpServerContext * c)
+{
+    if (!c)
+    {
+        return NULL;
+    }
+
+    return c->stream;
+}
+
+void
+HttpSendHeaders(HttpServerContext * c)
+{
+    Stream *fp = c->stream;
+
+    char *key;
+    char *val;
+
+    StreamPrintf(fp, "HTTP/1.0 %d %s\n", c->responseStatus, HttpStatusToString(c->responseStatus));
+
+    while (HashMapIterate(c->responseHeaders, &key, (void **) &val))
+    {
+        StreamPrintf(fp, "%s: %s\n", key, val);
+    }
+
+    StreamPuts(fp, "\n");
+}
+
+static Stream *
+DequeueConnection(HttpServer * server)
+{
+    Stream *fp;
+
+    if (!server)
+    {
+        return NULL;
+    }
+
+    pthread_mutex_lock(&server->connQueueMutex);
+    fp = QueuePop(server->connQueue);
+    pthread_mutex_unlock(&server->connQueueMutex);
+
+    return fp;
+}
+
+HttpServer *
+HttpServerCreate(HttpServerConfig * config)
+{
+    HttpServer *server;
+    struct sockaddr_in sa;
+
+    if (!config)
+    {
+        return NULL;
+    }
+
+    if (!config->handler)
+    {
+        return NULL;
+    }
+
+#ifndef TLS_IMPL
+    if (config->flags & HTTP_FLAG_TLS)
+    {
+        return NULL;
+    }
+#endif
+
+    server = Malloc(sizeof(HttpServer));
+    if (!server)
+    {
+        goto error;
+    }
+
+    memset(server, 0, sizeof(HttpServer));
+
+    server->config = *config;
+    server->config.tlsCert = StrDuplicate(config->tlsCert);
+    server->config.tlsKey = StrDuplicate(config->tlsKey);
+
+    server->threadPool = ArrayCreate();
+    if (!server->threadPool)
+    {
+        goto error;
+    }
+
+    server->connQueue = QueueCreate(config->maxConnections);
+    if (!server->connQueue)
+    {
+        goto error;
+    }
+
+    if (pthread_mutex_init(&server->connQueueMutex, NULL) != 0)
+    {
+        goto error;
+    }
+
+    server->sd = socket(AF_INET, SOCK_STREAM, 0);
+
+    if (server->sd < 0)
+    {
+        goto error;
+    }
+
+    if (fcntl(server->sd, F_SETFL, O_NONBLOCK) == -1)
+    {
+        goto error;
+    }
+
+    if (setsockopt(server->sd, SOL_SOCKET, SO_REUSEADDR, &ENABLE, sizeof(int)) < 0)
+    {
+        goto error;
+    }
+
+#ifdef SO_REUSEPORT
+    if (setsockopt(server->sd, SOL_SOCKET, SO_REUSEPORT, &ENABLE, sizeof(int)) < 0)
+    {
+        goto error;
+    }
+#endif
+
+    memset(&sa, 0, sizeof(struct sockaddr_in));
+
+    sa.sin_family = AF_INET;
+    sa.sin_port = htons(config->port);
+    sa.sin_addr.s_addr = htonl(INADDR_ANY);
+
+    if (bind(server->sd, (struct sockaddr *) & sa, sizeof(sa)) < 0)
+    {
+        goto error;
+    }
+
+    if (listen(server->sd, config->maxConnections) < 0)
+    {
+        goto error;
+    }
+
+    server->stop = 0;
+    server->isRunning = 0;
+
+    return server;
+
+error:
+    if (server)
+    {
+        if (server->connQueue)
+        {
+            QueueFree(server->connQueue);
+        }
+
+        pthread_mutex_destroy(&server->connQueueMutex);
+
+        if (server->threadPool)
+        {
+            ArrayFree(server->threadPool);
+        }
+
+        if (server->sd)
+        {
+            close(server->sd);
+        }
+
+        Free(server);
+    }
+    return NULL;
+}
+
+HttpServerConfig *
+HttpServerConfigGet(HttpServer * server)
+{
+    if (!server)
+    {
+        return NULL;
+    }
+
+    return &server->config;
+}
+
+void
+HttpServerFree(HttpServer * server)
+{
+    if (!server)
+    {
+        return;
+    }
+
+    close(server->sd);
+    QueueFree(server->connQueue);
+    pthread_mutex_destroy(&server->connQueueMutex);
+    ArrayFree(server->threadPool);
+    Free(server->config.tlsCert);
+    Free(server->config.tlsKey);
+    Free(server);
+}
+
+static void *
+HttpServerWorkerThread(void *args)
+{
+    HttpServerWorkerThreadArgs *wArgs = (HttpServerWorkerThreadArgs *) args;
+    HttpServer *server = wArgs->server;
+
+    while (!server->stop)
+    {
+        Stream *fp;
+        HttpServerContext *context;
+
+        char *line = NULL;
+        size_t lineSize = 0;
+        ssize_t lineLen = 0;
+
+        char *requestMethodPtr;
+        char *pathPtr;
+        char *requestPath;
+        char *requestProtocol;
+
+        HashMap *requestParams;
+        ssize_t requestPathLen;
+
+        ssize_t i = 0;
+        HttpRequestMethod requestMethod;
+
+        long firstRead;
+
+        fp = DequeueConnection(server);
+
+        if (!fp)
+        {
+            /* Block for 1 millisecond before continuing so we don't
+             * murder the CPU if the queue is empty. */
+            UtilSleepMillis(1);
+            continue;
+        }
+
+        /* Get the first line of the request.
+         * 
+         * Every once in a while, we're too fast for the client. When this
+         * happens, UtilGetLine() sets errno to EAGAIN. If we get
+         * EAGAIN, then clear the error on the stream and try again
+         * after a few ms. This is typically more than enough time for
+         * the client to send data. */
+        firstRead = UtilServerTs();
+        while ((lineLen = UtilGetLine(&line, &lineSize, fp)) == -1
+               && errno == EAGAIN)
+        {
+            StreamClearError(fp);
+
+            /* If the server is stopped, or it's been a while, just
+             * give up so we aren't wasting a thread on this client. */
+            if (server->stop || (UtilServerTs() - firstRead) > 1000 * 30)
+            {
+                goto finish;
+            }
+
+            UtilSleepMillis(5);
+        }
+
+        if (lineLen == -1)
+        {
+            goto bad_request;
+        }
+
+        requestMethodPtr = line;
+        for (i = 0; i < lineLen; i++)
+        {
+            if (line[i] == ' ')
+            {
+                line[i] = '\0';
+                break;
+            }
+        }
+
+        if (i == lineLen)
+        {
+            goto bad_request;
+        }
+
+        requestMethod = HttpRequestMethodFromString(requestMethodPtr);
+        if (requestMethod == HTTP_METHOD_UNKNOWN)
+        {
+            goto bad_request;
+        }
+
+        pathPtr = line + i + 1;
+
+        for (i = 0; i < (line + lineLen) - pathPtr; i++)
+        {
+            if (pathPtr[i] == ' ')
+            {
+                pathPtr[i] = '\0';
+                break;
+            }
+        }
+
+        requestPathLen = i;
+        requestPath = Malloc(((requestPathLen + 1) * sizeof(char)));
+        strncpy(requestPath, pathPtr, requestPathLen + 1);
+
+        requestProtocol = &pathPtr[i + 1];
+        line[lineLen - 2] = '\0';  /* Get rid of \r and \n */
+
+        if (!StrEquals(requestProtocol, "HTTP/1.1") && !StrEquals(requestProtocol, "HTTP/1.0"))
+        {
+            Free(requestPath);
+            goto bad_request;
+        }
+
+        /* Find request params */
+        for (i = 0; i < requestPathLen; i++)
+        {
+            if (requestPath[i] == '?')
+            {
+                break;
+            }
+        }
+
+        requestPath[i] = '\0';
+        requestParams = (i == requestPathLen) ? NULL : HttpParamDecode(requestPath + i + 1);
+
+        context = HttpServerContextCreate(requestMethod, requestPath, requestParams, fp);
+        if (!context)
+        {
+            Free(requestPath);
+            goto internal_error;
+        }
+
+        context->requestHeaders = HttpParseHeaders(fp);
+        if (!context->requestHeaders)
+        {
+            goto internal_error;
+        }
+
+        server->config.handler(context, server->config.handlerArgs);
+
+        HttpServerContextFree(context);
+        fp = NULL;                 /* The above call will close this
+                                    * Stream */
+        goto finish;
+
+internal_error:
+        StreamPuts(fp, "HTTP/1.0 500 Internal Server Error\n");
+        StreamPuts(fp, "Connection: close\n");
+        goto finish;
+
+bad_request:
+        StreamPuts(fp, "HTTP/1.0 400 Bad Request\n");
+        StreamPuts(fp, "Connection: close\n");
+        goto finish;
+
+finish:
+        Free(line);
+        if (fp)
+        {
+            StreamClose(fp);
+        }
+    }
+
+    return NULL;
+}
+
+static void *
+HttpServerEventThread(void *args)
+{
+    HttpServer *server = (HttpServer *) args;
+    struct pollfd pollFds[1];
+    Stream *fp;
+    size_t i;
+
+    server->isRunning = 1;
+    server->stop = 0;
+
+    pollFds[0].fd = server->sd;
+    pollFds[0].events = POLLIN;
+
+    for (i = 0; i < server->config.threads; i++)
+    {
+        HttpServerWorkerThreadArgs *workerThread = Malloc(sizeof(HttpServerWorkerThreadArgs));
+
+        if (!workerThread)
+        {
+            /* TODO: Make the event thread return an error to the main
+             * thread */
+            return NULL;
+        }
+
+        workerThread->server = server;
+        workerThread->id = i;
+
+        if (pthread_create(&workerThread->thread, NULL, HttpServerWorkerThread, workerThread) != 0)
+        {
+            /* TODO: Make the event thread return an error to the main
+             * thread */
+            return NULL;
+        }
+
+        ArrayAdd(server->threadPool, workerThread);
+    }
+
+    while (!server->stop)
+    {
+        struct sockaddr_storage addr;
+        socklen_t addrLen = sizeof(addr);
+        int connFd;
+        int pollResult;
+
+
+        pollResult = poll(pollFds, 1, 500);
+
+        if (pollResult < 0)
+        {
+            /* The poll either timed out, or was interrupted. */
+            continue;
+        }
+
+        pthread_mutex_lock(&server->connQueueMutex);
+
+        /* Don't even accept connections if the queue is full. */
+        if (!QueueFull(server->connQueue))
+        {
+            connFd = accept(server->sd, (struct sockaddr *) & addr, &addrLen);
+
+            if (connFd < 0)
+            {
+                pthread_mutex_unlock(&server->connQueueMutex);
+                continue;
+            }
+
+#ifdef TLS_IMPL
+            if (server->config.flags & HTTP_FLAG_TLS)
+            {
+                fp = TlsServerStream(connFd, server->config.tlsCert, server->config.tlsKey);
+            }
+            else
+            {
+                fp = StreamFd(connFd);
+            }
+#else
+            fp = StreamFd(connFd);
+#endif
+
+            if (!fp)
+            {
+                pthread_mutex_unlock(&server->connQueueMutex);
+                close(connFd);
+                continue;
+            }
+
+            QueuePush(server->connQueue, fp);
+        }
+        pthread_mutex_unlock(&server->connQueueMutex);
+    }
+
+    for (i = 0; i < server->config.threads; i++)
+    {
+        HttpServerWorkerThreadArgs *workerThread = ArrayGet(server->threadPool, i);
+
+        pthread_join(workerThread->thread, NULL);
+        Free(workerThread);
+    }
+
+    while ((fp = DequeueConnection(server)))
+    {
+        StreamClose(fp);
+    }
+
+    server->isRunning = 0;
+
+    return NULL;
+}
+
+int
+HttpServerStart(HttpServer * server)
+{
+    if (!server)
+    {
+        return 0;
+    }
+
+    if (server->isRunning)
+    {
+        return 1;
+    }
+
+    if (pthread_create(&server->socketThread, NULL, HttpServerEventThread, server) != 0)
+    {
+        return 0;
+    }
+
+    return 1;
+}
+
+void
+HttpServerJoin(HttpServer * server)
+{
+    if (!server)
+    {
+        return;
+    }
+
+    pthread_join(server->socketThread, NULL);
+}
+
+void
+HttpServerStop(HttpServer * server)
+{
+    if (!server)
+    {
+        return;
+    }
+
+    server->stop = 1;
+}