src/server/safs/proxy.c

/*
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 *
 * Copyright 2026 Olaf Wintermann. All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *   1. Redistributions of source code must retain the above copyright
 *      notice, this list of conditions and the following disclaimer.
 *
 *   2. Redistributions in binary form must reproduce the above copyright
 *      notice, this list of conditions and the following disclaimer in the
 *      documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include "proxy.h"

#include <sys/socket.h>
#include <arpa/inet.h>
#include <sys/types.h>
#include <netdb.h>
#include <ctype.h>
#include <string.h>

#include "../util/pblock.h"
#include "../util/util.h"
#include "../util/io.h"
#include "../proxy/httpclient.h"


typedef struct ProxyRequest {
    Session *sn;
    Request *rq;
    
    HttpClient *client;
    
    /*
     * request header rewrite map
     * name: header name
     * value: header value or an empty string, if the header should be removed
     */
    pblock *request_header_rewrite;
    
    /*
     * response header rewrite map
     * name: header name
     * value: header value or an empty string, if the header should be removed
     */
    pblock  *response_header_rewrite;
    
    /*
     * event structure for sn->csd
     */
    Event event;
    
    /*
     * websocket read buffer
     */
    char *read_buf;
    size_t read_buf_alloc;
    size_t read_buf_size;
    size_t read_buf_pos;
    
    /*
     * Has the response started (proxy_response_start called)
     */
    int response_started;
    
    /*
     * reference counter
     */
    int ref;
} ProxyRequest;

static void proxy_unref(ProxyRequest *proxy) {
    if(--proxy->ref == 0) {
        http_client_free(proxy->client);
        free(proxy->read_buf);
        free(proxy);
    }
}

static int proxy_response_start(HttpClient *client, int status, char *message, void *userdata) {
    ProxyRequest *proxy = userdata;
    
    if(status == 101) {
        pblock_fr("connection", proxy->response_header_rewrite, TRUE);
        proxy->rq->rq_attr.keep_alive = 0;
    }
    
    HeaderArray *headers = client->response_headers;
    while(headers) {
        for(int i=0;i<headers->len;i++) {
            cxmutstr name = headers->headers[i].name;
            cxmutstr value = headers->headers[i].value;
            // NSAPI uses lower case header names internally
            for(int c=0;c<name.length;c++) {
                name.ptr[c] = tolower(name.ptr[c]);
            }
            // HttpClient does not 0-terminate strings
            name.ptr[name.length] = 0;
            // check if this header should be modified
            char *rewrite = pblock_findval(name.ptr, proxy->response_header_rewrite);
            if(rewrite) {
                value = cx_mutstr(rewrite);
                if(value.length == 0) {
                    // empty header value -> skip
                    continue;
                }
            }
            
            // add header to response
            pblock_nvlinsert(name.ptr, name.length, value.ptr, value.length, proxy->rq->srvhdrs);
        }
        headers = headers->next;
    }
    
    protocol_status(proxy->sn, proxy->rq, status, message);
    protocol_start_response(proxy->sn, proxy->rq);
    proxy->response_started = 1;
      
    return 0;
}

static void proxy_response_finished(HttpClient *client, void *userdata) {
    ProxyRequest *proxy = userdata;
    
    int ret = REQ_PROCEED;
    if(!proxy->response_started) {
        protocol_status(proxy->sn, proxy->rq, 502, NULL);
        ret = REQ_ABORTED;
    }
    
    if(client->last_event != &proxy->event && proxy->event.fn != NULL) {
        if(!ev_remove_poll(client->ev, proxy->event.object)) {
            proxy_unref(proxy);
        } else {
            log_ereport(LOG_FAILURE, "proxy_response_finished: cannot remove poll");
        }
        proxy->event.finish = NULL;
    }
    
    // return to nsapi threadpool
    net_setnonblock(proxy->sn->csd, 0);
    nsapi_function_return(proxy->sn, proxy->rq, ret);
    
    proxy_unref(proxy);
}

static int proxy_request_read_event(EventHandler *eh, Event *event) {
    ProxyRequest *proxy = event->cookie;
    if(http_client_process(proxy->client, event)) {
        if(proxy->client->error == 0) {
            return 1;
        }
    }
    return 0;
}

static int proxy_request_read_finished(EventHandler *eh, Event *event) {
    ProxyRequest *proxy = event->cookie;
    proxy_unref(proxy);
    return 0;
}

static ssize_t proxy_request_read(HttpClient *client, void *buf, size_t nbytes, void *userdata) {
    ProxyRequest *proxy = userdata;
    int ret = netbuf_getbytes(proxy->sn->inbuf, buf, nbytes);
    if(ret == NETBUF_EOF) {
        ret = 0;
    } else if(ret < 0) {
        IOStream *st = proxy->sn->csd;
        if(st->io_errno == EWOULDBLOCK) {
            ret = HTTP_CLIENT_CALLBACK_WOULD_BLOCK;
            // is there already an poll event for csd?
            if(!proxy->event.fn) {
                proxy->ref++;
                proxy->event.cookie = proxy;
                proxy->event.fn = proxy_request_read_event;
                proxy->event.finish = proxy_request_read_finished;
                if(event_pollin(client->ev, st, &proxy->event)) {
                    ret = HTTP_CLIENT_CALLBACK_ERROR;
                }
            }
        } else {
            ret = HTTP_CLIENT_CALLBACK_ERROR;
        }
    }
    
    return ret;
}

static ssize_t proxy_response_write(HttpClient *client, void *buf, size_t nbytes, void *userdata) {
    ProxyRequest *proxy = userdata;
    ssize_t ret = net_write(proxy->sn->csd, buf, nbytes);
    if(ret == NETBUF_EOF) {
        ret = HTTP_CLIENT_CALLBACK_ERROR;
    } else if(ret < 0) {
        IOStream *st = proxy->sn->csd;
        if(st->io_errno == EWOULDBLOCK) {
            // Is there already an poll event for csd?
            if(client->last_event == &proxy->event) {
                // Yes, there is already an event, and it even is the event that triggert this call.
                // Make sure it is a write event
                proxy->event.events = EVENT_POLLOUT;
                return HTTP_CLIENT_CALLBACK_WOULD_BLOCK;
            } else if(proxy->event.fn != NULL) {
                // There is an inactive event, probably a read event
                // Remove the old event before we add a write event
                if(ev_remove_poll(client->ev, proxy->event.object)) {
                    log_ereport(LOG_FAILURE, "proxy_response_write: cannot remove poll");
                    return HTTP_CLIENT_CALLBACK_ERROR;
                }
            } else {
                proxy->ref++;
                proxy->event.cookie = proxy;
                proxy->event.fn = proxy_request_read_event;
                proxy->event.finish = proxy_request_read_finished;
            }
            // Add write event
            if(event_pollout(client->ev, proxy->sn->csd, &proxy->event)) {
                proxy_unref(proxy);
                log_ereport(LOG_FAILURE, "proxy_response_write: cannot add write poll");
            }
            return HTTP_CLIENT_CALLBACK_ERROR;
        }
    }
    return ret;
}

static ssize_t proxy_ws_write(HttpClient *client, void *buf, size_t nbytes, void *userdata) {
    ProxyRequest *proxy = userdata;
    ssize_t w = net_write(proxy->sn->csd, buf, nbytes);
    if(w < 0) {
        IOStream *stream = proxy->sn->csd;
        return stream->io_errno == EWOULDBLOCK ? HTTP_CLIENT_CALLBACK_WOULD_BLOCK : HTTP_CLIENT_CALLBACK_ERROR;
    }
    return w;
}

static int proxy_ws_msg_ready(HttpClient *client, void *userdata) {
    ProxyRequest *proxy = userdata;
    IOStream *st = proxy->sn->csd;
    
    if(proxy->read_buf == NULL) {
        // proxy not initialized for websockets yet
        
        st->setmode(st, IO_MODE_RAW);
        
        proxy->read_buf_alloc = 1024*64;
        proxy->read_buf = malloc(proxy->read_buf_alloc);
        if(!proxy->read_buf) {
            return 1;
        }
        
        proxy->event.cookie = proxy;
        proxy->event.fn = proxy_request_read_event;
        proxy->event.finish = proxy_request_read_finished;
        if(event_pollin(client->ev, st, &proxy->event)) {
            return 1;
        }
        proxy->ref++;
    }
    
    // is there still data in the read buffer, that needs to be transfered to
    // the HttpClient message buffer?
    while(proxy->read_buf_pos < proxy->read_buf_size) {
        char *msg = proxy->read_buf + proxy->read_buf_pos;
        size_t msglen = proxy->read_buf_size - proxy->read_buf_pos;
        int n = http_client_add_message(client, msg, msglen);
        if(n <= 0) {
            return 0; // message buffer not ready
        }
        proxy->read_buf_pos += n;
    }
    // readbuf flushed, reset
    proxy->read_buf_size = 0;
    proxy->read_buf_pos = 0;
    
    while(proxy->read_buf_size < proxy->read_buf_alloc) {
        ssize_t r = net_read(proxy->sn->csd, proxy->read_buf + proxy->read_buf_size, proxy->read_buf_alloc - proxy->read_buf_size);
        if(r <= 0) {
            if(proxy->read_buf_size > 0) {
                // we have read some data, the next proxy_ws_msg_ready will
                // transfer this to the message buffer
                return proxy_ws_msg_ready(client, userdata);
            } else {
                // no data read, return
                return st->io_errno == EWOULDBLOCK ? 0 : 1; // return 1 in case of an error
            }
        }
        proxy->read_buf_size += r;
    }
    
    // do this again, to add the now filled buffer to the HttpClient message buffer
    // http_client_add_message or net_read should fail at some point
    return proxy_ws_msg_ready(client, userdata);
}

int http_reverse_proxy_service(pblock *param, Session *sn, Request *rq) {
    EventHandler *ev = sn->ev;
    const char *method = pblock_findkeyval(pb_key_method, rq->reqpb);
    const char *uri = pblock_findkeyval(pb_key_uri, rq->reqpb);
    const char *query = pblock_findkeyval(pb_key_query, rq->reqpb);
    
    char *host_header = pblock_findval("host-header", param);
    
    const char *location = pblock_findval("location", param);
    if(location) {
        cxstring loc = cx_str(location);
        if(cx_strprefix(uri, loc)) {
            if(loc.length > 0 && loc.ptr[loc.length-1] == '/') {
                uri += loc.length - 1;
            } else {
                uri += loc.length;
            }
            if(uri[0] == '\0') {
                uri = "/";
            }
        } else {
            // location does not match
            return REQ_NOACTION;
        }
    }
    
    const char *server_url = pblock_findval("proxy-url", rq->vars);
    if(!server_url) {
        server_url = pblock_findval("url", param);
        if(!server_url) {
            log_ereport(LOG_MISCONFIG, "reverse-proxy: missing server url");
            return REQ_ABORTED;
        }
    }
    WSUri srv_url;
    if(!util_parse_uri(server_url, &srv_url)) {
        log_ereport(LOG_MISCONFIG, "reverse-proxy: invalid server url: %s", server_url);
        return REQ_ABORTED;
    }
    
    char srvport[16];
    snprintf(srvport, 16, "%d", (int)srv_url.port);
    
    char srvhost_static[256];
    char *srvhost;
    if(srv_url.hostlen + 10 < 256) {
        memcpy(srvhost_static, srv_url.host, srv_url.hostlen);
        srvhost_static[srv_url.hostlen] = 0;
        srvhost = srvhost_static;
    } else {
        srvhost = pool_malloc(sn->pool, srv_url.hostlen + 10);
        if(!srvhost) {
            return REQ_ABORTED;
        }
        memcpy(srvhost, srv_url.host, srv_url.hostlen);
        srvhost[srv_url.hostlen] = 0;
    }
    
    // build the new uri using <srv_url.path><uri>?<query>
    if(srv_url.pathlen > 0 && srv_url.path[srv_url.pathlen-1] == '/') {
        srv_url.pathlen--;
    }
    
    cxmutstr new_uri = CX_NULLSTR;
    size_t uri_len = strlen(uri);
    size_t query_len = query ? strlen(query) : 0;
    size_t new_uri_alloc = ((uri_len + query_len) * 3) + 2 + srv_url.pathlen;
    new_uri.ptr = pool_malloc(sn->pool, new_uri_alloc);
    if(!new_uri.ptr) {
        return REQ_ABORTED;
    }
    
    char *s = new_uri.ptr;
    size_t s_len = new_uri_alloc;
    if(srv_url.pathlen > 0) {
        memcpy(s, srv_url.path, srv_url.pathlen);
        s += srv_url.pathlen;
        new_uri.length = srv_url.pathlen;
        s_len -= srv_url.pathlen;
    }
    
    new_uri.length += util_uri_escape_s(s, s_len, uri);
    if(new_uri.length > 0 && query_len > 0) {
        new_uri.ptr[new_uri.length] = '?';
        memcpy(new_uri.ptr + new_uri.length + 1, query, query_len + 1);
        new_uri.length += query_len + 1;
    }
    
    // remove some response headers, that were previously set by ObjectType
    // or other SAFs
    pblock_removekey(pb_key_content_type, rq->srvhdrs);
    
    ProxyRequest *proxy = malloc(sizeof(ProxyRequest));
    if(!proxy) {
        return REQ_ABORTED;
    }
    memset(proxy, 0, sizeof(ProxyRequest));
    proxy->sn = sn;
    proxy->rq = rq;
    proxy->request_header_rewrite = pblock_create_pool(sn->pool, 16);
    proxy->response_header_rewrite = pblock_create_pool(sn->pool, 16);
    proxy->response_started = 0;
    proxy->read_buf = NULL;
    proxy->read_buf_alloc = 0;
    proxy->read_buf_size = 0;
    proxy->read_buf_pos = 0;
    proxy->ref = 1;
    
    // Some request/response headers should be removed or altered
    // An empty string means, the header should be removed
    if(!host_header) {
        pblock_nvinsert("host", "", proxy->request_header_rewrite);
    }
    //pblock_nvinsert("connection", "", proxy->request_header_rewrite);
    pblock_nvinsert("transfer-encoding", "", proxy->request_header_rewrite);
    pblock_nvinsert("content-length", "", proxy->request_header_rewrite);
    pblock_nvinsert("server", "", proxy->response_header_rewrite);
    pblock_nvinsert("connection", "", proxy->response_header_rewrite);
    
    // setup HttpClient
    HttpClient *client = http_client_new(ev);
    if(!client) {
        return REQ_ABORTED;
    }
    proxy->client = client;
    
    if(http_client_set_method(client, method)) {
        http_client_free(client);
        return REQ_ABORTED;
    }
    
    if(http_client_set_uri_len(client, new_uri.ptr, new_uri.length)) {
        http_client_free(client);
        return REQ_ABORTED;
    }
    
    // set server address
    struct addrinfo hints = { 0 };
    hints.ai_family = AF_INET; //AF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;
    struct addrinfo *srv_addr;
    if(getaddrinfo(srvhost, srvport, &hints, &srv_addr)) {
        http_client_free(client);
        log_ereport(LOG_FAILURE, "reverse-proxy: getaddr failed for address %s: %s", srvhost, strerror(errno));
        return REQ_ABORTED;
    }
    
    WSBool addr_set = FALSE;
    struct addrinfo *addr = srv_addr;
    while(addr) {
        if(addr->ai_family == AF_INET || addr->ai_family == AF_INET6) {
            if(!http_client_set_addr(client, addr->ai_family, addr->ai_addr, addr->ai_addrlen)) {
                addr_set = TRUE;
            }
            break;
        }
        
        addr = addr->ai_next;
    }
    freeaddrinfo(srv_addr);
    if(!addr_set) {
        http_client_free(client);
        return REQ_ABORTED;
    }
    
    // prepare host header
    if(!host_header) {
        if(!((srv_url.scheme_num == WS_URI_HTTP && srv_url.port == 80) || 
             (srv_url.scheme_num == WS_URI_HTTPS && srv_url.port == 443)))
        {
            // we have reserved enough space for the port
            srvhost[srv_url.hostlen] = ':';
            snprintf(srvhost + srv_url.hostlen + 1, 8, "%d", (int)srv_url.port);
        }
        host_header = srvhost;
    }
    
    // add host header
    if(http_client_add_request_header(client, cx_mutstr("host"), cx_mutstr(host_header))) {
        http_client_free(client);
        return REQ_ABORTED;
    }
    
    // add request headers to the client
    CxIterator i = pblock_iterator(rq->headers);
    cx_foreach(pb_entry*, entry, i) {
        cxmutstr header_value;
        char *rewrite_header = pblock_findval(entry->param->name, proxy->request_header_rewrite);
        if(rewrite_header) {
            if(!strcmp(entry->param->name, "transfer-encoding")) {
                if(!strcmp(entry->param->value, "chunked")) {
                    // enable chunked transfer encoding
                    if(http_client_enable_chunked_transfer_encoding(client)) {
                        http_client_free(client);
                        return REQ_ABORTED;
                    }
                    continue;
                }
            } else if(!strcmp(entry->param->name, "content-length")) {
                long long contentlength;
                if(!cx_strtoll(cx_str(entry->param->value), &contentlength, 10)) {
                    if(http_client_set_content_length(client, contentlength)) {
                        http_client_free(client);
                        return REQ_ABORTED;
                    }
                } else {
                    // illegal content-length
                    protocol_status(sn, rq, 400, NULL);
                    http_client_free(client);
                    return REQ_ABORTED;
                }
                continue;
            } else {
                // static header rewrite or remove header if it is empty
                header_value = cx_mutstr(rewrite_header);
                if(header_value.length == 0) {
                    continue;
                }
            }
        } else {
            header_value = cx_mutstr(entry->param->value);
        }
        
        if(http_client_add_request_header(client, cx_mutstr(entry->param->name), header_value)) {
            http_client_free(client);
            return REQ_ABORTED;
        }
    }
    
    client->request_body_read = proxy_request_read;
    client->request_body_read_userdata = proxy;
    client->response_start = proxy_response_start;
    client->response_start_userdata = proxy;
    client->response_body_write = proxy_response_write;
    client->response_body_write_userdata = proxy;
    client->response_finished = proxy_response_finished;
    client->response_finished_userdata = proxy;
    client->ws_write = proxy_ws_write;
    client->ws_write_userdata = proxy;
    client->ws_msg_ready = proxy_ws_msg_ready;
    client->ws_msg_ready_userdata = proxy;
    
    net_setnonblock(sn->csd, 1);
    if(http_client_start(client)) {
        net_setnonblock(sn->csd, 0);
        http_client_free(client);
        return REQ_ABORTED;
    }
    
    return REQ_PROCESSING;
}