1 /*
2 * libwebsockets - small server side websockets and web server implementation
3 *
4 * Copyright (C) 2019 - 2020 Andy Green <andy@warmcat.com>
5 *
6 * Permission is hereby granted, free of charge, to any person obtaining a copy
7 * of this software and associated documentation files (the "Software"), to
8 * deal in the Software without restriction, including without limitation the
9 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
10 * sell copies of the Software, and to permit persons to whom the Software is
11 * furnished to do so, subject to the following conditions:
12 *
13 * The above copyright notice and this permission notice shall be included in
14 * all copies or substantial portions of the Software.
15 *
16 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
19 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
21 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
22 * IN THE SOFTWARE.
23 *
24 *
25 * When the user code is in a different process, a non-tls unix domain socket
26 * proxy is used to asynchronusly transfer buffers in each direction via the
27 * network stack, without explicit IPC
28 *
29 * user_process{ [user code] | shim | socket-}------ lws_process{ lws }
30 *
31 * Lws exposes a listening unix domain socket in this case, the user processes
32 * connect to it and pass just info.streamtype in an initial tx packet. All
33 * packets are prepended by a 1-byte type field when used in this mode. See
34 * lws-secure-streams.h for documentation and definitions.
35 *
36 * Proxying in either direction can face the situation it cannot send the onward
37 * packet immediately and is subject to separating the write request from the
38 * write action. To make the best use of memory, a single preallocated buffer
39 * stashes pending packets in all four directions (c->p, p->c, p->ss, ss->p).
40 * This allows it to adapt to different traffic patterns without wasted areas
41 * dedicated to traffic that isn't coming in a particular application.
42 *
43 * A shim is provided to monitor the process' unix domain socket and regenerate
44 * the secure sockets api there with callbacks happening in the process thread
45 * context.
46 *
47 * This file implements the listening unix domain socket proxy... this code is
48 * only going to run on a Linux-class device with its implications about memory
49 * availability.
50 */
51
52 #include <private-lib-core.h>
53
54 /*
55 * Because both sides of the connection share the conn, we allocate it
56 * during accepted adoption, and both sides point to it.
57 *
58 * The last one of the accepted side and the onward side to close frees it.
59 */
60
61 struct conn {
62 struct lws_ss_serialization_parser parser;
63
64 lws_dsh_t *dsh; /* unified buffer for both sides */
65 struct lws *wsi; /* the client side */
66 lws_ss_handle_t *ss; /* the onward, ss side */
67
68 lws_ss_conn_states_t state;
69 };
70
71 struct raw_pss {
72 struct conn *conn;
73 };
74
75 /*
76 * Proxy - onward secure-stream handler
77 */
78
79 typedef struct ss_proxy_onward {
80 lws_ss_handle_t *ss;
81 struct conn *conn;
82 } ss_proxy_t;
83
84
85 /* secure streams payload interface */
86
87 static int
ss_proxy_onward_rx(void * userobj,const uint8_t * buf,size_t len,int flags)88 ss_proxy_onward_rx(void *userobj, const uint8_t *buf, size_t len, int flags)
89 {
90 ss_proxy_t *m = (ss_proxy_t *)userobj;
91 const char *rsp = NULL;
92 int n;
93
94 /*
95 * The onward secure stream connection has received something.
96 */
97
98 if (m->ss->rideshare != m->ss->policy && m->ss->rideshare) {
99 rsp = m->ss->rideshare->streamtype;
100 flags |= LWSSS_FLAG_RIDESHARE;
101 }
102
103 n = lws_ss_serialize_rx_payload(m->conn->dsh, buf, len, flags, rsp);
104 if (n)
105 return n;
106
107 if (m->conn->wsi) /* if possible, request client conn write */
108 lws_callback_on_writable(m->conn->wsi);
109
110 return 0;
111 }
112
113 /*
114 * we are transmitting buffered payload originally from the client on to the ss
115 */
116
117 static int
ss_proxy_onward_tx(void * userobj,lws_ss_tx_ordinal_t ord,uint8_t * buf,size_t * len,int * flags)118 ss_proxy_onward_tx(void *userobj, lws_ss_tx_ordinal_t ord, uint8_t *buf,
119 size_t *len, int *flags)
120 {
121 ss_proxy_t *m = (ss_proxy_t *)userobj;
122 void *p;
123 size_t si;
124
125 if (!m->conn->ss || m->conn->state != LPCS_OPERATIONAL) {
126 lwsl_notice("%s: ss not ready\n", __func__);
127 *len = 0;
128
129 return 1;
130 }
131
132 /*
133 * The onward secure stream says that we could send something to it
134 * (by putting it in buf, and setting *len and *flags)
135 */
136
137 if (lws_ss_deserialize_tx_payload(m->conn->dsh, m->ss->wsi,
138 ord, buf, len, flags))
139 return 1;
140
141 if (!lws_dsh_get_head(m->conn->dsh, KIND_C_TO_P, (void **)&p, &si))
142 lws_ss_request_tx(m->conn->ss);
143
144 if (!*len && !*flags)
145 return 1; /* we don't actually want to send anything */
146
147 lwsl_info("%s: onward tx %d fl 0x%x\n", __func__, (int)*len, *flags);
148
149 #if 0
150 {
151 int ff = open("/tmp/z", O_RDWR | O_CREAT | O_APPEND, 0666);
152 if (ff == -1)
153 lwsl_err("%s: errno %d\n", __func__, errno);
154 write(ff, buf, *len);
155 close(ff);
156 }
157 #endif
158
159 return 0;
160 }
161
162 static int
ss_proxy_onward_state(void * userobj,void * sh,lws_ss_constate_t state,lws_ss_tx_ordinal_t ack)163 ss_proxy_onward_state(void *userobj, void *sh,
164 lws_ss_constate_t state, lws_ss_tx_ordinal_t ack)
165 {
166 ss_proxy_t *m = (ss_proxy_t *)userobj;
167
168 switch (state) {
169 case LWSSSCS_CREATING:
170 break;
171
172 case LWSSSCS_DESTROYING:
173 if (!m->conn)
174 break;
175 if (!m->conn->wsi) {
176 /*
177 * Our onward secure stream is closing and our client
178 * connection has already gone away... destroy the conn.
179 */
180 lwsl_info("%s: Destroying conn\n", __func__);
181 lws_dsh_destroy(&m->conn->dsh);
182 free(m->conn);
183 m->conn = NULL;
184 return 0;
185 } else
186 lwsl_info("%s: ss DESTROYING, wsi up\n", __func__);
187 break;
188
189 default:
190 break;
191 }
192 if (!m->conn) {
193 lwsl_warn("%s: dropping state due to conn not up\n", __func__);
194
195 return 0;
196 }
197
198 lws_ss_serialize_state(m->conn->dsh, state, ack);
199
200 if (m->conn->wsi) /* if possible, request client conn write */
201 lws_callback_on_writable(m->conn->wsi);
202
203 return 0;
204 }
205
206 void
ss_proxy_onward_txcr(void * userobj,int bump)207 ss_proxy_onward_txcr(void *userobj, int bump)
208 {
209 ss_proxy_t *m = (ss_proxy_t *)userobj;
210
211 if (!m->conn)
212 return;
213
214 lws_ss_serialize_txcr(m->conn->dsh, bump);
215
216 if (m->conn->wsi) /* if possible, request client conn write */
217 lws_callback_on_writable(m->conn->wsi);
218 }
219
220 /*
221 * Client - Proxy connection on unix domain socket
222 */
223
224 static int
callback_ss_proxy(struct lws * wsi,enum lws_callback_reasons reason,void * user,void * in,size_t len)225 callback_ss_proxy(struct lws *wsi, enum lws_callback_reasons reason,
226 void *user, void *in, size_t len)
227 {
228 struct lws_context_per_thread *pt = &wsi->context->pt[(int)wsi->tsi];
229 struct raw_pss *pss = (struct raw_pss *)user;
230 const lws_ss_policy_t *rsp;
231 struct conn *conn = NULL;
232 lws_ss_info_t ssi;
233 const uint8_t *cp;
234 #if defined(LWS_WITH_DETAILED_LATENCY)
235 lws_usec_t us;
236 #endif
237 char s[128];
238 uint8_t *p;
239 size_t si;
240 char pay;
241 int n;
242
243 if (pss)
244 conn = pss->conn;
245
246 switch (reason) {
247 case LWS_CALLBACK_PROTOCOL_INIT:
248 break;
249
250 case LWS_CALLBACK_PROTOCOL_DESTROY:
251 break;
252
253 /* callbacks related to raw socket descriptor "accepted side" */
254
255 case LWS_CALLBACK_RAW_ADOPT:
256 lwsl_info("LWS_CALLBACK_RAW_ADOPT\n");
257 if (!pss)
258 return -1;
259 pss->conn = malloc(sizeof(struct conn));
260 if (!pss->conn)
261 return -1;
262 memset(pss->conn, 0, sizeof(*pss->conn));
263
264 pss->conn->dsh = lws_dsh_create(&pt->ss_dsh_owner,
265 LWS_SS_MTU * 160, 2);
266 if (!pss->conn->dsh) {
267 free(pss->conn);
268
269 return -1;
270 }
271
272 pss->conn->wsi = wsi;
273 pss->conn->state = LPCS_WAIT_INITIAL_TX;
274
275 /*
276 * Client is expected to follow the unix domain socket
277 * acceptance up rapidly with an initial tx containing the
278 * streamtype name. We can't create the stream until then.
279 */
280 lws_set_timeout(wsi,
281 PENDING_TIMEOUT_AWAITING_CLIENT_HS_SEND, 3);
282 break;
283
284 case LWS_CALLBACK_RAW_CLOSE:
285 lwsl_info("LWS_CALLBACK_RAW_CLOSE:\n");
286
287 /*
288 * the client unix domain socket connection has closed...
289 * eg, client has exited or otherwise has definitively finished
290 * with the proxying and onward connection
291 */
292
293 if (!conn)
294 break;
295
296 if (conn->ss) {
297 lwsl_info("%s: destroying ss\n", __func__);
298 /* sever relationship with ss about to be deleted */
299 lws_set_opaque_user_data(wsi, NULL);
300
301 conn->wsi = NULL;
302
303
304 lws_ss_destroy(&conn->ss);
305 /* conn may have gone */
306 break;
307 }
308
309 if (conn->state == LPCS_DESTROYED || !conn->ss) {
310 /*
311 * There's no onward secure stream and our client
312 * connection is closing. Destroy the conn.
313 */
314 lws_dsh_destroy(&conn->dsh);
315 free(conn);
316 pss->conn = NULL;
317 } else
318 lwsl_debug("%s: CLOSE; ss=%p\n", __func__, conn->ss);
319
320 break;
321
322 case LWS_CALLBACK_RAW_RX:
323 lwsl_info("%s: RX: rx %d\n", __func__, (int)len);
324
325 if (!conn || !conn->wsi) {
326 lwsl_err("%s: rx with bad conn state\n", __func__);
327
328 return -1;
329 }
330
331 // lwsl_hexdump_info(in, len);
332
333 if (conn->state == LPCS_WAIT_INITIAL_TX) {
334 memset(&ssi, 0, sizeof(ssi));
335 ssi.user_alloc = sizeof(ss_proxy_t);
336 ssi.handle_offset = offsetof(ss_proxy_t, ss);
337 ssi.opaque_user_data_offset =
338 offsetof(ss_proxy_t, conn);
339 ssi.rx = ss_proxy_onward_rx;
340 ssi.tx = ss_proxy_onward_tx;
341 ssi.state = ss_proxy_onward_state;
342 }
343
344 if (lws_ss_deserialize_parse(&conn->parser,
345 lws_get_context(wsi), conn->dsh, in, len,
346 &conn->state, conn, &conn->ss, &ssi, 0)) {
347 lwsl_err("%s: RAW_RX: deserialize_parse fail\n", __func__);
348 return -1;
349 }
350
351 if (conn->state == LPCS_REPORTING_FAIL ||
352 conn->state == LPCS_REPORTING_OK)
353 lws_callback_on_writable(conn->wsi);
354
355 break;
356
357 case LWS_CALLBACK_RAW_WRITEABLE:
358 // lwsl_notice("LWS_CALLBACK_RAW_PROXY_SRV_WRITEABLE\n");
359
360 /*
361 * We can transmit something back to the client from the dsh
362 * of stuff we received on its behalf from the ss
363 */
364
365 if (!conn || !conn->wsi)
366 break;
367
368 n = 0;
369 pay = 0;
370 s[3] = 0;
371 cp = (const uint8_t *)s;
372 switch (conn->state) {
373 case LPCS_REPORTING_FAIL:
374 s[3] = 1;
375 /* fallthru */
376 case LPCS_REPORTING_OK:
377 s[0] = LWSSS_SER_RXPRE_CREATE_RESULT;
378 s[1] = 0;
379 s[2] = 1;
380
381 n = 4;
382
383 /*
384 * If there's rideshare sequencing, it's added after the
385 * first 4 bytes or the create result, comma-separated
386 */
387
388 rsp = conn->ss->policy;
389
390 while (rsp) {
391 if (n != 4 && n < (int)sizeof(s) - 2)
392 s[n++] = ',';
393 n += lws_snprintf(&s[n], sizeof(s) - n,
394 "%s", rsp->streamtype);
395 rsp = lws_ss_policy_lookup(wsi->context,
396 rsp->rideshare_streamtype);
397 }
398 s[2] = n - 3;
399 conn->state = LPCS_OPERATIONAL;
400 lws_set_timeout(wsi, 0, 0);
401 break;
402 case LPCS_OPERATIONAL:
403 if (lws_dsh_get_head(conn->dsh, KIND_SS_TO_P,
404 (void **)&p, &si))
405 break;
406 cp = p;
407
408 #if defined(LWS_WITH_DETAILED_LATENCY)
409 if (cp[0] == LWSSS_SER_RXPRE_RX_PAYLOAD &&
410 wsi->context->detailed_latency_cb) {
411
412 /*
413 * we're fulfilling rx that came in on ss
414 * by sending it back out to the client on
415 * the Unix Domain Socket
416 *
417 * + 7 u32 write will compute latency here...
418 * + 11 u32 ust we received from ss
419 *
420 * lws_write will report it and fill in
421 * LAT_DUR_PROXY_CLIENT_REQ_TO_WRITE
422 */
423
424 us = lws_now_usecs();
425 lws_ser_wu32be(&p[7], us -
426 lws_ser_ru64be(&p[11]));
427 lws_ser_wu64be(&p[11], us);
428
429 wsi->detlat.acc_size =
430 wsi->detlat.req_size = si - 19;
431 /* time proxy held it */
432 wsi->detlat.latencies[
433 LAT_DUR_PROXY_RX_TO_ONWARD_TX] =
434 lws_ser_ru32be(&p[7]);
435 }
436 #endif
437
438 pay = 1;
439 n = (int)si;
440 break;
441 default:
442 break;
443 }
444 again:
445 if (!n)
446 break;
447
448 n = lws_write(wsi, (uint8_t *)cp, n, LWS_WRITE_RAW);
449 if (n < 0) {
450 lwsl_info("%s: WRITEABLE: %d\n", __func__, n);
451
452 goto hangup;
453 }
454
455 switch (conn->state) {
456 case LPCS_REPORTING_FAIL:
457 goto hangup;
458 case LPCS_OPERATIONAL:
459 if (pay)
460 lws_dsh_free((void **)&p);
461 if (!lws_dsh_get_head(conn->dsh, KIND_SS_TO_P,
462 (void **)&p, &si)) {
463 if (!lws_send_pipe_choked(wsi)) {
464 cp = p;
465 pay = 1;
466 n = (int)si;
467 goto again;
468 }
469 lws_callback_on_writable(wsi);
470 }
471 break;
472 default:
473 break;
474 }
475 break;
476
477 default:
478 break;
479 }
480
481 return lws_callback_http_dummy(wsi, reason, user, in, len);
482
483 hangup:
484 //lws_ss_destroy(&conn->ss);
485 //conn->state = LPCS_DESTROYED;
486
487 /* hang up on him */
488 return -1;
489 }
490
491 static const struct lws_protocols protocols[] = {
492 {
493 "ssproxy-protocol",
494 callback_ss_proxy,
495 sizeof(struct raw_pss),
496 2048, 2048, NULL, 0
497 },
498 { NULL, NULL, 0, 0, 0, NULL, 0 }
499 };
500
501 /*
502 * called from create_context()
503 */
504
505 int
lws_ss_proxy_create(struct lws_context * context,const char * bind,int port)506 lws_ss_proxy_create(struct lws_context *context, const char *bind, int port)
507 {
508 struct lws_context_creation_info info;
509
510 memset(&info, 0, sizeof(info));
511
512 info.vhost_name = "ssproxy";
513 info.options = LWS_SERVER_OPTION_ADOPT_APPLY_LISTEN_ACCEPT_CONFIG;
514 info.port = port;
515 if (!port) {
516 if (!bind)
517 bind = "@proxy.ss.lws";
518 info.options |= LWS_SERVER_OPTION_UNIX_SOCK;
519 }
520 info.iface = bind;
521 info.unix_socket_perms = "root:root";
522 info.listen_accept_role = "raw-skt";
523 info.listen_accept_protocol = "ssproxy-protocol";
524 info.protocols = protocols;
525
526 if (!lws_create_vhost(context, &info)) {
527 lwsl_err("%s: Failed to create ss proxy vhost\n", __func__);
528
529 return 1;
530 }
531
532 return 0;
533 }
534