1#if defined(CONF_WEBSOCKETS)
2
3#include "websockets.h"
4
5#include <base/dbg.h>
6#include <base/log.h>
7#include <base/mem.h>
8#include <base/str.h>
9
10#include <engine/shared/config.h>
11#include <engine/shared/network.h>
12#include <engine/shared/protocol.h>
13#include <engine/shared/ringbuffer.h>
14
15#if defined(CONF_FAMILY_UNIX)
16#include <arpa/inet.h>
17#elif defined(CONF_FAMILY_WINDOWS)
18#include <ws2tcpip.h>
19#endif
20#include <libwebsockets.h>
21
22#include <cstdlib>
23#include <map>
24#include <string>
25
26// NOLINTBEGIN(readability-identifier-naming)
27struct websocket_chunk
28{
29 size_t size;
30 size_t read;
31 NETADDR addr;
32 unsigned char data[0];
33};
34
35// Client opens two connections for whatever reason
36typedef CStaticRingBuffer<websocket_chunk, (MAX_CLIENTS * 2) * NET_CONN_BUFFERSIZE,
37 CRingBufferBase::FLAG_RECYCLE>
38 TRecvBuffer;
39typedef CStaticRingBuffer<websocket_chunk, NET_CONN_BUFFERSIZE,
40 CRingBufferBase::FLAG_RECYCLE>
41 TSendBuffer;
42
43struct per_session_data
44{
45 lws *wsi;
46 NETADDR addr;
47 TSendBuffer send_buffer;
48};
49
50struct context_data
51{
52 char bindaddr_str[NETADDR_MAXSTRSIZE];
53 lws_context_creation_info creation_info;
54 lws_context *context;
55 std::map<NETADDR, per_session_data *> port_map;
56 TRecvBuffer recv_buffer;
57};
58
59// Client has main, dummy and contact connections with IPv4 and IPv6
60static context_data contexts[3 * 2];
61static std::map<lws_context *, context_data *> contexts_map;
62
63static lws_context *websocket_context(int socket)
64{
65 dbg_assert(socket >= 0 && socket < (int)std::size(contexts), "socket index invalid: %d", socket);
66 lws_context *context = contexts[socket].context;
67 dbg_assert(context != nullptr, "socket context not initialized: %d", socket);
68 return context;
69}
70
71static void receive_chunk(context_data *ctx_data, per_session_data *pss, const void *in, size_t len)
72{
73 websocket_chunk *chunk = ctx_data->recv_buffer.Allocate(Size: len + sizeof(websocket_chunk));
74 dbg_assert(chunk != nullptr, "failed to allocate websocket receive buffer chunk of size %" PRIzu, len);
75 chunk->size = len;
76 chunk->read = 0;
77 chunk->addr = pss->addr;
78 mem_copy(dest: &chunk->data[0], source: in, size: len);
79}
80
81static void sockaddr_to_netaddr_websocket(const sockaddr *src, socklen_t src_len, NETADDR *dst)
82{
83 *dst = NETADDR_ZEROED;
84 if(src->sa_family == AF_INET && src_len >= (socklen_t)sizeof(sockaddr_in))
85 {
86 const sockaddr_in *src_in = (const sockaddr_in *)src;
87 dst->type = NETTYPE_WEBSOCKET_IPV4;
88 dst->port = htons(hostshort: src_in->sin_port);
89 static_assert(sizeof(dst->ip) >= sizeof(src_in->sin_addr.s_addr));
90 mem_copy(dest: dst->ip, source: &src_in->sin_addr.s_addr, size: sizeof(src_in->sin_addr.s_addr));
91 }
92 else if(src->sa_family == AF_INET6 && src_len >= (socklen_t)sizeof(sockaddr_in6))
93 {
94 const sockaddr_in6 *src_in6 = (const sockaddr_in6 *)src;
95 dst->type = NETTYPE_WEBSOCKET_IPV6;
96 dst->port = htons(hostshort: src_in6->sin6_port);
97 static_assert(sizeof(dst->ip) >= sizeof(src_in6->sin6_addr.s6_addr));
98 mem_copy(dest: dst->ip, source: &src_in6->sin6_addr.s6_addr, size: sizeof(src_in6->sin6_addr.s6_addr));
99 }
100 else
101 {
102 log_warn("websockets", "Cannot convert sockaddr of family %d", src->sa_family);
103 }
104}
105
106static int websocket_protocol_callback(lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len)
107{
108 per_session_data *pss = (per_session_data *)user;
109 lws_context *context = lws_get_context(wsi);
110 context_data *ctx_data = contexts_map[context];
111 switch(reason)
112 {
113 case LWS_CALLBACK_WSI_CREATE:
114 if(pss == nullptr)
115 {
116 return 0;
117 }
118 [[fallthrough]];
119 case LWS_CALLBACK_ESTABLISHED:
120 {
121 sockaddr_storage peersockaddr;
122 socklen_t peersockaddr_size = sizeof(peersockaddr);
123 getpeername(fd: lws_get_socket_fd(wsi), addr: (sockaddr *)&peersockaddr, len: &peersockaddr_size);
124 NETADDR addr;
125 sockaddr_to_netaddr_websocket(src: (sockaddr *)&peersockaddr, src_len: peersockaddr_size, dst: &addr);
126 if(addr.type == NETTYPE_INVALID)
127 {
128 return 0;
129 }
130
131 pss->wsi = wsi;
132 pss->addr = addr;
133 pss->send_buffer.Init();
134 ctx_data->port_map[addr] = pss;
135
136 char addr_str[NETADDR_MAXSTRSIZE];
137 net_addr_str(addr: &addr, string: addr_str, max_length: sizeof(addr_str), add_port: true);
138 log_trace("websockets", "Connection established with '%s'", addr_str);
139 return 0;
140 }
141
142 case LWS_CALLBACK_CLIENT_CLOSED:
143 [[fallthrough]];
144 case LWS_CALLBACK_CLOSED:
145 {
146 char addr_str[NETADDR_MAXSTRSIZE];
147 net_addr_str(addr: &pss->addr, string: addr_str, max_length: sizeof(addr_str), add_port: true);
148 log_trace("websockets", "Connection closed with '%s'", addr_str);
149
150 static const unsigned char CLOSE_PACKET[] = {0x10, 0x0e, 0x00, 0x04};
151 receive_chunk(ctx_data, pss, in: &CLOSE_PACKET, len: sizeof(CLOSE_PACKET));
152 return 0;
153 }
154
155 case LWS_CALLBACK_WSI_DESTROY:
156 {
157 if(pss == nullptr)
158 {
159 return 0;
160 }
161 pss->wsi = nullptr;
162 ctx_data->port_map.erase(x: pss->addr);
163 return 0;
164 }
165
166 case LWS_CALLBACK_CLIENT_WRITEABLE:
167 [[fallthrough]];
168 case LWS_CALLBACK_SERVER_WRITEABLE:
169 {
170 websocket_chunk *chunk = pss->send_buffer.First();
171 if(chunk == nullptr)
172 {
173 return 0;
174 }
175
176 int chunk_len = chunk->size - chunk->read;
177 int n = lws_write(wsi, buf: &chunk->data[LWS_SEND_BUFFER_PRE_PADDING + chunk->read], len: chunk->size - chunk->read, protocol: LWS_WRITE_BINARY);
178 if(n < 0)
179 {
180 return 1;
181 }
182
183 if(n < chunk_len)
184 {
185 chunk->read += n;
186 lws_callback_on_writable(wsi);
187 return 0;
188 }
189
190 pss->send_buffer.PopFirst();
191 lws_callback_on_writable(wsi);
192 return 0;
193 }
194
195 case LWS_CALLBACK_CLIENT_RECEIVE:
196 [[fallthrough]];
197 case LWS_CALLBACK_RECEIVE:
198 receive_chunk(ctx_data, pss, in, len);
199 return 0;
200
201 default:
202 return 0;
203 }
204}
205
206static const lws_protocols protocols[] = {
207 {.name: "binary", .callback: websocket_protocol_callback, .per_session_data_size: sizeof(per_session_data)},
208 {.name: "base64", .callback: websocket_protocol_callback, .per_session_data_size: sizeof(per_session_data)},
209 {.name: nullptr, .callback: nullptr, .per_session_data_size: 0}};
210
211static LEVEL websocket_level_to_loglevel(int level)
212{
213 switch(level)
214 {
215 case LLL_ERR:
216 return LEVEL_ERROR;
217 case LLL_WARN:
218 return LEVEL_WARN;
219 case LLL_NOTICE:
220 case LLL_INFO:
221 return LEVEL_DEBUG;
222 default:
223 dbg_assert_failed("invalid log level: %d", level);
224 }
225}
226
227static void websocket_log_callback(int level, const char *line)
228{
229 if((level == LLL_NOTICE || level == LLL_INFO) && !g_Config.m_DbgWebsockets)
230 {
231 return;
232 }
233
234 // Truncate duplicate timestamp from beginning and newline from end
235 char line_truncated[4096]; // Longest log line length
236 const char *line_time_end = str_find(haystack: line, needle: "] ");
237 dbg_assert(line_time_end != nullptr, "unexpected log format");
238 str_copy(dst&: line_truncated, src: line_time_end + 2);
239 const int length = str_length(str: line_truncated);
240 if(line_truncated[length - 1] == '\n')
241 {
242 line_truncated[length - 1] = '\0';
243 }
244 if(line_truncated[length - 2] == '\r')
245 {
246 line_truncated[length - 2] = '\0';
247 }
248 log_log(level: websocket_level_to_loglevel(level), sys: "websockets", fmt: "%s", line_truncated);
249}
250
251void websocket_init()
252{
253 lws_set_log_level(LLL_ERR | LLL_WARN | LLL_NOTICE | LLL_INFO, log_emit_function: websocket_log_callback);
254}
255
256int websocket_create(const NETADDR *bindaddr)
257{
258 // find free context
259 int first_free = -1;
260 for(int i = 0; i < (int)std::size(contexts); i++)
261 {
262 if(contexts[i].context == nullptr)
263 {
264 first_free = i;
265 break;
266 }
267 }
268 if(first_free == -1)
269 {
270 log_error("websockets", "Failed to create websocket: no free contexts available");
271 return -1;
272 }
273
274 context_data *ctx_data = &contexts[first_free];
275 mem_zero(block: &ctx_data->creation_info, size: sizeof(ctx_data->creation_info));
276 ctx_data->creation_info.options = LWS_SERVER_OPTION_FAIL_UPON_UNABLE_TO_BIND;
277 if(bindaddr->type == NETTYPE_WEBSOCKET_IPV6)
278 {
279 // Set IPv6-only mode and socket option for IPv6 Websockets.
280 ctx_data->creation_info.options |= LWS_SERVER_OPTION_IPV6_V6ONLY_VALUE | LWS_SERVER_OPTION_IPV6_V6ONLY_MODIFY;
281 }
282 net_addr_str(addr: bindaddr, string: ctx_data->bindaddr_str, max_length: sizeof(ctx_data->bindaddr_str), add_port: false);
283 if(ctx_data->bindaddr_str[0] == '[' && ctx_data->bindaddr_str[str_length(str: ctx_data->bindaddr_str) - 1] == ']')
284 {
285 // Bindaddr must not be enclosed in brackets for IPv6 Websockets.
286 ctx_data->bindaddr_str[str_length(str: ctx_data->bindaddr_str) - 1] = '\0';
287 mem_move(dest: &ctx_data->bindaddr_str[0], source: &ctx_data->bindaddr_str[1], size: str_length(str: ctx_data->bindaddr_str) + 1);
288 }
289 ctx_data->creation_info.iface = ctx_data->bindaddr_str;
290 ctx_data->creation_info.port = bindaddr->port;
291 ctx_data->creation_info.protocols = protocols;
292 ctx_data->creation_info.gid = -1;
293 ctx_data->creation_info.uid = -1;
294 ctx_data->creation_info.user = ctx_data;
295
296 ctx_data->context = lws_create_context(info: &ctx_data->creation_info);
297 if(ctx_data->context == nullptr)
298 {
299 return -1;
300 }
301 contexts_map[ctx_data->context] = ctx_data;
302 ctx_data->recv_buffer.Init();
303 return first_free;
304}
305
306void websocket_destroy(int socket)
307{
308 lws_context *context = websocket_context(socket);
309 lws_context_destroy(context);
310 contexts_map.erase(x: context);
311 contexts[socket].context = nullptr;
312}
313
314int websocket_recv(int socket, unsigned char *data, size_t maxsize, NETADDR *addr)
315{
316 lws_context *context = websocket_context(socket);
317 const int service_result = lws_service(context, timeout_ms: -1);
318 if(service_result < 0)
319 {
320 return service_result;
321 }
322
323 context_data *ctx_data = contexts_map[context];
324 websocket_chunk *chunk = ctx_data->recv_buffer.First();
325 if(chunk == nullptr)
326 {
327 return 0;
328 }
329
330 if(maxsize >= chunk->size - chunk->read)
331 {
332 const int len = chunk->size - chunk->read;
333 mem_copy(dest: data, source: &chunk->data[chunk->read], size: len);
334 *addr = chunk->addr;
335 ctx_data->recv_buffer.PopFirst();
336 return len;
337 }
338 else
339 {
340 mem_copy(dest: data, source: &chunk->data[chunk->read], size: maxsize);
341 *addr = chunk->addr;
342 chunk->read += maxsize;
343 return maxsize;
344 }
345}
346
347int websocket_send(int socket, const unsigned char *data, size_t size, const NETADDR *addr)
348{
349 lws_context *context = websocket_context(socket);
350 context_data *ctx_data = contexts_map[context];
351 per_session_data *pss = ctx_data->port_map[*addr];
352 if(pss == nullptr)
353 {
354 char addr_str[NETADDR_MAXSTRSIZE];
355 net_addr_str(addr, string: addr_str, max_length: sizeof(addr_str), add_port: false);
356 lws_client_connect_info ccinfo = {};
357 ccinfo.context = context;
358 ccinfo.address = addr_str;
359 ccinfo.port = addr->port;
360 ccinfo.protocol = protocols[0].name;
361 lws *wsi = lws_client_connect_via_info(ccinfo: &ccinfo);
362 if(wsi == nullptr)
363 {
364 return -1;
365 }
366 lws_service(context, timeout_ms: -1);
367 pss = ctx_data->port_map[*addr];
368 if(pss == nullptr)
369 {
370 return -1;
371 }
372 }
373
374 const size_t chunk_size = size + sizeof(websocket_chunk) + LWS_SEND_BUFFER_PRE_PADDING + LWS_SEND_BUFFER_POST_PADDING;
375 websocket_chunk *chunk = pss->send_buffer.Allocate(Size: chunk_size);
376 dbg_assert(chunk != nullptr, "failed to allocate websocket send buffer chunk of size %" PRIzu, size);
377 mem_zero(block: chunk, size: chunk_size);
378 chunk->size = size;
379 chunk->read = 0;
380 chunk->addr = pss->addr;
381 mem_copy(dest: &chunk->data[LWS_SEND_BUFFER_PRE_PADDING], source: data, size);
382 lws_callback_on_writable(wsi: pss->wsi);
383 lws_service(context, timeout_ms: -1);
384 return size;
385}
386
387int websocket_fd_set(int socket, fd_set *set)
388{
389 lws_context *context = websocket_context(socket);
390 lws_service(context, timeout_ms: -1);
391
392 context_data *ctx_data = contexts_map[context];
393 int max = 0;
394 for(const auto &[_, pss] : ctx_data->port_map)
395 {
396 if(pss == nullptr)
397 {
398 continue;
399 }
400 int fd = lws_get_socket_fd(wsi: pss->wsi);
401 max = std::max(a: fd, b: max);
402 FD_SET(fd, set);
403 }
404 return max;
405}
406
407int websocket_fd_get(int socket, fd_set *set)
408{
409 lws_context *context = websocket_context(socket);
410 lws_service(context, timeout_ms: -1);
411
412 context_data *ctx_data = contexts_map[context];
413 for(const auto &[_, pss] : ctx_data->port_map)
414 {
415 if(pss == nullptr)
416 {
417 continue;
418 }
419 if(FD_ISSET(lws_get_socket_fd(pss->wsi), set))
420 {
421 return 1;
422 }
423 }
424 return 0;
425}
426// NOLINTEND(readability-identifier-naming)
427
428#endif
429