summaryrefslogtreecommitdiff
path: root/plugins/websocket/websocket.c
diff options
context:
space:
mode:
Diffstat (limited to 'plugins/websocket/websocket.c')
-rw-r--r--plugins/websocket/websocket.c723
1 files changed, 723 insertions, 0 deletions
diff --git a/plugins/websocket/websocket.c b/plugins/websocket/websocket.c
new file mode 100644
index 0000000..4ff2230
--- /dev/null
+++ b/plugins/websocket/websocket.c
@@ -0,0 +1,723 @@
+/*
+ * websocket — RFC6455 forward-proxy transport. INI: [transport.websocket].
+ *
+ * Raw byte stream to the session core (no HyBBX auth at the wire layer).
+ * Intended behind a TLS reverse proxy; see docs/WEBSOCKET.md.
+ */
+#include "hybbx/plugin.h"
+#include "hybbx/service.h"
+#include "hybbx/session.h"
+#include "hybbx/socket.h"
+#include "hybbx/security_ban.h"
+#include "hybbx/websocket.h"
+#include "hybbx/util.h"
+#include "ws_proto.h"
+#include "ws_tls.h"
+#include "hybbx/log.h"
+
+#include <arpa/inet.h>
+#include <errno.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <poll.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+#include <unistd.h>
+
+#define WS_CLIENT_POLL_MS 5000
+#define WS_CLIENT_PING_IDLE_SEC 20
+
+typedef struct ws_client {
+ hybbx_ws_connection_t ws;
+ hybbx_session_t *hbx_session;
+ hybbx_result_t last_rc;
+ int slot_held;
+ time_t last_io_at;
+ uint8_t tx_buf[HYBBX_WS_FRAME_PAYLOAD_MAX];
+ size_t tx_len;
+} ws_client_t;
+
+typedef struct ws_client_ctx {
+ ws_client_t *client;
+} ws_client_ctx_t;
+
+extern const hybbx_transport_plugin_t hybbx_plugin_websocket;
+
+static hybbx_service_t *g_service;
+static hybbx_websocket_config_t g_config;
+static pthread_t g_accept_thread;
+static int g_listen_v4 = -1;
+static int g_listen_v6 = -1;
+static volatile int g_ws_running = 0;
+static unsigned g_active_clients;
+static pthread_mutex_t g_client_lock = PTHREAD_MUTEX_INITIALIZER;
+
+static hybbx_result_t ws_plugin_stop(void);
+
+static int set_socket_options(int fd, int family)
+{
+ int on = 1;
+
+ if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)) != 0) {
+ return -1;
+ }
+
+ if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &on, sizeof(on)) != 0) {
+ return -1;
+ }
+
+ hybbx_socket_nosigpipe(fd);
+
+#ifdef IPV6_V6ONLY
+ if (family == AF_INET6) {
+ if (setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &on, sizeof(on)) != 0) {
+ return -1;
+ }
+ }
+#else
+ (void)family;
+#endif
+
+ return 0;
+}
+
+static int create_listen_socket(int family, const char *bind_addr, unsigned port)
+{
+ int fd;
+ int rc;
+
+ fd = socket(family, SOCK_STREAM, 0);
+ if (fd < 0) {
+ return -1;
+ }
+
+ if (set_socket_options(fd, family) != 0) {
+ close(fd);
+ return -1;
+ }
+
+ if (family == AF_INET6) {
+ struct sockaddr_in6 addr6;
+
+ memset(&addr6, 0, sizeof(addr6));
+ addr6.sin6_family = AF_INET6;
+ addr6.sin6_port = htons((uint16_t)port);
+
+ if (inet_pton(AF_INET6, bind_addr, &addr6.sin6_addr) != 1) {
+ close(fd);
+ return -1;
+ }
+
+ rc = bind(fd, (struct sockaddr *)&addr6, sizeof(addr6));
+ } else {
+ struct sockaddr_in addr4;
+
+ memset(&addr4, 0, sizeof(addr4));
+ addr4.sin_family = AF_INET;
+ addr4.sin_port = htons((uint16_t)port);
+
+ if (inet_pton(AF_INET, bind_addr, &addr4.sin_addr) != 1) {
+ close(fd);
+ return -1;
+ }
+
+ rc = bind(fd, (struct sockaddr *)&addr4, sizeof(addr4));
+ }
+
+ if (rc != 0) {
+ close(fd);
+ return -1;
+ }
+
+ if (listen(fd, 16) != 0) {
+ close(fd);
+ return -1;
+ }
+
+ return fd;
+}
+
+static void log_bind_failure(const char *bind_addr, unsigned port)
+{
+ int err = errno;
+
+ hybbx_socket_log_bind_failure("websocket", bind_addr, port);
+ if (err == EACCES && port < 1024u) {
+ hybbx_log_warn("[websocket] port %u is privileged — run as root, grant "
+ "bind permission, or set port >= 1024 in hybbx.ini",
+ port);
+ }
+}
+
+static size_t utf8_char_span(const uint8_t *buf, size_t len, size_t pos)
+{
+ uint8_t b;
+ size_t n;
+ size_t i;
+
+ if (pos >= len) {
+ return 0;
+ }
+
+ b = buf[pos];
+ if (b < 0x80u) {
+ return 1;
+ }
+ if ((b & 0xE0u) == 0xC0u) {
+ n = 2;
+ } else if ((b & 0xF0u) == 0xE0u) {
+ n = 3;
+ } else if ((b & 0xF8u) == 0xF0u) {
+ n = 4;
+ } else {
+ return 1;
+ }
+
+ if (pos + n > len) {
+ return 0;
+ }
+
+ for (i = 1; i < n; i++) {
+ if ((buf[pos + i] & 0xC0u) != 0x80u) {
+ return 1;
+ }
+ }
+
+ return n;
+}
+
+static size_t utf8_complete_prefix(const uint8_t *buf, size_t len)
+{
+ size_t pos = 0;
+
+ while (pos < len) {
+ size_t span = utf8_char_span(buf, len, pos);
+
+ if (span == 0) {
+ break;
+ }
+ pos += span;
+ }
+
+ return pos;
+}
+
+static hybbx_result_t ws_tx_flush(ws_client_t *client, int force_tail)
+{
+ size_t n;
+ hybbx_result_t rc;
+
+ if (client == NULL || client->tx_len == 0) {
+ return HYBBX_OK;
+ }
+
+ n = utf8_complete_prefix(client->tx_buf, client->tx_len);
+ if (n == 0 && force_tail) {
+ client->tx_buf[0] = '?';
+ n = 1;
+ }
+ if (n == 0) {
+ return HYBBX_OK;
+ }
+
+ rc = hybbx_ws_write_text(&client->ws, (const char *)client->tx_buf, n);
+ if (rc != HYBBX_OK) {
+ return rc;
+ }
+ client->last_io_at = time(NULL);
+
+ if (n < client->tx_len) {
+ memmove(client->tx_buf, client->tx_buf + n, client->tx_len - n);
+ }
+ client->tx_len -= n;
+
+ return HYBBX_OK;
+}
+
+static hybbx_result_t ws_plugin_write(hybbx_session_t *session,
+ const char *data, size_t len)
+{
+ ws_client_t *client;
+ size_t i;
+ hybbx_result_t rc;
+
+ if (session == NULL || data == NULL) {
+ return HYBBX_ERR_INVALID;
+ }
+
+ client = (ws_client_t *)session->transport_data;
+ if (client == NULL || !client->ws.established) {
+ return HYBBX_ERR_INVALID;
+ }
+
+ for (i = 0; i < len; i++) {
+ char ch = data[i];
+
+ if (ch == '\n' && (i == 0 || data[i - 1] != '\r')) {
+ if (client->tx_len + 2 > sizeof(client->tx_buf)) {
+ rc = ws_tx_flush(client, 1);
+ if (rc != HYBBX_OK) {
+ return rc;
+ }
+ }
+ client->tx_buf[client->tx_len++] = '\r';
+ }
+
+ if (client->tx_len >= sizeof(client->tx_buf)) {
+ rc = ws_tx_flush(client, 1);
+ if (rc != HYBBX_OK) {
+ return rc;
+ }
+ }
+
+ client->tx_buf[client->tx_len++] = (uint8_t)ch;
+
+ rc = ws_tx_flush(client, 0);
+ if (rc != HYBBX_OK) {
+ return rc;
+ }
+ }
+
+ return HYBBX_OK;
+}
+
+static void ws_send_busy(ws_client_t *client)
+{
+ static const char msg[] = "All nodes in use. Try later.\r\n";
+
+ if (client == NULL) {
+ return;
+ }
+
+ (void)hybbx_ws_write_text(&client->ws, msg, sizeof(msg) - 1u);
+}
+
+static void ws_send_limit(ws_client_t *client)
+{
+ static const char msg[] =
+ "All WebSocket connections in use. Try later.\r\n";
+
+ if (client == NULL) {
+ return;
+ }
+
+ (void)hybbx_ws_write_text(&client->ws, msg, sizeof(msg) - 1u);
+}
+
+static int ws_client_acquire_slot(ws_client_t *client)
+{
+ if (client == NULL) {
+ return 0;
+ }
+
+ pthread_mutex_lock(&g_client_lock);
+ if (g_active_clients >= g_config.max_connections) {
+ pthread_mutex_unlock(&g_client_lock);
+ return 0;
+ }
+
+ g_active_clients++;
+ client->slot_held = 1;
+ pthread_mutex_unlock(&g_client_lock);
+ return 1;
+}
+
+static void ws_client_release_slot(ws_client_t *client)
+{
+ if (client == NULL || !client->slot_held) {
+ return;
+ }
+
+ pthread_mutex_lock(&g_client_lock);
+ if (g_active_clients > 0) {
+ g_active_clients--;
+ }
+ client->slot_held = 0;
+ pthread_mutex_unlock(&g_client_lock);
+}
+
+static void ws_on_user_data(void *ctx, const uint8_t *data, size_t len)
+{
+ ws_client_ctx_t *wctx = (ws_client_ctx_t *)ctx;
+ hybbx_result_t rc;
+
+ if (wctx == NULL || wctx->client == NULL ||
+ wctx->client->hbx_session == NULL || data == NULL || len == 0) {
+ return;
+ }
+
+ rc = hybbx_session_handle_input(wctx->client->hbx_session, data, len);
+ if (rc != HYBBX_OK) {
+ wctx->client->last_rc = rc;
+ }
+}
+
+static void *ws_client_thread(void *arg)
+{
+ ws_client_t *client = (ws_client_t *)arg;
+ ws_client_ctx_t wctx;
+ hybbx_result_t rc;
+ char remote[64];
+
+ if (client == NULL || g_service == NULL) {
+ free(client);
+ return NULL;
+ }
+
+ remote[0] = '\0';
+ (void)hybbx_socket_peer_name(client->ws.fd, remote, sizeof(remote));
+
+ if (hybbx_ws_tls_server_enabled()) {
+ void *tls = hybbx_ws_tls_accept(client->ws.fd);
+
+ if (tls == NULL) {
+ goto cleanup;
+ }
+ hybbx_ws_connection_set_tls(&client->ws, tls);
+ }
+
+ rc = hybbx_ws_server_handshake(&client->ws, g_config.path);
+ if (rc != HYBBX_OK) {
+ goto cleanup;
+ }
+
+ if (!ws_client_acquire_slot(client)) {
+ ws_send_limit(client);
+ goto cleanup;
+ }
+
+ rc = hybbx_session_open(g_service, &hybbx_plugin_websocket, client,
+ &client->hbx_session);
+ if (rc != HYBBX_OK || client->hbx_session == NULL) {
+ if (rc == HYBBX_ERR_BUSY) {
+ ws_send_busy(client);
+ }
+ goto cleanup;
+ }
+
+ if (remote[0] != '\0') {
+ (void)hybbx_session_set_remote(client->hbx_session, remote);
+ }
+
+ wctx.client = client;
+ client->last_rc = HYBBX_OK;
+ client->last_io_at = time(NULL);
+
+ while (g_ws_running && client->ws.established) {
+ struct pollfd pfd;
+ int pr;
+
+ pfd.fd = client->ws.fd;
+ pfd.events = POLLIN;
+ pfd.revents = 0;
+
+ pr = poll(&pfd, 1, WS_CLIENT_POLL_MS);
+ if (pr < 0) {
+ if (errno == EINTR) {
+ continue;
+ }
+ break;
+ }
+ if (pr == 0) {
+ time_t now = time(NULL);
+
+ if (now != (time_t)-1 &&
+ (client->last_io_at == 0 ||
+ now - client->last_io_at >= WS_CLIENT_PING_IDLE_SEC)) {
+ rc = hybbx_ws_ping(&client->ws);
+ if (rc != HYBBX_OK) {
+ break;
+ }
+ client->last_io_at = now;
+ }
+
+ rc = hybbx_session_tick(client->hbx_session);
+ if (rc == HYBBX_SESSION_END) {
+ client->last_rc = HYBBX_SESSION_END;
+ break;
+ }
+ continue;
+ }
+ if ((pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) != 0) {
+ break;
+ }
+ if ((pfd.revents & POLLIN) == 0) {
+ continue;
+ }
+
+ rc = hybbx_ws_read_frames(&client->ws, ws_on_user_data, &wctx);
+ if (rc != HYBBX_OK) {
+ if (client->last_rc == HYBBX_SESSION_END) {
+ break;
+ }
+ break;
+ }
+ client->last_io_at = time(NULL);
+
+ if (client->last_rc == HYBBX_SESSION_END) {
+ break;
+ }
+
+ rc = hybbx_session_tick(client->hbx_session);
+ if (rc == HYBBX_SESSION_END) {
+ client->last_rc = HYBBX_SESSION_END;
+ break;
+ }
+ }
+
+cleanup:
+ if (client->hbx_session != NULL) {
+ hybbx_session_close(client->hbx_session);
+ client->hbx_session = NULL;
+ }
+
+ ws_client_release_slot(client);
+
+ (void)ws_tx_flush(client, 1);
+
+ if (client->ws.established) {
+ (void)hybbx_ws_close(&client->ws);
+ }
+
+ hybbx_ws_connection_cleanup(&client->ws);
+
+ if (client->ws.fd >= 0) {
+ shutdown(client->ws.fd, SHUT_RDWR);
+ close(client->ws.fd);
+ client->ws.fd = -1;
+ }
+
+ free(client);
+ return NULL;
+}
+
+static void accept_client(int client_fd)
+{
+ ws_client_t *client;
+ pthread_t thread;
+ pthread_attr_t attr;
+
+ client = calloc(1, sizeof(*client));
+ if (client == NULL) {
+ close(client_fd);
+ return;
+ }
+
+ hybbx_ws_connection_init(&client->ws, client_fd);
+
+ pthread_attr_init(&attr);
+ pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
+
+ if (pthread_create(&thread, &attr, ws_client_thread, client) != 0) {
+ close(client_fd);
+ free(client);
+ }
+
+ pthread_attr_destroy(&attr);
+}
+
+static void *ws_accept_thread(void *arg)
+{
+ (void)arg;
+
+ while (g_ws_running) {
+ struct pollfd fds[2];
+ nfds_t nfds = 0;
+ int i;
+ int ready;
+
+ memset(fds, 0, sizeof(fds));
+
+ if (g_listen_v4 >= 0) {
+ fds[nfds].fd = g_listen_v4;
+ fds[nfds].events = POLLIN;
+ nfds++;
+ }
+
+ if (g_listen_v6 >= 0) {
+ fds[nfds].fd = g_listen_v6;
+ fds[nfds].events = POLLIN;
+ nfds++;
+ }
+
+ if (nfds == 0) {
+ break;
+ }
+
+ ready = poll(fds, nfds, 500);
+ if (ready < 0) {
+ if (errno == EINTR) {
+ continue;
+ }
+ break;
+ }
+
+ if (ready == 0) {
+ continue;
+ }
+
+ for (i = 0; i < (int)nfds; i++) {
+ if (fds[i].revents & POLLIN) {
+ int client_fd = accept(fds[i].fd, NULL, NULL);
+
+ if (client_fd >= 0) {
+ if (!hybbx_security_ban_accept_fd(client_fd)) {
+ close(client_fd);
+ continue;
+ }
+ accept_client(client_fd);
+ }
+ }
+ }
+ }
+
+ return NULL;
+}
+
+static hybbx_result_t ws_plugin_init(hybbx_service_t *service)
+{
+ g_service = service;
+ return HYBBX_OK;
+}
+
+static void ws_plugin_shutdown(void)
+{
+ ws_plugin_stop();
+}
+
+static hybbx_result_t ws_plugin_start(const char *config)
+{
+ hybbx_result_t rc;
+
+ if (g_ws_running) {
+ return HYBBX_ERR_BUSY;
+ }
+
+ rc = hybbx_websocket_config_parse(config, &g_config);
+ if (rc != HYBBX_OK) {
+ return rc;
+ }
+
+ g_listen_v4 = -1;
+ g_listen_v6 = -1;
+
+ if (g_config.ipv4) {
+ g_listen_v4 = create_listen_socket(AF_INET, g_config.bind_v4,
+ g_config.port);
+ if (g_listen_v4 < 0) {
+ log_bind_failure(g_config.bind_v4, g_config.port);
+ return HYBBX_ERR_IO;
+ }
+ }
+
+ if (g_config.ipv6) {
+ g_listen_v6 = create_listen_socket(AF_INET6, g_config.bind_v6,
+ g_config.port);
+ if (g_listen_v6 < 0) {
+ hybbx_log_warn("[websocket] IPv6 bind [%s]:%u skipped (%s)",
+ g_config.bind_v6, g_config.port, strerror(errno));
+ }
+ }
+
+ if (g_listen_v4 < 0 && g_listen_v6 < 0) {
+ return HYBBX_ERR_IO;
+ }
+
+ if (hybbx_ws_tls_compiled()) {
+ char cert_dir[HYBBX_PATH_MAX];
+ hybbx_result_t tls_rc;
+
+ if (hybbx_path_resolve(cert_dir, sizeof(cert_dir),
+ g_config.cert_dir) != HYBBX_OK) {
+ hybbx_strlcpy(cert_dir, g_config.cert_dir, sizeof(cert_dir));
+ }
+
+ tls_rc = hybbx_ws_tls_ensure_certs(cert_dir);
+
+ if (tls_rc == HYBBX_OK) {
+ tls_rc = hybbx_ws_tls_server_start(cert_dir);
+ }
+ if (tls_rc != HYBBX_OK) {
+ hybbx_log_warn("[websocket] TLS init failed (%d), plain ws only",
+ (int)tls_rc);
+ }
+ }
+
+ g_ws_running = 1;
+ g_active_clients = 0;
+
+ if (pthread_create(&g_accept_thread, NULL, ws_accept_thread, NULL) != 0) {
+ g_ws_running = 0;
+ if (g_listen_v4 >= 0) {
+ close(g_listen_v4);
+ g_listen_v4 = -1;
+ }
+ if (g_listen_v6 >= 0) {
+ close(g_listen_v6);
+ g_listen_v6 = -1;
+ }
+ return HYBBX_ERR_IO;
+ }
+
+ {
+ char msg[384];
+ size_t pos = 0;
+
+ pos += (size_t)snprintf(msg + pos, sizeof(msg) - pos, "[websocket] listening");
+ if (g_listen_v4 >= 0) {
+ pos += (size_t)snprintf(msg + pos, sizeof(msg) - pos, " IPv4 %s:%u",
+ g_config.bind_v4, g_config.port);
+ }
+ if (g_listen_v6 >= 0) {
+ pos += (size_t)snprintf(msg + pos, sizeof(msg) - pos, " IPv6 [%s]:%u",
+ g_config.bind_v6, g_config.port);
+ }
+ pos += (size_t)snprintf(msg + pos, sizeof(msg) - pos,
+ " path=%s max_connections=%u (%s, forward-proxy)",
+ g_config.path, g_config.max_connections,
+ hybbx_ws_tls_server_enabled() ? "wss" : "ws");
+ hybbx_log_info("%s", msg);
+ }
+
+ return HYBBX_OK;
+}
+
+static hybbx_result_t ws_plugin_stop(void)
+{
+ if (!g_ws_running) {
+ return HYBBX_OK;
+ }
+
+ g_ws_running = 0;
+
+ if (g_listen_v4 >= 0) {
+ shutdown(g_listen_v4, SHUT_RDWR);
+ close(g_listen_v4);
+ g_listen_v4 = -1;
+ }
+
+ if (g_listen_v6 >= 0) {
+ shutdown(g_listen_v6, SHUT_RDWR);
+ close(g_listen_v6);
+ g_listen_v6 = -1;
+ }
+
+ pthread_join(g_accept_thread, NULL);
+ hybbx_ws_tls_server_stop();
+ hybbx_log_info("[websocket] stop");
+ return HYBBX_OK;
+}
+
+const hybbx_transport_plugin_t hybbx_plugin_websocket = {
+ .name = "websocket",
+ .kind = HYBBX_TRANSPORT_WEBSOCKET,
+ .version = 1,
+ .init = ws_plugin_init,
+ .shutdown = ws_plugin_shutdown,
+ .start = ws_plugin_start,
+ .stop = ws_plugin_stop,
+ .write = ws_plugin_write,
+};
git clone -b <branch> https://cgit.mode42.com/<repo>.git
git clone -b <branch> git://cgit.mode42.com/<repo>.git


info@mode42.com