udo/uce
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add archive helpers and harden task runtime

Browse Source
This commit is contained in:
udo
2026-05-21 00:00:23 +00:00
parent 9f7625c7fd
commit 02e153a6a7
71 changed files with 12817 additions and 305 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/fastcgi/src/fcgicc.cc
+348 -162
View File
@@ -44,10 +44,42 @@
#include <netinet/in.h> // sockaddr_in, INADDR_*
#include <sys/select.h> // select, fd_set, FD_*, timeval
#include <sys/socket.h> // socket, bind, accept, listen, sockaddr, AF_*, SOCK_*
#include <sys/stat.h> // mkdir
#include <sys/un.h> // sockaddr_un
#include "../fastcgi_devkit/fastcgi.h"
namespace {
struct TransportLimits {
u64 max_client_connections = 256;
u64 max_http_header_bytes = 16 * 1024;
u64 max_http_body_bytes = 1024 * 1024;
u64 max_websocket_frame_bytes = 1024 * 1024;
u64 max_websocket_message_bytes = 1024 * 1024;
u64 max_websocket_output_bytes = 4 * 1024 * 1024;
f64 http_request_timeout_seconds = 15.0;
f64 connection_idle_timeout_seconds = 120.0;
u64 max_http_buffer_bytes() const
{
return(max_http_header_bytes + max_http_body_bytes);
}
u64 max_websocket_buffer_bytes() const
{
return(max_websocket_message_bytes + 16 * 1024);
}
};
const TransportLimits& transport_limits()
{
static const TransportLimits limits;
return(limits);
}
}
static String
make_http_text_response(String status_line, String body, String extra_headers = "")
{
@@ -75,6 +107,27 @@ is_valid_close_code(u16 status_code)
return(false);
}
static void
ensure_parent_directories(const std::string& path)
{
std::string::size_type slash = path.rfind('/');
if(slash == std::string::npos || slash == 0)
return;
std::string current;
std::string directory = path.substr(0, slash);
for(std::string::size_type i = 1; i <= directory.length(); ++i)
{
if(i != directory.length() && directory[i] != '/')
continue;
current = directory.substr(0, i);
if(current == "")
continue;
if(::mkdir(current.c_str(), 0775) == -1 && errno != EEXIST)
throw std::runtime_error("mkdir() failed for Unix socket directory");
}
}
static void
set_socket_nonblocking(int socket_handle)
{
@@ -132,6 +185,15 @@ FastCGIServer::listen_http(unsigned tcp_port)
return server_socket;
}
int
FastCGIServer::listen_cli(const std::string& local_path)
{
int server_socket = listen(local_path);
server_socket_types[server_socket] = 'C';
printf("(P) CLI command socket ready at %s\n", local_path.c_str());
return server_socket;
}
int
FastCGIServer::listen(unsigned tcp_port)
{
@@ -188,6 +250,7 @@ FastCGIServer::listen(const std::string& local_path)
std::memcpy(sa.sun_path, local_path.data(), size);
ensure_parent_directories(local_path);
file_unlink(local_path);
try {
if (bind(server_socket, (struct sockaddr*)&sa,
@@ -250,9 +313,95 @@ FastCGIServer::close_http_listeners()
}
}
bool
FastCGIServer::is_http_like_type(char type)
{
return(type == 'H' || type == 'C');
}
FastCGIServer::Connection*
FastCGIServer::open_client_connection(int server_socket, int client_socket)
{
set_socket_nonblocking(client_socket);
printf("Opening socket %i\n", client_socket);
Connection* connection = new Connection();
connection->client_socket = client_socket;
connection->server_socket = server_socket;
connection->type = server_socket_types[server_socket];
connection->opened_at = time_precise();
connection->last_activity_at = connection->opened_at;
client_sockets[client_socket] = connection;
if(is_http_like_type(connection->type))
{
FastCGIRequest* new_request = new FastCGIRequest();
new_request->resources.client_socket = client_socket;
new_request->resources.server_socket = server_socket;
new_request->stats.time_init = connection->opened_at;
connection->requests[client_socket] = new_request;
}
return(connection);
}
bool
FastCGIServer::reject_http_connection(Connection& connection, String status_line, String body, String extra_headers)
{
connection.output_buffer += make_http_text_response(status_line, body, extra_headers);
connection.close_socket = true;
return(false);
}
void
FastCGIServer::enforce_connection_timeouts(Connection& connection)
{
if(connection.close_socket)
return;
const TransportLimits& limits = transport_limits();
f64 now = time_precise();
if(is_http_like_type(connection.type) && !connection.is_websocket && !connection.requests.empty())
{
FastCGIRequest* pending_request = connection.requests.begin()->second;
if(!pending_request->flags.input_closed && now - connection.opened_at > limits.http_request_timeout_seconds)
{
reject_http_connection(connection, "HTTP/1.1 408 Request Timeout", "request timed out\n");
return;
}
}
if(now - connection.last_activity_at > limits.connection_idle_timeout_seconds)
connection.close_socket = true;
}
bool
FastCGIServer::queue_websocket_frame(Connection& connection, String frame)
{
if(connection.output_buffer.length() + frame.length() > transport_limits().max_websocket_output_bytes)
{
fail_websocket_connection(connection, 1013, "websocket output queue is full");
return(false);
}
connection.output_buffer += frame;
return(true);
}
bool
FastCGIServer::queue_websocket_payload(Connection& connection, String message, bool binary)
{
if(message.length() > transport_limits().max_websocket_message_bytes)
return(false);
return(queue_websocket_frame(connection, ws_encode_frame(message, binary ? 0x2 : 0x1)));
}
void
FastCGIServer::close_websocket_connection(Connection& connection, u16 status_code, String reason)
{
if(!is_valid_close_code(status_code))
status_code = 1002;
if(reason.length() > 123)
reason = reason.substr(0, 123);
if(!ws_is_valid_utf8(reason))
reason = "";
if(!connection.close_socket)
connection.output_buffer += ws_close_frame(status_code, reason);
connection.close_socket = true;
@@ -261,6 +410,12 @@ FastCGIServer::close_websocket_connection(Connection& connection, u16 status_cod
void
FastCGIServer::fail_websocket_connection(Connection& connection, u16 status_code, String reason)
{
// Drop stale application frames on failure so slow peers cannot keep a large
// queued buffer alive. Preserve an unsent 101 response, though: if the client
// pipelined a bad WebSocket frame immediately after the HTTP upgrade, the
// close frame must still follow the accepted upgrade response.
if(!str_starts_with(connection.output_buffer, "HTTP/1.1 101 Switching Protocols\r\n"))
connection.output_buffer = "";
close_websocket_connection(connection, status_code, reason);
}
@@ -339,20 +494,14 @@ FastCGIServer::process(int timeout_ms)
throw std::runtime_error("accept() failed");
}
set_socket_nonblocking(client_socket);
printf("Opening socket %i\n", client_socket);
client_sockets[client_socket] = new Connection();
client_sockets[client_socket]->client_socket = client_socket;
client_sockets[client_socket]->server_socket = socket_handle;
client_sockets[client_socket]->type = server_socket_types[socket_handle];
if(client_sockets[client_socket]->type == 'H')
if(client_sockets.size() >= transport_limits().max_client_connections)
{
FastCGIRequest* new_request = new FastCGIRequest();
new_request->resources.client_socket = client_socket;
new_request->resources.server_socket = socket_handle;
new_request->stats.time_init = time_precise();
client_sockets[client_socket]->requests[client_socket] = new_request;
printf("(!) rejecting socket %i: too many clients\n", client_socket);
close(client_socket);
continue;
}
open_client_connection(socket_handle, client_socket);
}
}
@@ -361,6 +510,7 @@ FastCGIServer::process(int timeout_ms)
{
int read_socket = it->first;
Connection* connection = it->second;
enforce_connection_timeouts(*connection);
if(FD_ISSET(read_socket, &fs_read))
{
@@ -378,15 +528,12 @@ FastCGIServer::process(int timeout_ms)
{
connection->close_socket = true;
}
else if(connection->type == 'H' && connection->input_buffer != "")
else if(is_http_like_type(connection->type) && connection->input_buffer != "")
{
process_http_request(
*client_sockets[connection->client_socket]->requests[connection->client_socket],
connection->input_buffer
);
if(connection->close_socket || !connection->output_buffer.empty())
process_http_like_connection_input(*connection);
if(!connection->is_websocket && (connection->close_socket || !connection->output_buffer.empty()))
connection->input_buffer = "";
else
else if(!connection->is_websocket)
connection->close_socket = true;
}
else
@@ -396,22 +543,16 @@ FastCGIServer::process(int timeout_ms)
}
else
{
connection->last_activity_at = time_precise();
connection->input_buffer.append(buffer, read_result);
if(connection->type == 'H')
if(is_http_like_type(connection->type))
{
if(connection->is_websocket)
{
process_websocket_input(*connection);
}
if(!connection->is_websocket && connection->input_buffer.length() > transport_limits().max_http_buffer_bytes())
reject_http_connection(*connection, "HTTP/1.1 413 Payload Too Large", "request is too large\n");
else
{
process_http_request(
*client_sockets[connection->client_socket]->requests[connection->client_socket],
connection->input_buffer
);
if(connection->close_socket || !connection->output_buffer.empty())
connection->input_buffer = "";
}
process_http_like_connection_input(*connection);
if(!connection->is_websocket && (connection->close_socket || !connection->output_buffer.empty()))
connection->input_buffer = "";
}
else
{
@@ -420,6 +561,9 @@ FastCGIServer::process(int timeout_ms)
}
}
if(connection->is_websocket && connection->output_buffer.length() > transport_limits().max_websocket_output_bytes)
fail_websocket_connection(*connection, 1013, "websocket output queue is full");
if(!connection->output_buffer.empty() && FD_ISSET(read_socket, &fs_write))
{
if(connection->type == 'F')
@@ -428,6 +572,9 @@ FastCGIServer::process(int timeout_ms)
goto close_socket;
}
if(connection->is_websocket && connection->output_buffer.empty() && !connection->input_buffer.empty() && !connection->close_socket)
process_websocket_input(*connection);
if(connection->close_socket && connection->output_buffer.empty())
{
close_socket:
@@ -452,141 +599,160 @@ FastCGIServer::process(int timeout_ms)
}
}
void
FastCGIServer::process_http_request(FastCGIRequest& request, String& data)
bool
FastCGIServer::parse_http_message(FastCGIRequest& request, String& data)
{
Connection* connection = client_sockets[request.resources.client_socket];
const TransportLimits& limits = transport_limits();
auto header_end = data.find("\r\n\r\n");
if(header_end == String::npos)
return;
{
if(data.length() > limits.max_http_header_bytes)
reject_http_connection(*connection, "HTTP/1.1 431 Request Header Fields Too Large", "request headers are too large\n");
return(false);
}
if(header_end > limits.max_http_header_bytes)
return(reject_http_connection(*connection, "HTTP/1.1 431 Request Header Fields Too Large", "request headers are too large\n"));
if(request.params.size() == 0)
{
request.params = split_http_headers(data.substr(0, header_end));
request.flags.params_closed = true;
if(request.params["HTTP_SCRIPT_FILENAME"] != "")
request.params["SCRIPT_FILENAME"] = request.params["HTTP_SCRIPT_FILENAME"];
else if(request.params["SCRIPT_FILENAME"] == "" && request.params["DOCUMENT_URI"] != "")
{
String document_root = first(request.params["DOCUMENT_ROOT"], cwd_get());
if(document_root.length() > 1 && document_root[document_root.length()-1] == '/')
document_root.resize(document_root.length()-1);
request.params["DOCUMENT_ROOT"] = document_root;
request.params["SCRIPT_FILENAME"] = document_root + request.params["DOCUMENT_URI"];
}
}
u64 content_length = int_val(first(request.params["CONTENT_LENGTH"], "0"));
if(content_length > limits.max_http_body_bytes)
return(reject_http_connection(*connection, "HTTP/1.1 413 Payload Too Large", "request body is too large\n"));
u64 request_size = header_end + 4 + content_length;
if(request_size > limits.max_http_buffer_bytes())
return(reject_http_connection(*connection, "HTTP/1.1 413 Payload Too Large", "request is too large\n"));
if(data.length() < request_size)
return;
return(false);
request.in = data.substr(header_end + 4, content_length);
request.flags.input_closed = true;
data.erase(0, request_size);
return(true);
}
void
FastCGIServer::process_http_like_connection_input(Connection& connection)
{
if(connection.type == 'H' && connection.is_websocket)
{
process_websocket_input(connection);
return;
}
FastCGIRequest& request = *connection.requests[connection.client_socket];
if(connection.type == 'C')
process_cli_request(request, connection.input_buffer);
else
process_http_request(request, connection.input_buffer);
}
void
FastCGIServer::process_cli_request(FastCGIRequest& request, String& data)
{
if(!parse_http_message(request, data))
return;
Connection* connection = client_sockets[request.resources.client_socket];
if(!on_cli_complete)
{
connection->output_buffer += make_http_text_response(
"HTTP/1.1 500 Internal Server Error",
"UCE CLI dispatcher is not configured\n"
);
}
else
{
on_cli_complete(request);
assemble_output_buffer(request, connection);
}
connection->close_socket = true;
}
bool
FastCGIServer::validate_websocket_upgrade(FastCGIRequest& request, Connection& connection)
{
String request_method = trim(request.params["REQUEST_METHOD"]);
String request_protocol = trim(request.params["SERVER_PROTOCOL"]);
String websocket_connection = to_lower(request.params["HTTP_CONNECTION"]);
String websocket_host = trim(request.params["HTTP_HOST"]);
String websocket_key = trim(request.params["HTTP_SEC_WEBSOCKET_KEY"]);
String websocket_version = trim(request.params["HTTP_SEC_WEBSOCKET_VERSION"]);
if(request_method != "GET")
return(reject_http_connection(connection, "HTTP/1.1 405 Method Not Allowed", "websocket upgrades require GET"));
if(request_protocol != "HTTP/1.1")
return(reject_http_connection(connection, "HTTP/1.1 400 Bad Request", "websocket upgrades require HTTP/1.1"));
if(websocket_host == "")
return(reject_http_connection(connection, "HTTP/1.1 400 Bad Request", "missing Host header"));
if(websocket_connection.find("upgrade") == String::npos)
return(reject_http_connection(connection, "HTTP/1.1 400 Bad Request", "missing Connection: Upgrade header"));
if(websocket_key == "")
return(reject_http_connection(connection, "HTTP/1.1 400 Bad Request", "missing Sec-WebSocket-Key header"));
if(!ws_is_valid_client_key(websocket_key))
return(reject_http_connection(connection, "HTTP/1.1 400 Bad Request", "invalid Sec-WebSocket-Key header"));
if(websocket_version == "")
return(reject_http_connection(connection, "HTTP/1.1 400 Bad Request", "missing Sec-WebSocket-Version header"));
if(websocket_version != "13")
return(reject_http_connection(connection, "HTTP/1.1 426 Upgrade Required", "unsupported websocket version", "Sec-WebSocket-Version: 13\r\n"));
return(true);
}
void
FastCGIServer::begin_websocket_upgrade(FastCGIRequest& request, Connection& connection, String& data)
{
connection.is_websocket = true;
connection.websocket_connection_id = std::to_string(getpid()) + ":" + std::to_string(connection.client_socket);
connection.websocket_scope = first(
request.params["SCRIPT_FILENAME"],
request.params["DOCUMENT_URI"],
request.params["REQUEST_URI"]
);
request.resources.is_websocket = true;
request.resources.websocket_connection_id = connection.websocket_connection_id;
request.resources.websocket_scope = connection.websocket_scope;
request.resources.websocket_connection_state = &connection.websocket_state;
request.connection.set_reference(&connection.websocket_state);
connection.output_buffer +=
"HTTP/1.1 101 Switching Protocols\r\n"
"Upgrade: websocket\r\n"
"Connection: Upgrade\r\n"
"Sec-WebSocket-Accept: " + ws_make_accept_key(trim(request.params["HTTP_SEC_WEBSOCKET_KEY"])) + "\r\n\r\n";
if(!data.empty())
process_websocket_input(connection);
}
void
FastCGIServer::process_http_request(FastCGIRequest& request, String& data)
{
if(!parse_http_message(request, data))
return;
if(request.params["HTTP_SCRIPT_FILENAME"] != "")
request.params["SCRIPT_FILENAME"] = request.params["HTTP_SCRIPT_FILENAME"];
else if(request.params["SCRIPT_FILENAME"] == "" && request.params["DOCUMENT_URI"] != "")
{
String document_root = first(request.params["DOCUMENT_ROOT"], cwd_get());
if(document_root.length() > 1 && document_root[document_root.length()-1] == '/')
document_root.resize(document_root.length()-1);
request.params["DOCUMENT_ROOT"] = document_root;
request.params["SCRIPT_FILENAME"] = document_root + request.params["DOCUMENT_URI"];
}
if(to_lower(request.params["HTTP_UPGRADE"]) == "websocket")
{
Connection* connection = client_sockets[request.resources.client_socket];
String request_method = trim(request.params["REQUEST_METHOD"]);
String request_protocol = trim(request.params["SERVER_PROTOCOL"]);
String websocket_connection = to_lower(request.params["HTTP_CONNECTION"]);
String websocket_host = trim(request.params["HTTP_HOST"]);
String websocket_key = trim(request.params["HTTP_SEC_WEBSOCKET_KEY"]);
String websocket_version = trim(request.params["HTTP_SEC_WEBSOCKET_VERSION"]);
if(request_method != "GET")
{
connection->output_buffer += make_http_text_response(
"HTTP/1.1 405 Method Not Allowed",
"websocket upgrades require GET"
);
connection->close_socket = true;
return;
}
if(request_protocol != "HTTP/1.1")
{
connection->output_buffer += make_http_text_response(
"HTTP/1.1 400 Bad Request",
"websocket upgrades require HTTP/1.1"
);
connection->close_socket = true;
return;
}
if(websocket_host == "")
{
connection->output_buffer += make_http_text_response(
"HTTP/1.1 400 Bad Request",
"missing Host header"
);
connection->close_socket = true;
return;
}
if(websocket_connection.find("upgrade") == String::npos)
{
connection->output_buffer += make_http_text_response(
"HTTP/1.1 400 Bad Request",
"missing Connection: Upgrade header"
);
connection->close_socket = true;
return;
}
if(websocket_key == "")
{
connection->output_buffer += make_http_text_response(
"HTTP/1.1 400 Bad Request",
"missing Sec-WebSocket-Key header"
);
connection->close_socket = true;
return;
}
if(!ws_is_valid_client_key(websocket_key))
{
connection->output_buffer += make_http_text_response(
"HTTP/1.1 400 Bad Request",
"invalid Sec-WebSocket-Key header"
);
connection->close_socket = true;
return;
}
if(websocket_version == "")
{
connection->output_buffer += make_http_text_response(
"HTTP/1.1 400 Bad Request",
"missing Sec-WebSocket-Version header"
);
connection->close_socket = true;
return;
}
if(websocket_version != "" && websocket_version != "13")
{
connection->output_buffer += make_http_text_response(
"HTTP/1.1 426 Upgrade Required",
"unsupported websocket version",
"Sec-WebSocket-Version: 13\r\n"
);
connection->close_socket = true;
return;
}
connection->is_websocket = true;
connection->websocket_connection_id = std::to_string(getpid()) + ":" + std::to_string(connection->client_socket);
connection->websocket_scope = first(
request.params["SCRIPT_FILENAME"],
request.params["DOCUMENT_URI"],
request.params["REQUEST_URI"]
);
request.resources.is_websocket = true;
request.resources.websocket_connection_id = connection->websocket_connection_id;
request.resources.websocket_scope = connection->websocket_scope;
request.resources.websocket_connection_state = &connection->websocket_state;
request.connection.set_reference(&connection->websocket_state);
connection->output_buffer +=
"HTTP/1.1 101 Switching Protocols\r\n"
"Upgrade: websocket\r\n"
"Connection: Upgrade\r\n"
"Sec-WebSocket-Accept: " + ws_make_accept_key(websocket_key) + "\r\n\r\n";
if(!data.empty())
process_websocket_input(*connection);
if(validate_websocket_upgrade(request, *connection))
begin_websocket_upgrade(request, *connection, data);
}
else
{
@@ -606,6 +772,13 @@ FastCGIServer::process_http_request(FastCGIRequest& request, String& data)
void
FastCGIServer::process_websocket_input(Connection& connection)
{
const TransportLimits& limits = transport_limits();
if(connection.input_buffer.length() > limits.max_websocket_buffer_bytes())
{
fail_websocket_connection(connection, 1009, "websocket input buffer is too large");
return;
}
while(!connection.input_buffer.empty())
{
WSFrame frame;
@@ -613,10 +786,13 @@ FastCGIServer::process_websocket_input(Connection& connection)
if(!frame.parse(connection.input_buffer, error))
{
if(error != "")
{
connection.output_buffer += ws_close_frame(1002, error);
connection.close_socket = true;
}
fail_websocket_connection(connection, 1002, error);
return;
}
if(frame.payload_length > limits.max_websocket_frame_bytes)
{
fail_websocket_connection(connection, 1009, "websocket frame is too large");
return;
}
@@ -645,6 +821,11 @@ FastCGIServer::process_websocket_input(Connection& connection)
return;
}
if(connection.websocket_fragment_buffer.length() + frame.payload.length() > limits.max_websocket_message_bytes)
{
fail_websocket_connection(connection, 1009, "websocket message is too large");
return;
}
connection.websocket_fragment_buffer += frame.payload;
if(!frame.is_final_fragment)
break;
@@ -672,6 +853,12 @@ FastCGIServer::process_websocket_input(Connection& connection)
return;
}
if(frame.payload.length() > limits.max_websocket_message_bytes)
{
fail_websocket_connection(connection, 1009, "websocket message is too large");
return;
}
if(frame.is_final_fragment)
{
if(frame.opcode == 0x1 && !ws_is_valid_utf8(frame.payload))
@@ -720,7 +907,8 @@ FastCGIServer::process_websocket_input(Connection& connection)
return;
}
case 0x9:
connection.output_buffer += ws_encode_frame(frame.payload, 0xA);
if(!queue_websocket_frame(connection, ws_encode_frame(frame.payload, 0xA)))
return;
break;
case 0xA:
break;
@@ -734,15 +922,11 @@ FastCGIServer::process_websocket_input(Connection& connection)
bool
FastCGIServer::websocket_send_to(String connection_id, String message, bool binary)
{
u8 opcode = binary ? 0x2 : 0x1;
for(auto& item : client_sockets)
{
Connection* connection = item.second;
if(connection->is_websocket && connection->websocket_connection_id == connection_id)
{
connection->output_buffer += ws_encode_frame(message, opcode);
return(true);
}
return(queue_websocket_payload(*connection, message, binary));
}
return(false);
}
@@ -751,7 +935,10 @@ u64
FastCGIServer::websocket_broadcast(String scope, String message, bool binary)
{
u64 sent = 0;
u8 opcode = binary ? 0x2 : 0x1;
if(message.length() > transport_limits().max_websocket_message_bytes)
return(sent);
String frame = ws_encode_frame(message, binary ? 0x2 : 0x1);
for(auto& item : client_sockets)
{
Connection* connection = item.second;
@@ -759,8 +946,8 @@ FastCGIServer::websocket_broadcast(String scope, String message, bool binary)
continue;
if(scope != "" && connection->websocket_scope != scope)
continue;
connection->output_buffer += ws_encode_frame(message, opcode);
sent += 1;
if(queue_websocket_frame(*connection, frame))
sent += 1;
}
return(sent);
}
@@ -789,8 +976,7 @@ FastCGIServer::websocket_close(String connection_id, u16 status_code, String rea
Connection* connection = item.second;
if(connection->is_websocket && connection->websocket_connection_id == connection_id)
{
connection->output_buffer += ws_close_frame(status_code, reason);
connection->close_socket = true;
close_websocket_connection(*connection, status_code, reason);
return(true);
}
}
src/fastcgi/src/fcgicc.h
+16 -1
View File
@@ -48,10 +48,12 @@ public:
std::function<int(FastCGIRequest&)> on_request = 0;
std::function<int(FastCGIRequest&)> on_data = 0;
std::function<int(FastCGIRequest&)> on_complete = 0;
std::function<int(FastCGIRequest&)> on_cli_complete = 0;
std::function<int(FastCGIRequest&, const String&, u8)> on_websocket_message = 0;
int listen(unsigned tcp_port);
int listen_http(unsigned tcp_port);
int listen_cli(const std::string& local_path);
int listen(const std::string& local_path);
void process(int timeout_ms = -1); // timeout_ms<0 blocks forever
@@ -75,7 +77,9 @@ public:
DTree websocket_state;
String websocket_fragment_buffer;
u8 websocket_fragment_opcode = 0;
char type = 'F'; // F = FastCGI, H = HttpServer
f64 opened_at = 0;
f64 last_activity_at = 0;
char type = 'F'; // F = FastCGI, H = HttpServer/WebSocket, C = CLI-over-HTTP Unix socket
};
typedef StringMap Pairs;
@@ -88,7 +92,15 @@ public:
void close_http_listeners();
void read_fgci(Connection&);
static bool is_http_like_type(char type);
Connection* open_client_connection(int server_socket, int client_socket);
void enforce_connection_timeouts(Connection& connection);
bool reject_http_connection(Connection& connection, String status_line, String body, String extra_headers = "");
bool queue_websocket_frame(Connection& connection, String frame);
bool queue_websocket_payload(Connection& connection, String message, bool binary = false);
void process_websocket_input(Connection&);
bool validate_websocket_upgrade(FastCGIRequest& request, Connection& connection);
void begin_websocket_upgrade(FastCGIRequest& request, Connection& connection, String& data);
void fail_websocket_connection(Connection& connection, u16 status_code, String reason = "");
void close_websocket_connection(Connection& connection, u16 status_code = 1000, String reason = "");
void dispatch_websocket_message(Connection& connection, RequestID request_id, String payload, u8 opcode);
@@ -105,7 +117,10 @@ public:
const std::string& input, unsigned char type);
static void assemble_output_buffer(FastCGIRequest& request, Connection* connection = 0);
int send_output_buffer(Connection& con);
bool parse_http_message(FastCGIRequest& request, String& data);
void process_http_request(FastCGIRequest& request, String& data);
void process_cli_request(FastCGIRequest& request, String& data);
void process_http_like_connection_input(Connection& connection);
};
src/lib/cli.cpp
+45
View File
@@ -0,0 +1,45 @@
#include "cli.h"
DTree cli_input(Request& context)
{
DTree input;
for(auto& item : context.get)
input[item.first] = item.second;
for(auto& item : context.post)
input[item.first] = item.second;
String content_type_info = context.params["CONTENT_TYPE"];
String content_type = to_lower(trim(nibble(content_type_info, ";")));
if(content_type == "application/json" || str_ends_with(content_type, "+json"))
{
String body = trim(context.in);
if(body != "")
{
DTree parsed = json_decode(body);
if(parsed.type == 'M')
{
for(auto& item : parsed._map)
input[item.first] = item.second;
}
else
{
input["_"] = parsed;
}
}
}
return(input);
}
String cli_arg(Request& context, String key, String default_value)
{
DTree input = cli_input(context);
DTree* value = input.key(key);
if(!value)
return(default_value);
String result = value->to_string();
if(result == "")
return(default_value);
return(result);
}
src/lib/cli.h
+4
View File
@@ -0,0 +1,4 @@
#pragma once
DTree cli_input(Request& context);
String cli_arg(Request& context, String key, String default_value = "");
src/lib/compiler.cpp
+92 -28
View File
@@ -13,6 +13,7 @@ const char* UCE_SETUP_SYMBOL = "__uce_set_current_request";
const char* UCE_RENDER_SYMBOL = "__uce_render";
const char* UCE_COMPONENT_SYMBOL = "__uce_component";
const char* UCE_WEBSOCKET_SYMBOL = "__uce_websocket";
const char* UCE_CLI_SYMBOL = "__uce_cli";
const char* UCE_ONCE_SYMBOL = "__uce_once";
const char* UCE_INIT_SYMBOL = "__uce_init";
const u64 UCE_UNIT_ABI_VERSION = 1;
@@ -690,6 +691,8 @@ void load_shared_unit(Request* context, SharedUnit* su)
dlerror();
su->on_websocket = (request_ref_handler)dlsym(su->so_handle, UCE_WEBSOCKET_SYMBOL);
dlerror();
su->on_cli = (request_ref_handler)dlsym(su->so_handle, UCE_CLI_SYMBOL);
dlerror();
su->on_once = (request_ref_handler)dlsym(su->so_handle, UCE_ONCE_SYMBOL);
dlerror();
su->on_init = (request_ref_handler)dlsym(su->so_handle, UCE_INIT_SYMBOL);
@@ -1191,7 +1194,8 @@ enum class UnitCallMacroKind
render,
component,
once,
init
init,
cli
};
struct UnitCallMacroTarget
@@ -1234,6 +1238,7 @@ void compiler_unload_failed_shared_unit(SharedUnit* su)
su->on_render = 0;
su->on_component = 0;
su->on_websocket = 0;
su->on_cli = 0;
su->on_once = 0;
su->on_init = 0;
}
@@ -1358,6 +1363,11 @@ UnitCallMacroTarget unit_call_macro_target(String function_name)
target.kind = UnitCallMacroKind::init;
return(target);
}
if(function_name == "CLI")
{
target.kind = UnitCallMacroKind::cli;
return(target);
}
return(target);
}
@@ -1465,6 +1475,8 @@ String compiler_missing_request_handler_message(UnitCallMacroKind kind, String h
}
if(kind == UnitCallMacroKind::once)
return("no ONCE() entry point");
if(kind == UnitCallMacroKind::cli)
return("no CLI() entry point");
if(kind == UnitCallMacroKind::init)
return("no INIT() entry point");
return("request handler not found");
@@ -1544,20 +1556,24 @@ request_ref_handler get_component_handler(SharedUnit* su, String render_name)
return(handler);
}
bool compiler_invoke_render(Request* context, String file_name, String render_name, String* error_out = 0)
bool compiler_invoke_loaded_request_handler(
Request* context,
SharedUnit* su,
request_ref_handler handler,
UnitCallMacroKind kind,
String handler_name,
bool count_request,
String runtime_error_status,
String* error_out = 0
)
{
auto su = compiler_load_shared_unit(context, file_name, "", false);
if(!su)
return(false);
if(!compiler_prepare_request_handler(context, su, error_out, true))
return(false);
auto handler = get_page_render_handler(su, render_name);
if(!handler)
{
if(error_out)
*error_out = compiler_missing_request_handler_message(UnitCallMacroKind::render, render_name);
*error_out = compiler_missing_request_handler_message(kind, handler_name);
return(false);
}
@@ -1565,37 +1581,46 @@ bool compiler_invoke_render(Request* context, String file_name, String render_na
context,
su,
handler,
compiler_is_request_entry_unit(context, su),
"uncaught exception during render"
count_request,
runtime_error_status
);
return(true);
}
bool compiler_invoke_render(Request* context, String file_name, String render_name, String* error_out = 0)
{
auto su = compiler_load_shared_unit(context, file_name, "", false);
if(!su)
return(false);
return(compiler_invoke_loaded_request_handler(
context,
su,
get_page_render_handler(su, render_name),
UnitCallMacroKind::render,
render_name,
compiler_is_request_entry_unit(context, su),
"uncaught exception during render",
error_out
));
}
bool compiler_invoke_component(Request* context, String file_name, String render_name, String* error_out = 0)
{
auto su = compiler_load_shared_unit(context, file_name, "", false);
if(!su)
return(false);
if(!compiler_prepare_request_handler(context, su, error_out, true))
return(false);
auto handler = get_component_handler(su, render_name);
if(!handler)
{
if(error_out)
*error_out = compiler_missing_request_handler_message(UnitCallMacroKind::component, render_name);
return(false);
}
compiler_execute_request_handler(
return(compiler_invoke_loaded_request_handler(
context,
su,
handler,
get_component_handler(su, render_name),
UnitCallMacroKind::component,
render_name,
false,
"uncaught exception during component render"
);
return(true);
"uncaught exception during component render",
error_out
));
}
void compiler_invoke(Request* context, String file_name)
@@ -1610,6 +1635,33 @@ void compiler_invoke(Request* context, String file_name)
}
}
void compiler_invoke_cli(Request* context, String file_name)
{
auto su = compiler_load_shared_unit(context, file_name, "", false);
if(!su)
return;
String error_message = "";
if(!compiler_invoke_loaded_request_handler(
context,
su,
su->on_cli,
UnitCallMacroKind::cli,
"",
compiler_is_request_entry_unit(context, su),
"uncaught exception during cli handler",
&error_message
))
{
if(!su->on_cli)
context->set_status(404, "CLI Entry Point Not Found");
else
context->set_status(500, "CLI Unit Error");
if(error_message != "")
print(error_message);
}
}
void compiler_invoke_websocket(Request* context, String file_name)
{
auto su = compiler_load_shared_unit(context, file_name, "", false);
@@ -1767,19 +1819,31 @@ DTree* unit_call(String file_name, String function_name, DTree* call_param)
}
else
{
UnitInvocationScope invoke_scope(context, su);
su->on_setup(context);
request_ref_handler handler = 0;
if(macro_target.kind == UnitCallMacroKind::once)
handler = su->on_once;
else if(macro_target.kind == UnitCallMacroKind::init)
handler = su->on_init;
else if(macro_target.kind == UnitCallMacroKind::cli)
handler = su->on_cli;
if(!handler)
print("Error: unit_call() ", compiler_missing_request_handler_message(macro_target.kind, macro_target.handler_name));
else if(macro_target.kind == UnitCallMacroKind::cli)
{
String prepare_error = "";
if(!compiler_prepare_request_handler(context, su, &prepare_error, true))
print("Error: unit_call() ", prepare_error);
else
compiler_execute_request_handler(context, su, handler, false, "uncaught exception during cli handler");
}
else
{
UnitInvocationScope invoke_scope(context, su);
su->on_setup(context);
handler(*context);
}
}
}
else
src/lib/compiler.h
+2
View File
@@ -5,6 +5,7 @@
#define ONCE(X) extern "C" void __uce_once(X)
#define INIT(X) extern "C" void __uce_init(X)
#define WS(X) extern "C" void __uce_websocket(X)
#define CLI(X) extern "C" void __uce_cli(X)
#define EXPORT extern "C"
String preprocess_shared_unit(Request* context, SharedUnit* su);
@@ -14,6 +15,7 @@ void compile_shared_unit(Request* context, SharedUnit* su);
SharedUnit* get_shared_unit(Request* context, String file_name, bool opt_so_optional = false);
void compiler_invoke(Request* context, String file_name);
void compiler_invoke_websocket(Request* context, String file_name);
void compiler_invoke_cli(Request* context, String file_name);
SharedUnit* compiler_load_shared_unit(Request* context, String file_name, String current_path = "", bool opt_so_optional = false);
String compiler_site_directory(Request* context);
StringList compiler_scan_site_units(Request* context);
src/lib/functionlib.cpp
+1121
View File
File diff suppressed because it is too large Load Diff
src/lib/functionlib.h
+4
View File
@@ -94,6 +94,10 @@ String html_escape(f64 a);
String json_encode(String s, char quote_char = '"');
String json_encode(DTree t, char quote_char = '"');
DTree json_decode(String s);
String xml_encode(DTree t, String root_name = "root");
DTree xml_decode(String s);
String yaml_encode(DTree t);
DTree yaml_decode(String s);
String var_dump(StringMap map, String prefix = "", String postfix = "\n");
String var_dump(StringList slist, String prefix = "", String postfix = "\n");
src/lib/sys.cpp
+150 -33
View File
@@ -613,89 +613,201 @@ pid_t spawn_subprocess(std::function<void()> exec_after_spawn)
}
}
String task_safe_key(String key)
{
key = trim(key);
if(key == "")
throw std::runtime_error("task key cannot be empty");
return(gen_sha1(key));
}
String task_file_prefix(String key)
{
return(path_join(context->server->config["BIN_DIRECTORY"], "task-" + task_safe_key(key)));
}
struct TaskStatus {
pid_t pid = 0;
String process_start_ticks = "";
};
TaskStatus task_status_parse(String status_file)
{
TaskStatus status;
auto lines = split(trim(status_file), "\n");
if(lines.size() > 0)
status.pid = (pid_t)int_val(trim(lines[0]));
if(lines.size() > 1)
status.process_start_ticks = trim(lines[1]);
return(status);
}
String task_process_start_ticks(pid_t pid)
{
if(pid <= 0)
return("");
String stat_file_name = "/proc/" + std::to_string(pid) + "/stat";
int fd = open(stat_file_name.c_str(), O_RDONLY);
if(fd == -1)
return("");
char buffer[4096];
ssize_t bytes_read = read(fd, buffer, sizeof(buffer) - 1);
close(fd);
if(bytes_read <= 0)
return("");
buffer[bytes_read] = '\0';
String stat = buffer;
size_t command_end = stat.rfind(") ");
if(command_end == String::npos)
return("");
String after_command = stat.substr(command_end + 2);
auto fields = split_space(after_command);
if(fields.size() <= 19)
return("");
return(fields[19]);
}
String task_status_content(pid_t pid)
{
return(std::to_string(pid) + "\n" + task_process_start_ticks(pid) + "\n");
}
bool task_status_is_alive(TaskStatus status)
{
if(status.pid <= 0)
return(false);
if(kill(status.pid, 0) != 0)
return(false);
if(status.process_start_ticks == "")
return(true);
return(task_process_start_ticks(status.pid) == status.process_start_ticks);
}
int task_kill(pid_t pid, int sig)
{
if(pid <= 0)
{
errno = EINVAL;
return(-1);
}
return(kill(pid, sig));
}
pid_t task_pid(String key)
{
String status_file_name = context->server->config["BIN_DIRECTORY"] + "/task-" + key;
String status_file_name = task_file_prefix(key);
String lock_file_name = status_file_name + ".lock";
int lock_fd = file_open_locked(lock_file_name, O_RDWR | O_CREAT, LOCK_EX, 0644);
int lock_fd = file_open_locked(lock_file_name, O_RDWR | O_CREAT, LOCK_EX, 0644, FILE_LOCK_WAIT_TIMEOUT_SECONDS, "task-pid:" + key);
if(lock_fd == -1)
{
fprintf(stderr, "task_pid(): could not lock task key '%s'\n", key.c_str());
return(0);
}
String status_file = file_get_contents(status_file_name);
pid_t p = 0;
if(status_file != "")
{
p = int_val(status_file);
if(task_kill(p, 0) == 0) // process is still running
TaskStatus status = task_status_parse(status_file);
if(task_status_is_alive(status))
{
file_close_locked(lock_fd);
return(p);
return(status.pid);
}
file_unlink(status_file_name);
}
file_close_locked(lock_fd);
return(p);
return(0);
}
pid_t task(String key, std::function<void()> exec_after_spawn, u64 timeout)
{
String status_file_name = context->server->config["BIN_DIRECTORY"] + "/task-" + key;
String status_file_name = task_file_prefix(key);
String lock_file_name = status_file_name + ".lock";
int lock_fd = file_open_locked(lock_file_name, O_RDWR | O_CREAT, LOCK_EX, 0644, FILE_LOCK_WAIT_TIMEOUT_SECONDS, "task:" + key);
if(lock_fd == -1)
{
fprintf(stderr, "task(): could not lock task key '%s'\n", key.c_str());
return(0);
}
String status_file = file_get_contents(status_file_name);
pid_t p;
pid_t p = 0;
if(status_file != "")
{
p = int_val(status_file);
if(task_kill(p, 0) == 0) // process is still running
TaskStatus status = task_status_parse(status_file);
if(task_status_is_alive(status))
{
printf("(P) worker process '%s' already running: PID %i\n", key.c_str(), p);
printf("(P) worker process '%s' already running: PID %i\n", key.c_str(), status.pid);
file_close_locked(lock_fd);
return(p);
return(status.pid);
}
//printf("(P) worker process '%s' had crashed: PID %i\n", key.c_str(), p);
file_unlink(status_file_name);
}
p = fork();
if(p < 0)
{
fprintf(stderr, "task(): fork failed for key '%s': %s\n", key.c_str(), strerror(errno));
file_close_locked(lock_fd);
return(0);
}
if(p == 0)
{
file_release_process_locks("task child startup");
file_close_locked(lock_fd);
my_pid = getpid();
prctl(PR_SET_PDEATHSIG, SIGHUP);
if(timeout > 0)
alarm(timeout);
close(context->resources.client_socket);
context->resources.client_socket = 0;
//printf("(C) child procress started, PID:%i\n", my_pid);
//prctl(PR_SET_PDEATHSIG, SIGHUP);
if(context->resources.client_socket > 0)
{
close(context->resources.client_socket);
context->resources.client_socket = 0;
}
exec_after_spawn();
int exit_lock_fd = file_open_locked(lock_file_name, O_RDWR | O_CREAT, LOCK_EX, 0644, FILE_LOCK_WAIT_TIMEOUT_SECONDS, "task-exit:" + key);
file_unlink(status_file_name);
file_close_locked(exit_lock_fd);
if(exit_lock_fd != -1)
{
file_unlink(status_file_name);
file_close_locked(exit_lock_fd);
}
else
{
fprintf(stderr, "task(): could not lock task key '%s' during child cleanup\n", key.c_str());
}
printf("(P) worker process '%s' terminated: PID %i\n", key.c_str(), my_pid);
exit(0);
}
else
if(!file_put_contents(status_file_name, task_status_content(p)))
{
file_put_contents(status_file_name, std::to_string(p));
fprintf(stderr, "task(): could not write status file for key '%s'; terminating child PID %i\n", key.c_str(), p);
kill(p, SIGTERM);
file_close_locked(lock_fd);
printf("(P) worker process '%s' spawned: PID %i\n", key.c_str(), p);
return(p);
return(0);
}
file_close_locked(lock_fd);
printf("(P) worker process '%s' spawned: PID %i\n", key.c_str(), p);
return(p);
}
#include <unistd.h>
pid_t task_repeat(String key, f64 interval, std::function<void()> exec_after_spawn, u64 timeout)
{
auto repeater_function = [&]() {
while (true)
if(interval <= 0)
throw std::runtime_error("task_repeat(): interval must be greater than zero");
auto repeater_function = [key, interval, exec_after_spawn, timeout]() {
f64 started_at = time_precise();
while (timeout == 0 || time_precise() - started_at < (f64)timeout)
{
exec_after_spawn();
f64 elapsed = time_precise() - started_at;
if(timeout > 0 && elapsed >= (f64)timeout)
break;
f64 sleep_seconds = interval;
if(timeout > 0 && elapsed + sleep_seconds > (f64)timeout)
sleep_seconds = (f64)timeout - elapsed;
printf("(P) worker process '%s' sleeping\n", key.c_str());
usleep((s64)(interval*1000000));
if(sleep_seconds > 0)
usleep((useconds_t)(sleep_seconds * 1000000.0));
}
};
return(task(key, repeater_function, timeout));
@@ -703,17 +815,21 @@ pid_t task_repeat(String key, f64 interval, std::function<void()> exec_after_spa
void on_child_exit(int sig)
{
pid_t pid;
int status;
if ((pid = waitpid(-1, &status, WNOHANG)) != -1)
{
(void)sig;
pid_t pid;
int status;
while((pid = waitpid(-1, &status, WNOHANG)) > 0)
{
if(workers.count(pid) > 0)
{
workers.erase(pid);
printf("(P) child terminated (PID:%i)\n", pid);
//spawn_subprocess();
}
}
else
{
printf("(P) task child reaped (PID:%i)\n", pid);
}
}
}
StringList ls(String dir)
@@ -731,6 +847,7 @@ StringMap make_server_settings()
cfg["LIT_ESC"] = "3d5b5_1";
cfg["CONTENT_TYPE"] = "text/html; charset=utf-8";
cfg["FCGI_SOCKET_PATH"] = "/run/uce.sock";
cfg["CLI_SOCKET_PATH"] = "/run/uce/cli.sock";
cfg["TMP_UPLOAD_PATH"] = "/tmp/uce/uploads";
cfg["SESSION_PATH"] = "/tmp/uce/sessions";
cfg["COMPILER_SYS_PATH"] = ".";
src/lib/types.h
+2
View File
@@ -85,6 +85,7 @@ struct SharedUnit {
request_ref_handler on_render = 0;
request_ref_handler on_component = 0;
request_ref_handler on_websocket = 0;
request_ref_handler on_cli = 0;
request_ref_handler on_once = 0;
request_ref_handler on_init = 0;
@@ -210,6 +211,7 @@ struct Request {
u64 client_socket = 0;
u64 server_socket = 0;
bool is_websocket = false;
bool is_cli = false;
String websocket_connection_id = "";
String websocket_scope = "";
DTree* websocket_connection_state = 0;
src/lib/uce_lib.cpp
+2
View File
@@ -6,7 +6,9 @@
#include "hash.cpp"
#include "sys.cpp"
#include "uri.cpp"
#include "cli.cpp"
#include "compiler-parser.cpp"
#include "compiler.cpp"
#include "markdown.cpp"
#include "zip.cpp"
#include "mysql-connector.cpp"
src/lib/uce_lib.h
+2
View File
@@ -6,6 +6,8 @@
#include "functionlib.h"
#include "sys.h"
#include "uri.h"
#include "cli.h"
#include "compiler.h"
#include "markdown.h"
#include "zip.h"
#include "mysql-connector.h"
src/lib/uri.cpp
+10
View File
@@ -707,6 +707,11 @@ bool WSFrame::parse(const String& buffer, String& error)
if(buffer.length() < 4)
return(false);
payload_length = ((u64)raw[2] << 8) | (u64)raw[3];
if(payload_length < 126)
{
error = "non-minimal websocket frame length";
return(false);
}
header_length = 4;
}
else if(payload_length == 127)
@@ -721,6 +726,11 @@ bool WSFrame::parse(const String& buffer, String& error)
payload_length = 0;
for(u32 i = 0; i < 8; i++)
payload_length = (payload_length << 8) | (u64)raw[2 + i];
if(payload_length <= 0xFFFF)
{
error = "non-minimal websocket frame length";
return(false);
}
header_length = 10;
}
src/lib/zip.cpp
+350
View File
@@ -0,0 +1,350 @@
#include "zip.h"
#include <cstring>
extern "C" {
#include "../../vendor/miniz/miniz.c"
#include "../../vendor/miniz/miniz_tdef.c"
#include "../../vendor/miniz/miniz_tinfl.c"
#include "../../vendor/miniz/miniz_zip.c"
}
String zip_error(String api, String detail)
{
return(api + "(): " + detail);
}
bool zip_entry_name_safe(String name)
{
if(name == "")
return(false);
if(name[0] == '/' || name[0] == '\\')
return(false);
if(name.find(":") != String::npos)
return(false);
auto parts = split(replace(name, "\\", "/"), "/");
for(auto part : parts)
{
if(part == "..")
return(false);
}
return(true);
}
String zip_normalize_entry_name(String name)
{
name = replace(name, "\\", "/");
while(name.find("//") != String::npos)
name = replace(name, "//", "/");
return(name);
}
bool zip_ensure_parent_directory(String path)
{
String parent = dirname(path);
if(parent == "" || parent == "." || parent == "/")
return(true);
if(file_exists(parent))
return(true);
return(mkdir(parent));
}
void zip_add_file_info(DTree& files, mz_zip_archive* archive, mz_uint index)
{
mz_zip_archive_file_stat stat;
std::memset(&stat, 0, sizeof(stat));
if(!mz_zip_reader_file_stat(archive, index, &stat))
throw std::runtime_error(zip_error("zip_list", "could not read file metadata at index " + std::to_string((u64)index)));
DTree item;
item["name"] = String(stat.m_filename);
item["index"] = (f64)index;
item["size"] = (f64)stat.m_uncomp_size;
item["compressed_size"] = (f64)stat.m_comp_size;
item["is_directory"].set_bool(mz_zip_reader_is_file_a_directory(archive, index));
item["method"] = (f64)stat.m_method;
files.push(item);
}
DTree zip_list(String zip_file_name)
{
mz_zip_archive archive;
std::memset(&archive, 0, sizeof(archive));
if(!mz_zip_reader_init_file(&archive, zip_file_name.c_str(), 0))
throw std::runtime_error(zip_error("zip_list", "could not open " + zip_file_name));
DTree result;
try
{
mz_uint count = mz_zip_reader_get_num_files(&archive);
result["file"] = zip_file_name;
result["count"] = (f64)count;
result["entries"].set_array();
for(mz_uint i = 0; i < count; i++)
zip_add_file_info(result["entries"], &archive, i);
mz_zip_reader_end(&archive);
return(result);
}
catch(...)
{
mz_zip_reader_end(&archive);
throw;
}
}
String zip_read(String zip_file_name, String entry_name)
{
entry_name = zip_normalize_entry_name(entry_name);
if(!zip_entry_name_safe(entry_name))
throw std::runtime_error(zip_error("zip_read", "unsafe or empty entry name '" + entry_name + "'"));
mz_zip_archive archive;
std::memset(&archive, 0, sizeof(archive));
if(!mz_zip_reader_init_file(&archive, zip_file_name.c_str(), 0))
throw std::runtime_error(zip_error("zip_read", "could not open " + zip_file_name));
size_t size = 0;
void* data = mz_zip_reader_extract_file_to_heap(&archive, entry_name.c_str(), &size, 0);
if(!data)
{
mz_zip_reader_end(&archive);
throw std::runtime_error(zip_error("zip_read", "entry not found or not readable: " + entry_name));
}
String result((char*)data, size);
mz_free(data);
mz_zip_reader_end(&archive);
return(result);
}
String zip_entry_content(DTree item)
{
DTree* content = item.key("content");
if(content)
return(content->to_string());
DTree* file_name = item.key("file");
if(file_name)
return(file_get_contents(file_name->to_string()));
return(item.to_string());
}
String zip_entry_name(String key, DTree item)
{
DTree* name = item.key("name");
if(name)
return(name->to_string());
return(key);
}
void zip_add_entry(mz_zip_archive* archive, String name, String content)
{
name = zip_normalize_entry_name(name);
if(!zip_entry_name_safe(name))
throw std::runtime_error(zip_error("zip_create", "unsafe or empty entry name '" + name + "'"));
if(!mz_zip_writer_add_mem(archive, name.c_str(), content.data(), content.size(), MZ_BEST_COMPRESSION))
throw std::runtime_error(zip_error("zip_create", "could not add entry " + name));
}
bool zip_create(String zip_file_name, DTree entries)
{
if(!zip_ensure_parent_directory(zip_file_name))
throw std::runtime_error(zip_error("zip_create", "could not create parent directory for " + zip_file_name));
mz_zip_archive archive;
std::memset(&archive, 0, sizeof(archive));
if(!mz_zip_writer_init_file(&archive, zip_file_name.c_str(), 0))
throw std::runtime_error(zip_error("zip_create", "could not open " + zip_file_name + " for writing"));
try
{
entries.each([&](DTree item, String key)
{
String name = zip_entry_name(key, item);
String content = zip_entry_content(item);
zip_add_entry(&archive, name, content);
});
if(!mz_zip_writer_finalize_archive(&archive))
throw std::runtime_error(zip_error("zip_create", "could not finalize " + zip_file_name));
mz_zip_writer_end(&archive);
return(true);
}
catch(...)
{
mz_zip_writer_end(&archive);
throw;
}
}
void gz_append_u32_le(String& out, u32 value)
{
out.push_back((char)(value & 0xff));
out.push_back((char)((value >> 8) & 0xff));
out.push_back((char)((value >> 16) & 0xff));
out.push_back((char)((value >> 24) & 0xff));
}
u32 gz_read_u32_le(String src, size_t offset)
{
return(
((u32)(u8)src[offset]) |
((u32)(u8)src[offset + 1] << 8) |
((u32)(u8)src[offset + 2] << 16) |
((u32)(u8)src[offset + 3] << 24)
);
}
void gz_skip_zero_terminated(String compressed, size_t& offset)
{
while(offset < compressed.size() && compressed[offset] != '\0')
offset++;
if(offset >= compressed.size())
throw std::runtime_error("gz_uncompress(): truncated gzip header");
offset++;
}
size_t gz_deflate_offset(String compressed)
{
if(compressed.size() < 18)
throw std::runtime_error("gz_uncompress(): compressed data is too short");
if((u8)compressed[0] != 0x1f || (u8)compressed[1] != 0x8b)
throw std::runtime_error("gz_uncompress(): missing gzip header");
if((u8)compressed[2] != 8)
throw std::runtime_error("gz_uncompress(): unsupported gzip compression method");
u8 flags = (u8)compressed[3];
if(flags & 0xe0)
throw std::runtime_error("gz_uncompress(): gzip header uses reserved flags");
size_t offset = 10;
if(flags & 0x04)
{
if(offset + 2 > compressed.size())
throw std::runtime_error("gz_uncompress(): truncated gzip extra header");
u16 extra_len = ((u16)(u8)compressed[offset]) | ((u16)(u8)compressed[offset + 1] << 8);
offset += 2 + extra_len;
if(offset > compressed.size())
throw std::runtime_error("gz_uncompress(): truncated gzip extra data");
}
if(flags & 0x08)
gz_skip_zero_terminated(compressed, offset);
if(flags & 0x10)
gz_skip_zero_terminated(compressed, offset);
if(flags & 0x02)
{
offset += 2;
if(offset > compressed.size())
throw std::runtime_error("gz_uncompress(): truncated gzip header crc");
}
if(offset + 8 > compressed.size())
throw std::runtime_error("gz_uncompress(): missing gzip footer");
return(offset);
}
String gz_compress(String src)
{
mz_stream stream;
std::memset(&stream, 0, sizeof(stream));
int status = mz_deflateInit2(&stream, MZ_DEFAULT_COMPRESSION, MZ_DEFLATED, -MZ_DEFAULT_WINDOW_BITS, 9, MZ_DEFAULT_STRATEGY);
if(status != MZ_OK)
throw std::runtime_error("gz_compress(): could not initialize compressor");
mz_ulong bound = mz_deflateBound(&stream, src.size());
String deflated;
deflated.resize(bound);
stream.next_in = (const unsigned char*)src.data();
stream.avail_in = src.size();
stream.next_out = (unsigned char*)deflated.data();
stream.avail_out = deflated.size();
status = mz_deflate(&stream, MZ_FINISH);
if(status != MZ_STREAM_END)
{
mz_deflateEnd(&stream);
throw std::runtime_error("gz_compress(): compression failed");
}
deflated.resize(stream.total_out);
mz_deflateEnd(&stream);
String result;
result.reserve(10 + deflated.size() + 8);
result.push_back((char)0x1f);
result.push_back((char)0x8b);
result.push_back((char)0x08);
result.push_back((char)0x00);
gz_append_u32_le(result, 0);
result.push_back((char)0x00);
result.push_back((char)0xff);
result.append(deflated);
gz_append_u32_le(result, (u32)mz_crc32(MZ_CRC32_INIT, (const unsigned char*)src.data(), src.size()));
gz_append_u32_le(result, (u32)src.size());
return(result);
}
String gz_uncompress(String compressed)
{
size_t deflate_offset = gz_deflate_offset(compressed);
size_t footer_offset = compressed.size() - 8;
size_t deflate_size = footer_offset - deflate_offset;
size_t out_len = 0;
void* out = tinfl_decompress_mem_to_heap(compressed.data() + deflate_offset, deflate_size, &out_len, 0);
if(!out)
throw std::runtime_error("gz_uncompress(): decompression failed");
String result((char*)out, out_len);
mz_free(out);
u32 expected_crc = gz_read_u32_le(compressed, footer_offset);
u32 expected_size = gz_read_u32_le(compressed, footer_offset + 4);
u32 actual_crc = (u32)mz_crc32(MZ_CRC32_INIT, (const unsigned char*)result.data(), result.size());
if(actual_crc != expected_crc)
throw std::runtime_error("gz_uncompress(): crc check failed");
if((u32)result.size() != expected_size)
throw std::runtime_error("gz_uncompress(): size check failed");
return(result);
}
bool zip_extract(String zip_file_name, String destination_directory)
{
if(destination_directory == "")
throw std::runtime_error(zip_error("zip_extract", "destination directory is empty"));
if(!file_exists(destination_directory) && !mkdir(destination_directory))
throw std::runtime_error(zip_error("zip_extract", "could not create destination directory " + destination_directory));
mz_zip_archive archive;
std::memset(&archive, 0, sizeof(archive));
if(!mz_zip_reader_init_file(&archive, zip_file_name.c_str(), 0))
throw std::runtime_error(zip_error("zip_extract", "could not open " + zip_file_name));
try
{
mz_uint count = mz_zip_reader_get_num_files(&archive);
for(mz_uint i = 0; i < count; i++)
{
mz_zip_archive_file_stat stat;
std::memset(&stat, 0, sizeof(stat));
if(!mz_zip_reader_file_stat(&archive, i, &stat))
throw std::runtime_error(zip_error("zip_extract", "could not read file metadata at index " + std::to_string((u64)i)));
String name = zip_normalize_entry_name(stat.m_filename);
if(!zip_entry_name_safe(name))
throw std::runtime_error(zip_error("zip_extract", "refusing unsafe entry name '" + name + "'"));
String target = path_join(destination_directory, name);
if(mz_zip_reader_is_file_a_directory(&archive, i))
{
if(!file_exists(target) && !mkdir(target))
throw std::runtime_error(zip_error("zip_extract", "could not create directory " + target));
continue;
}
if(!zip_ensure_parent_directory(target))
throw std::runtime_error(zip_error("zip_extract", "could not create parent directory for " + target));
if(!mz_zip_reader_extract_to_file(&archive, i, target.c_str(), 0))
throw std::runtime_error(zip_error("zip_extract", "could not extract " + name));
}
mz_zip_reader_end(&archive);
return(true);
}
catch(...)
{
mz_zip_reader_end(&archive);
throw;
}
}
src/lib/zip.h
+8
View File
@@ -0,0 +1,8 @@
#pragma once
DTree zip_list(String zip_file_name);
String zip_read(String zip_file_name, String entry_name);
bool zip_create(String zip_file_name, DTree entries);
bool zip_extract(String zip_file_name, String destination_directory);
String gz_compress(String src);
String gz_uncompress(String compressed);
src/linux_fastcgi.cpp
+188 -53
View File
@@ -26,6 +26,7 @@ static DTree websocket_exec_inflight_job;
static String websocket_exec_write_buffer = "";
void close_inherited_server_sockets();
u64 request_seed_from_time(f64 time_value);
Request* set_active_request(Request& request)
{
@@ -315,6 +316,25 @@ void websocket_exec_append_command(DTree command)
context->resources.websocket_dispatch_commands.push(command);
}
DTree websocket_exec_make_message_command(String action, String message, bool binary)
{
DTree command;
command["action"] = action;
command["binary"].set_bool(binary);
command["message_b64"] = websocket_ipc_base64_encode(message);
return(command);
}
DTree websocket_exec_make_close_command(String connection_id, u16 status_code = 1000, String reason = "")
{
DTree command;
command["action"] = "close";
command["connection_id"] = connection_id;
command["status_code"] = (f64)status_code;
command["reason"] = reason;
return(command);
}
void websocket_exec_apply_command(DTree command)
{
String action = command["action"].to_string();
@@ -470,7 +490,7 @@ Request websocket_exec_build_event_request(DTree job, String message)
event_request.stats.time_init = time_precise();
event_request.stats.time_start = event_request.stats.time_init;
event_request.random_index = 0;
event_request.random_seed = gen_noise64(*reinterpret_cast<u64*>(&event_request.stats.time_start));
event_request.random_seed = request_seed_from_time(event_request.stats.time_start);
event_request.response_code = "WEBSOCKET";
event_request.header["Content-Type"] = server_state.config["CONTENT_TYPE"];
event_request.in = message;
@@ -747,28 +767,23 @@ u64 ws_connection_count(String scope)
bool ws_send(String message, bool binary, String scope)
{
String normalized_scope = normalize_ws_scope(scope);
if(context && context->resources.websocket_dispatch_capture)
{
DTree command;
command["action"] = "broadcast";
command["scope"] = normalize_ws_scope(scope);
command["binary"].set_bool(binary);
command["message_b64"] = websocket_ipc_base64_encode(message);
DTree command = websocket_exec_make_message_command("broadcast", message, binary);
command["scope"] = normalized_scope;
websocket_exec_append_command(command);
return(true);
}
return(server.websocket_broadcast(normalize_ws_scope(scope), message, binary) > 0);
return(server.websocket_broadcast(normalized_scope, message, binary) > 0);
}
bool ws_send_to(String connection_id, String message, bool binary)
{
if(context && context->resources.websocket_dispatch_capture)
{
DTree command;
command["action"] = "send_to";
DTree command = websocket_exec_make_message_command("send_to", message, binary);
command["connection_id"] = connection_id;
command["binary"].set_bool(binary);
command["message_b64"] = websocket_ipc_base64_encode(message);
websocket_exec_append_command(command);
return(true);
}
@@ -783,17 +798,156 @@ bool ws_close(String connection_id)
return(false);
if(context && context->resources.websocket_dispatch_capture)
{
DTree command;
command["action"] = "close";
command["connection_id"] = connection_id;
command["status_code"] = (f64)1000;
command["reason"] = "";
websocket_exec_append_command(command);
websocket_exec_append_command(websocket_exec_make_close_command(connection_id));
return(true);
}
return(server.websocket_close(connection_id));
}
bool cli_path_is_safe(String command)
{
for(auto& segment : split(command, "/"))
{
if(segment == "..")
return(false);
}
return(true);
}
String cli_resolve_unit_path(Request& request, String command, String& document_root)
{
document_root = first(request.params["DOCUMENT_ROOT"], cwd_get());
if(document_root.length() > 1 && document_root[document_root.length()-1] == '/')
document_root.resize(document_root.length()-1);
String script_filename = document_root + command;
if(file_exists(script_filename))
return(script_filename);
String site_filename = document_root + "/site" + command;
if(file_exists(site_filename))
{
document_root = document_root + "/site";
return(site_filename);
}
return("");
}
u64 request_seed_from_time(f64 time_value)
{
u64 bits = 0;
static_assert(sizeof(bits) == sizeof(time_value));
memcpy(&bits, &time_value, sizeof(bits));
return(gen_noise64(bits));
}
void prepare_request_body_maps(Request& request)
{
request.get = parse_query(request.params["QUERY_STRING"]);
if(request.params["HTTP_COOKIE"].length() > 0)
request.cookies = parse_cookies(request.params["HTTP_COOKIE"]);
String ct_info = request.params["CONTENT_TYPE"];
String ct_type = nibble(ct_info, ";");
if(request.params["REQUEST_METHOD"] == "POST")
{
if(ct_type == "multipart/form-data")
{
nibble("boundary=", ct_info);
request.post = parse_multipart(request.in, String("--")+ct_info, request.uploaded_files);
}
else
{
request.post = parse_query(request.in);
}
}
}
int handle_cli_complete(FastCGIRequest& request)
{
Request* previous_context = set_active_request(request);
server_state.request_count += 1;
request.server = &server_state;
request.resources.is_cli = true;
request.params["UCE_CLI"] = "1";
request.stats.time_start = time_precise();
request.header["Content-Type"] = "text/plain; charset=utf-8";
request.random_index = 0;
request.random_seed = request_seed_from_time(request.stats.time_start);
request.ob_start();
prepare_request_body_maps(request);
String method = trim(request.params["REQUEST_METHOD"]);
String command = trim(first(request.params["DOCUMENT_URI"], request.params["REQUEST_URI"]));
try
{
if(method != "GET" && method != "POST")
{
request.set_status(405, "Method Not Allowed");
request.header["Allow"] = "GET, POST";
print("UCE CLI socket accepts GET and POST commands only\n");
}
else if(command == "/" || command == "/help")
{
print("UCE CLI command socket\n\nAvailable test hooks:\n GET /ping\n GET /status\n");
}
else if(command == "/ping" || command == "/test")
{
print("uce-cli: ok\n");
}
else if(command == "/status")
{
print("pid=", std::to_string(getpid()), "\nclients=", std::to_string(server.client_sockets.size()), "\n");
}
else if(command.length() >= 4 && command.substr(command.length() - 4) == ".uce")
{
if(!cli_path_is_safe(command))
{
request.set_status(400, "Bad Request");
print("invalid UCE CLI unit path\n");
}
else
{
String document_root;
String script_filename = cli_resolve_unit_path(request, command, document_root);
if(script_filename == "")
{
request.set_status(404, "Not Found");
print("UCE CLI unit not found: ", command, "\n");
}
else
{
request.params["DOCUMENT_ROOT"] = document_root;
request.params["SCRIPT_FILENAME"] = script_filename;
request.props = DTree();
compiler_invoke_cli(&request, script_filename);
}
}
}
else
{
request.set_status(404, "Not Found");
print("unknown UCE CLI command: ", command, "\n");
}
}
catch(const std::exception& e)
{
render_request_failure(request, "uncaught exception during CLI request", e.what(), capture_backtrace_string(32, 1), 500);
}
catch(...)
{
render_request_failure(request, "unknown uncaught exception during CLI request", "", capture_backtrace_string(32, 1), 500);
}
for(auto &f : request.uploaded_files)
file_unlink(f.tmp_name);
cleanup_mysql_connections();
restore_active_request(previous_context);
return(request.flags.status);
}
int handle_request(FastCGIRequest& request) {
// This is always the first event to occur. It occurs when the
// server receives all parameters. There may be more data coming on the
@@ -805,20 +959,9 @@ int handle_request(FastCGIRequest& request) {
}
int handle_data(FastCGIRequest& request) {
// This event occurs when data is received on the standard input stream.
// A simple String is used to hold the input stream, so it is the
// responsibility of the application to remember which data it has
// processed. The application may modify it; new data will be appended
// to it by the server. The same goes for the output and error streams:
// the application should append data to them; the server will remove
// all sent data from them.
return 0; // still OK
std::transform(request.in.begin(), request.in.end(),
std::back_inserter(request.err),
std::bind1st(std::plus<char>(), 1));
request.in.clear(); // don't process it again
return 0; // still OK
// Request bodies are accumulated by the FastCGI transport and parsed once
// the input stream is closed in handle_complete().
return 0;
}
int handle_complete(FastCGIRequest& request) {
@@ -833,10 +976,10 @@ int handle_complete(FastCGIRequest& request) {
request.stats.time_start = time_precise();
//request.stats.mem_alloc = 0;
//request.stats.mem_high = 0;
request.header["Content-Type"] = context->server->config["CONTENT_TYPE"];
request.header["Content-Type"] = (request.resources.is_cli ? "text/plain; charset=utf-8" : context->server->config["CONTENT_TYPE"]);
request.get = parse_query(request.params["QUERY_STRING"]);
request.random_index = 0;
request.random_seed = gen_noise64(*reinterpret_cast<u64*>(&request.stats.time_start));
request.random_seed = request_seed_from_time(request.stats.time_start);
request.ob_start();
request_fault_request = &request;
request_fault_active = 1;
@@ -858,27 +1001,12 @@ int handle_complete(FastCGIRequest& request) {
{
try
{
if(request.params["HTTP_COOKIE"].length() > 0)
request.cookies = parse_cookies(request.params["HTTP_COOKIE"]);
String ct_info = request.params["CONTENT_TYPE"];
String ct_type = nibble(";", ct_info);
if(request.params["REQUEST_METHOD"] == "POST")
{
if(ct_type == "multipart/form-data")
{
nibble("boundary=", ct_info);
request.post = parse_multipart(request.in, String("--")+ct_info, request.uploaded_files);
}
else
{
request.post = parse_query(request.in);
}
}
prepare_request_body_maps(request);
request.props = DTree();
compiler_invoke(&request, request.params["SCRIPT_FILENAME"]);
if(request.resources.is_cli)
compiler_invoke_cli(&request, request.params["SCRIPT_FILENAME"]);
else
compiler_invoke(&request, request.params["SCRIPT_FILENAME"]);
}
catch(const std::exception& e)
{
@@ -1142,6 +1270,7 @@ void listen_for_connections()
server.on_request = &handle_request;
server.on_data = &handle_data;
server.on_complete = &handle_complete;
server.on_cli_complete = &handle_cli_complete;
server.on_websocket_message = &handle_websocket_message;
if(worker_accepts_http)
ensure_websocket_executor();
@@ -1176,6 +1305,12 @@ void init_base_process()
chmod(server_state.config["FCGI_SOCKET_PATH"].c_str(), S_IRWXU | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
}
if(server_state.config["CLI_SOCKET_PATH"] != "")
{
server.listen_cli(server_state.config["CLI_SOCKET_PATH"]);
chmod(server_state.config["CLI_SOCKET_PATH"].c_str(), S_IRWXU | S_IRGRP | S_IWGRP);
}
if(server_state.config["HTTP_PORT"] != "")
server.listen_http(int_val(server_state.config["HTTP_PORT"]));