1 /*
2 * libwebsockets - small server side websockets and web server implementation
3 *
4 * Copyright (C) 2019 - 2021 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 #include <private-lib-core.h>
26
27 static const struct ss_pcols *ss_pcols[] = {
28 #if defined(LWS_ROLE_H1)
29 &ss_pcol_h1, /* LWSSSP_H1 */
30 #else
31 NULL,
32 #endif
33 #if defined(LWS_ROLE_H2)
34 &ss_pcol_h2, /* LWSSSP_H2 */
35 #else
36 NULL,
37 #endif
38 #if defined(LWS_ROLE_WS)
39 &ss_pcol_ws, /* LWSSSP_WS */
40 #else
41 NULL,
42 #endif
43 #if defined(LWS_ROLE_MQTT)
44 &ss_pcol_mqtt, /* LWSSSP_MQTT */
45 #else
46 NULL,
47 #endif
48 &ss_pcol_raw, /* LWSSSP_RAW */
49 NULL,
50 };
51
52 static const char *state_names[] = {
53 "(unset)",
54 "LWSSSCS_CREATING",
55 "LWSSSCS_DISCONNECTED",
56 "LWSSSCS_UNREACHABLE",
57 "LWSSSCS_AUTH_FAILED",
58 "LWSSSCS_CONNECTED",
59 "LWSSSCS_CONNECTING",
60 "LWSSSCS_DESTROYING",
61 "LWSSSCS_POLL",
62 "LWSSSCS_ALL_RETRIES_FAILED",
63 "LWSSSCS_QOS_ACK_REMOTE",
64 "LWSSSCS_QOS_NACK_REMOTE",
65 "LWSSSCS_QOS_ACK_LOCAL",
66 "LWSSSCS_QOS_NACK_LOCAL",
67 "LWSSSCS_TIMEOUT",
68 "LWSSSCS_SERVER_TXN",
69 "LWSSSCS_SERVER_UPGRADE",
70 "LWSSSCS_EVENT_WAIT_CANCELLED",
71 "LWSSSCS_UPSTREAM_LINK_RETRY",
72 };
73
74 /*
75 * For each "current state", set bit offsets for valid "next states".
76 *
77 * Since there are complicated ways to arrive at state transitions like proxying
78 * and asynchronous destruction etc, so we monitor the state transitions we are
79 * giving the ss user code to ensure we never deliver illegal state transitions
80 * (because we will assert if we have bugs that do it)
81 */
82
83 const uint32_t ss_state_txn_validity[] = {
84
85 /* if we was last in this state... we can legally go to these states */
86
87 [0] = (1 << LWSSSCS_CREATING) |
88 (1 << LWSSSCS_DESTROYING),
89
90 [LWSSSCS_CREATING] = (1 << LWSSSCS_CONNECTING) |
91 (1 << LWSSSCS_TIMEOUT) |
92 (1 << LWSSSCS_POLL) |
93 (1 << LWSSSCS_SERVER_UPGRADE) |
94 (1 << LWSSSCS_DESTROYING),
95
96 [LWSSSCS_DISCONNECTED] = (1 << LWSSSCS_CONNECTING) |
97 (1 << LWSSSCS_TIMEOUT) |
98 (1 << LWSSSCS_POLL) |
99 (1 << LWSSSCS_DESTROYING),
100
101 [LWSSSCS_UNREACHABLE] = (1 << LWSSSCS_ALL_RETRIES_FAILED) |
102 (1 << LWSSSCS_TIMEOUT) |
103 (1 << LWSSSCS_POLL) |
104 (1 << LWSSSCS_CONNECTING) |
105 /* win conn failure > retry > succ */
106 (1 << LWSSSCS_CONNECTED) |
107 (1 << LWSSSCS_DESTROYING),
108
109 [LWSSSCS_AUTH_FAILED] = (1 << LWSSSCS_ALL_RETRIES_FAILED) |
110 (1 << LWSSSCS_TIMEOUT) |
111 (1 << LWSSSCS_CONNECTING) |
112 (1 << LWSSSCS_DESTROYING),
113
114 [LWSSSCS_CONNECTED] = (1 << LWSSSCS_SERVER_UPGRADE) |
115 (1 << LWSSSCS_SERVER_TXN) |
116 (1 << LWSSSCS_AUTH_FAILED) |
117 (1 << LWSSSCS_QOS_ACK_REMOTE) |
118 (1 << LWSSSCS_QOS_NACK_REMOTE) |
119 (1 << LWSSSCS_QOS_ACK_LOCAL) |
120 (1 << LWSSSCS_QOS_NACK_LOCAL) |
121 (1 << LWSSSCS_DISCONNECTED) |
122 (1 << LWSSSCS_TIMEOUT) |
123 (1 << LWSSSCS_POLL) | /* proxy retry */
124 (1 << LWSSSCS_DESTROYING),
125
126 [LWSSSCS_CONNECTING] = (1 << LWSSSCS_UNREACHABLE) |
127 (1 << LWSSSCS_AUTH_FAILED) |
128 (1 << LWSSSCS_CONNECTING) |
129 (1 << LWSSSCS_CONNECTED) |
130 (1 << LWSSSCS_TIMEOUT) |
131 (1 << LWSSSCS_POLL) |
132 (1 << LWSSSCS_DISCONNECTED) | /* proxy retry */
133 (1 << LWSSSCS_DESTROYING),
134
135 [LWSSSCS_DESTROYING] = 0,
136
137 [LWSSSCS_POLL] = (1 << LWSSSCS_CONNECTING) |
138 (1 << LWSSSCS_TIMEOUT) |
139 (1 << LWSSSCS_DESTROYING),
140
141 [LWSSSCS_ALL_RETRIES_FAILED] = (1 << LWSSSCS_CONNECTING) |
142 (1 << LWSSSCS_TIMEOUT) |
143 (1 << LWSSSCS_DESTROYING),
144
145 [LWSSSCS_QOS_ACK_REMOTE] = (1 << LWSSSCS_DISCONNECTED) |
146 (1 << LWSSSCS_TIMEOUT) |
147 #if defined(LWS_ROLE_MQTT)
148 (1 << LWSSSCS_QOS_ACK_REMOTE) |
149 #endif
150 (1 << LWSSSCS_DESTROYING),
151
152 [LWSSSCS_QOS_NACK_REMOTE] = (1 << LWSSSCS_DISCONNECTED) |
153 (1 << LWSSSCS_TIMEOUT) |
154 (1 << LWSSSCS_DESTROYING),
155
156 [LWSSSCS_QOS_ACK_LOCAL] = (1 << LWSSSCS_DISCONNECTED) |
157 (1 << LWSSSCS_TIMEOUT) |
158 (1 << LWSSSCS_DESTROYING),
159
160 [LWSSSCS_QOS_NACK_LOCAL] = (1 << LWSSSCS_DESTROYING) |
161 (1 << LWSSSCS_TIMEOUT),
162
163 /* he can get the timeout at any point and take no action... */
164 [LWSSSCS_TIMEOUT] = (1 << LWSSSCS_CONNECTING) |
165 (1 << LWSSSCS_CONNECTED) |
166 (1 << LWSSSCS_QOS_ACK_REMOTE) |
167 (1 << LWSSSCS_QOS_NACK_REMOTE) |
168 (1 << LWSSSCS_POLL) |
169 (1 << LWSSSCS_TIMEOUT) |
170 (1 << LWSSSCS_DISCONNECTED) |
171 (1 << LWSSSCS_UNREACHABLE) |
172 (1 << LWSSSCS_DESTROYING),
173
174 [LWSSSCS_SERVER_TXN] = (1 << LWSSSCS_DISCONNECTED) |
175 (1 << LWSSSCS_TIMEOUT) |
176 (1 << LWSSSCS_DESTROYING),
177
178 [LWSSSCS_SERVER_UPGRADE] = (1 << LWSSSCS_SERVER_TXN) |
179 (1 << LWSSSCS_TIMEOUT) |
180 (1 << LWSSSCS_DISCONNECTED) |
181 (1 << LWSSSCS_DESTROYING),
182 };
183
184 #if defined(LWS_WITH_CONMON)
185
186 /*
187 * Convert any conmon data to JSON and attach to the ss handle.
188 */
189
190 lws_ss_state_return_t
lws_conmon_ss_json(lws_ss_handle_t * h)191 lws_conmon_ss_json(lws_ss_handle_t *h)
192 {
193 char ads[48], *end, *buf, *obuf;
194 const struct addrinfo *ai;
195 lws_ss_state_return_t ret = LWSSSSRET_OK;
196 struct lws_conmon cm;
197 size_t len = 500;
198
199 if (!h->policy || !(h->policy->flags & LWSSSPOLF_PERF) || !h->wsi ||
200 h->wsi->perf_done)
201 return LWSSSSRET_OK;
202
203 if (h->conmon_json)
204 lws_free_set_NULL(h->conmon_json);
205
206 h->conmon_json = lws_malloc(len, __func__);
207 if (!h->conmon_json)
208 return LWSSSSRET_OK;
209
210 obuf = buf = h->conmon_json;
211 end = buf + len - 1;
212
213 lws_conmon_wsi_take(h->wsi, &cm);
214
215 lws_sa46_write_numeric_address(&cm.peer46, ads, sizeof(ads));
216 buf += lws_snprintf(buf, lws_ptr_diff_size_t(end, buf),
217 "{\"peer\":\"%s\","
218 "\"dns_us\":%u,"
219 "\"dns_disp\":%u,"
220 "\"sockconn_us\":%u,"
221 "\"tls_us\":%u,"
222 "\"txn_resp_us\":%u,"
223 "\"dns\":[",
224 ads,
225 (unsigned int)cm.ciu_dns,
226 (unsigned int)cm.dns_disposition,
227 (unsigned int)cm.ciu_sockconn,
228 (unsigned int)cm.ciu_tls,
229 (unsigned int)cm.ciu_txn_resp);
230
231 ai = cm.dns_results_copy;
232 while (ai) {
233 lws_sa46_write_numeric_address((lws_sockaddr46 *)ai->ai_addr, ads, sizeof(ads));
234 buf += lws_snprintf(buf, lws_ptr_diff_size_t(end, buf), "\"%s\"", ads);
235 if (ai->ai_next && buf < end - 2)
236 *buf++ = ',';
237 ai = ai->ai_next;
238 }
239
240 buf += lws_snprintf(buf, lws_ptr_diff_size_t(end, buf), "]");
241
242 switch (cm.pcol) {
243 case LWSCONMON_PCOL_HTTP:
244 buf += lws_snprintf(buf, lws_ptr_diff_size_t(end, buf),
245 ",\"prot_specific\":{\"protocol\":\"http\",\"resp\":%u}",
246 (unsigned int)cm.protocol_specific.http.response);
247 break;
248 default:
249 break;
250 }
251
252 buf += lws_snprintf(buf, lws_ptr_diff_size_t(end, buf), "}");
253
254 /*
255 * This destroys the DNS list in the lws_conmon that we took
256 * responsibility for when we used lws_conmon_wsi_take()
257 */
258
259 lws_conmon_release(&cm);
260
261 h->conmon_len = (uint16_t)lws_ptr_diff(buf, obuf);
262
263 #if defined(LWS_WITH_SECURE_STREAMS_PROXY_API)
264 if (h->proxy_onward) {
265
266 /*
267 * ask to forward it on the proxy link
268 */
269
270 ss_proxy_onward_link_req_writeable(h);
271 return LWSSSSRET_OK;
272 }
273 #endif
274
275 /*
276 * We can deliver it directly
277 */
278
279 if (h->info.rx)
280 ret = h->info.rx(ss_to_userobj(h), (uint8_t *)h->conmon_json,
281 (unsigned int)h->conmon_len,
282 (int)(LWSSS_FLAG_SOM | LWSSS_FLAG_EOM |
283 LWSSS_FLAG_PERF_JSON));
284
285 lws_free_set_NULL(h->conmon_json);
286
287 return ret;
288 }
289 #endif
290
291 int
lws_ss_check_next_state(lws_lifecycle_t * lc,uint8_t * prevstate,lws_ss_constate_t cs)292 lws_ss_check_next_state(lws_lifecycle_t *lc, uint8_t *prevstate,
293 lws_ss_constate_t cs)
294 {
295 if (cs >= LWSSSCS_USER_BASE ||
296 cs == LWSSSCS_EVENT_WAIT_CANCELLED ||
297 cs == LWSSSCS_SERVER_TXN ||
298 cs == LWSSSCS_UPSTREAM_LINK_RETRY)
299 /*
300 * we can't judge user or transient states, leave the old state
301 * and just wave them through
302 */
303 return 0;
304
305 if (cs >= LWS_ARRAY_SIZE(ss_state_txn_validity)) {
306 /* we don't recognize this state as usable */
307 lwsl_err("%s: %s: bad new state %u\n", __func__, lc->gutag, cs);
308 assert(0);
309 return 1;
310 }
311
312 if (*prevstate >= LWS_ARRAY_SIZE(ss_state_txn_validity)) {
313 /* existing state is broken */
314 lwsl_err("%s: %s: bad existing state %u\n", __func__,
315 lc->gutag, (unsigned int)*prevstate);
316 assert(0);
317 return 1;
318 }
319
320 if (ss_state_txn_validity[*prevstate] & (1u << cs)) {
321
322 lwsl_notice("%s: %s: %s -> %s\n", __func__, lc->gutag,
323 lws_ss_state_name((int)*prevstate),
324 lws_ss_state_name((int)cs));
325
326 /* this is explicitly allowed, update old state to new */
327 *prevstate = (uint8_t)cs;
328
329 return 0;
330 }
331
332 lwsl_err("%s: %s: transition from %s -> %s is illegal\n", __func__,
333 lc->gutag, lws_ss_state_name((int)*prevstate),
334 lws_ss_state_name((int)cs));
335
336 assert(0);
337
338 return 1;
339 }
340
341 int
lws_ss_check_next_state_ss(lws_ss_handle_t * ss,uint8_t * prevstate,lws_ss_constate_t cs)342 lws_ss_check_next_state_ss(lws_ss_handle_t *ss, uint8_t *prevstate,
343 lws_ss_constate_t cs)
344 {
345 if (cs >= LWSSSCS_USER_BASE ||
346 cs == LWSSSCS_EVENT_WAIT_CANCELLED ||
347 cs == LWSSSCS_UPSTREAM_LINK_RETRY)
348 /*
349 * we can't judge user or transient states, leave the old state
350 * and just wave them through
351 */
352 return 0;
353
354 if (cs >= LWS_ARRAY_SIZE(ss_state_txn_validity)) {
355 /* we don't recognize this state as usable */
356 lwsl_ss_err(ss, "bad new state %u", cs);
357 assert(0);
358 return 1;
359 }
360
361 if (*prevstate >= LWS_ARRAY_SIZE(ss_state_txn_validity)) {
362 /* existing state is broken */
363 lwsl_ss_err(ss, "bad existing state %u",
364 (unsigned int)*prevstate);
365 assert(0);
366 return 1;
367 }
368
369 if (ss_state_txn_validity[*prevstate] & (1u << cs)) {
370
371 lwsl_ss_notice(ss, "%s -> %s",
372 lws_ss_state_name((int)*prevstate),
373 lws_ss_state_name((int)cs));
374
375 /* this is explicitly allowed, update old state to new */
376 *prevstate = (uint8_t)cs;
377
378 return 0;
379 }
380
381 lwsl_ss_err(ss, "transition from %s -> %s is illegal",
382 lws_ss_state_name((int)*prevstate),
383 lws_ss_state_name((int)cs));
384
385 assert(0);
386
387 return 1;
388 }
389
390 const char *
lws_ss_state_name(int state)391 lws_ss_state_name(int state)
392 {
393 if (state >= LWSSSCS_USER_BASE)
394 return "user state";
395
396 if (state >= (int)LWS_ARRAY_SIZE(state_names))
397 return "unknown";
398
399 return state_names[state];
400 }
401
402 lws_ss_state_return_t
lws_ss_event_helper(lws_ss_handle_t * h,lws_ss_constate_t cs)403 lws_ss_event_helper(lws_ss_handle_t *h, lws_ss_constate_t cs)
404 {
405 lws_ss_state_return_t r;
406
407 if (!h)
408 return LWSSSSRET_OK;
409
410 if (lws_ss_check_next_state_ss(h, &h->prev_ss_state, cs))
411 return LWSSSSRET_DESTROY_ME;
412
413 if (cs == LWSSSCS_CONNECTED)
414 h->ss_dangling_connected = 1;
415 if (cs == LWSSSCS_DISCONNECTED)
416 h->ss_dangling_connected = 0;
417
418 #if defined(LWS_WITH_SEQUENCER)
419 /*
420 * A parent sequencer for the ss is optional, if we have one, keep it
421 * informed of state changes on the ss connection
422 */
423 if (h->seq && cs != LWSSSCS_DESTROYING)
424 lws_seq_queue_event(h->seq, LWSSEQ_SS_STATE_BASE + cs,
425 (void *)h, NULL);
426 #endif
427
428 if (h->info.state) {
429 h->h_in_svc = h;
430 r = h->info.state(ss_to_userobj(h), NULL, cs,
431 cs == LWSSSCS_UNREACHABLE &&
432 h->wsi && h->wsi->dns_reachability);
433 h->h_in_svc = NULL;
434 #if defined(LWS_WITH_SERVER)
435 if ((h->info.flags & LWSSSINFLAGS_ACCEPTED) &&
436 cs == LWSSSCS_DISCONNECTED)
437 r = LWSSSSRET_DESTROY_ME;
438 #endif
439 return r;
440 }
441
442 return LWSSSSRET_OK;
443 }
444
445 int
_lws_ss_handle_state_ret_CAN_DESTROY_HANDLE(lws_ss_state_return_t r,struct lws * wsi,lws_ss_handle_t ** ph)446 _lws_ss_handle_state_ret_CAN_DESTROY_HANDLE(lws_ss_state_return_t r, struct lws *wsi,
447 lws_ss_handle_t **ph)
448 {
449 if (r == LWSSSSRET_DESTROY_ME) {
450 lwsl_info("%s: DESTROY ME: %s, %s\n", __func__,
451 lws_wsi_tag(wsi), lws_ss_tag(*ph));
452 if (wsi) {
453 lws_set_opaque_user_data(wsi, NULL);
454 lws_set_timeout(wsi, 1, LWS_TO_KILL_ASYNC);
455 } else {
456 if ((*ph)->wsi) {
457 lws_set_opaque_user_data((*ph)->wsi, NULL);
458 lws_set_timeout((*ph)->wsi, 1, LWS_TO_KILL_ASYNC);
459 }
460 }
461 (*ph)->wsi = NULL;
462 lws_ss_destroy(ph);
463 }
464
465 return -1; /* close connection */
466 }
467
468 static void
lws_ss_timeout_sul_check_cb(lws_sorted_usec_list_t * sul)469 lws_ss_timeout_sul_check_cb(lws_sorted_usec_list_t *sul)
470 {
471 lws_ss_state_return_t r;
472 lws_ss_handle_t *h = lws_container_of(sul, lws_ss_handle_t, sul);
473
474 lwsl_info("%s: retrying %s after backoff\n", __func__, lws_ss_tag(h));
475 /* we want to retry... */
476 h->seqstate = SSSEQ_DO_RETRY;
477
478 r = _lws_ss_request_tx(h);
479 _lws_ss_handle_state_ret_CAN_DESTROY_HANDLE(r, NULL, &h);
480 }
481
482 int
lws_ss_exp_cb_metadata(void * priv,const char * name,char * out,size_t * pos,size_t olen,size_t * exp_ofs)483 lws_ss_exp_cb_metadata(void *priv, const char *name, char *out, size_t *pos,
484 size_t olen, size_t *exp_ofs)
485 {
486 lws_ss_handle_t *h = (lws_ss_handle_t *)priv;
487 const char *replace = NULL;
488 size_t total, budget;
489 lws_ss_metadata_t *md = lws_ss_policy_metadata(h->policy, name),
490 *hmd = lws_ss_get_handle_metadata(h, name);
491
492 if (!md) {
493 lwsl_err("%s: Unknown metadata %s\n", __func__, name);
494
495 return LSTRX_FATAL_NAME_UNKNOWN;
496 }
497
498 if (!hmd)
499 return LSTRX_FILLED_OUT;
500
501 replace = hmd->value__may_own_heap;
502
503 if (!replace)
504 return LSTRX_DONE;
505
506 total = hmd->length;
507
508 budget = olen - *pos;
509 total -= *exp_ofs;
510 if (total < budget)
511 budget = total;
512
513 if (out)
514 memcpy(out + *pos, replace + (*exp_ofs), budget);
515 *exp_ofs += budget;
516 *pos += budget;
517
518 if (budget == total)
519 return LSTRX_DONE;
520
521 return LSTRX_FILLED_OUT;
522 }
523
524 int
lws_ss_set_timeout_us(lws_ss_handle_t * h,lws_usec_t us)525 lws_ss_set_timeout_us(lws_ss_handle_t *h, lws_usec_t us)
526 {
527 struct lws_context_per_thread *pt = &h->context->pt[h->tsi];
528
529 h->sul.cb = lws_ss_timeout_sul_check_cb;
530 __lws_sul_insert_us(&pt->pt_sul_owner[
531 !!(h->policy->flags & LWSSSPOLF_WAKE_SUSPEND__VALIDITY)],
532 &h->sul, us);
533
534 return 0;
535 }
536
537 lws_ss_state_return_t
_lws_ss_backoff(lws_ss_handle_t * h,lws_usec_t us_override)538 _lws_ss_backoff(lws_ss_handle_t *h, lws_usec_t us_override)
539 {
540 uint64_t ms;
541 char conceal;
542
543 lws_service_assert_loop_thread(h->context, h->tsi);
544
545 if (h->seqstate == SSSEQ_RECONNECT_WAIT)
546 return LWSSSSRET_OK;
547
548 /* figure out what we should do about another retry */
549
550 lwsl_info("%s: %s: retry backoff after failure\n", __func__, lws_ss_tag(h));
551 ms = lws_retry_get_delay_ms(h->context, h->policy->retry_bo,
552 &h->retry, &conceal);
553 if (!conceal) {
554 lwsl_info("%s: %s: abandon conn attempt \n",__func__, lws_ss_tag(h));
555
556 if (h->seqstate == SSSEQ_IDLE) /* been here? */
557 return LWSSSSRET_OK;
558
559 h->seqstate = SSSEQ_IDLE;
560
561 return lws_ss_event_helper(h, LWSSSCS_ALL_RETRIES_FAILED);
562 }
563
564 /* Only increase our planned backoff, or go with it */
565
566 if (us_override < (lws_usec_t)ms * LWS_US_PER_MS)
567 us_override = (lws_usec_t)(ms * LWS_US_PER_MS);
568
569 h->seqstate = SSSEQ_RECONNECT_WAIT;
570 lws_ss_set_timeout_us(h, us_override);
571
572 lwsl_info("%s: %s: retry wait %dms\n", __func__, lws_ss_tag(h),
573 (int)(us_override / 1000));
574
575 return LWSSSSRET_OK;
576 }
577
578 lws_ss_state_return_t
lws_ss_backoff(lws_ss_handle_t * h)579 lws_ss_backoff(lws_ss_handle_t *h)
580 {
581 return _lws_ss_backoff(h, 0);
582 }
583
584 #if defined(LWS_WITH_SYS_SMD)
585
586 /*
587 * Local SMD <-> SS
588 *
589 * We pass received messages through to the SS handler synchronously, using the
590 * lws service thread context.
591 *
592 * After the SS is created and registered, still nothing is going to come here
593 * until the peer sends us his rx_class_mask and we update his registration with
594 * it, because from SS creation his rx_class_mask defaults to 0.
595 */
596
597 static int
lws_smd_ss_cb(void * opaque,lws_smd_class_t _class,lws_usec_t timestamp,void * buf,size_t len)598 lws_smd_ss_cb(void *opaque, lws_smd_class_t _class,
599 lws_usec_t timestamp, void *buf, size_t len)
600 {
601 lws_ss_handle_t *h = (lws_ss_handle_t *)opaque;
602 uint8_t *p = (uint8_t *)buf - LWS_SMD_SS_RX_HEADER_LEN;
603
604 lws_service_assert_loop_thread(h->context, h->tsi);
605
606 /*
607 * When configured with SS enabled, lws over-allocates
608 * LWS_SMD_SS_RX_HEADER_LEN bytes behind the payload of the queued
609 * message, for prepending serialized class and timestamp data in-band
610 * with the payload.
611 */
612
613 lws_ser_wu64be(p, _class);
614 lws_ser_wu64be(p + 8, (uint64_t)timestamp);
615
616 if (h->info.rx)
617 h->info.rx((void *)&h[1], p, len + LWS_SMD_SS_RX_HEADER_LEN,
618 LWSSS_FLAG_SOM | LWSSS_FLAG_EOM);
619
620 return 0;
621 }
622
623 static void
lws_ss_smd_tx_cb(lws_sorted_usec_list_t * sul)624 lws_ss_smd_tx_cb(lws_sorted_usec_list_t *sul)
625 {
626 lws_ss_handle_t *h = lws_container_of(sul, lws_ss_handle_t, u.smd.sul_write);
627 uint8_t buf[LWS_SMD_SS_RX_HEADER_LEN + LWS_SMD_MAX_PAYLOAD], *p;
628 size_t len = sizeof(buf);
629 lws_smd_class_t _class;
630 int flags = 0, n;
631
632 lws_service_assert_loop_thread(h->context, h->tsi);
633
634 if (!h->info.tx)
635 return;
636
637 n = h->info.tx(&h[1], h->txord++, buf, &len, &flags);
638 if (n)
639 /* nonzero return means don't want to send anything */
640 return;
641
642 // lwsl_notice("%s: (SS %p bound to _lws_smd creates message) tx len %d\n", __func__, h, (int)len);
643 // lwsl_hexdump_notice(buf, len);
644
645 assert(len >= LWS_SMD_SS_RX_HEADER_LEN);
646 _class = (lws_smd_class_t)lws_ser_ru64be(buf);
647 p = lws_smd_msg_alloc(h->context, _class, len - LWS_SMD_SS_RX_HEADER_LEN);
648 if (!p) {
649 // this can be rejected if nobody listening for this class
650 //lwsl_notice("%s: failed to alloc\n", __func__);
651 return;
652 }
653
654 memcpy(p, buf + LWS_SMD_SS_RX_HEADER_LEN, len - LWS_SMD_SS_RX_HEADER_LEN);
655 if (lws_smd_msg_send(h->context, p)) {
656 lwsl_notice("%s: failed to queue\n", __func__);
657 return;
658 }
659 }
660
661 #endif
662
663 lws_ss_state_return_t
_lws_ss_client_connect(lws_ss_handle_t * h,int is_retry,void * conn_if_sspc_onw)664 _lws_ss_client_connect(lws_ss_handle_t *h, int is_retry, void *conn_if_sspc_onw)
665 {
666 const char *prot, *_prot, *ipath, *_ipath, *ads, *_ads;
667 struct lws_client_connect_info i;
668 const struct ss_pcols *ssp;
669 size_t used_in, used_out;
670 union lws_ss_contemp ct;
671 lws_ss_state_return_t r;
672 int port, _port, tls;
673 char *path, ep[96];
674 lws_strexp_t exp;
675 struct lws *wsi;
676
677 lws_service_assert_loop_thread(h->context, h->tsi);
678
679 if (!h->policy) {
680 lwsl_err("%s: ss with no policy\n", __func__);
681
682 return LWSSSSRET_OK;
683 }
684
685 /*
686 * We are already bound to a sink?
687 */
688
689 // if (h->h_sink)
690 // return 0;
691
692 if (!is_retry)
693 h->retry = 0;
694
695 #if defined(LWS_WITH_SYS_SMD)
696 if (h->policy == &pol_smd) {
697
698 if (h->u.smd.smd_peer)
699 return LWSSSSRET_OK;
700
701 // lwsl_notice("%s: received connect for _lws_smd, registering for class mask 0x%x\n",
702 // __func__, h->info.manual_initial_tx_credit);
703
704 h->u.smd.smd_peer = lws_smd_register(h->context, h,
705 (h->info.flags & LWSSSINFLAGS_PROXIED) ?
706 LWSSMDREG_FLAG_PROXIED_SS : 0,
707 (lws_smd_class_t)h->info.manual_initial_tx_credit,
708 lws_smd_ss_cb);
709 if (!h->u.smd.smd_peer)
710 return LWSSSSRET_TX_DONT_SEND;
711
712 if (lws_ss_event_helper(h, LWSSSCS_CONNECTING))
713 return LWSSSSRET_TX_DONT_SEND;
714
715 if (lws_ss_event_helper(h, LWSSSCS_CONNECTED))
716 return LWSSSSRET_TX_DONT_SEND;
717 return LWSSSSRET_OK;
718 }
719 #endif
720
721 /*
722 * We're going to substitute ${metadata} in the endpoint at connection-
723 * time, so this can be set dynamically...
724 */
725
726 lws_strexp_init(&exp, (void *)h, lws_ss_exp_cb_metadata, ep, sizeof(ep));
727
728 if (lws_strexp_expand(&exp, h->policy->endpoint,
729 strlen(h->policy->endpoint),
730 &used_in, &used_out) != LSTRX_DONE) {
731 lwsl_err("%s: address strexp failed\n", __func__);
732
733 return LWSSSSRET_TX_DONT_SEND;
734 }
735
736 /*
737 * ... in some cases, we might want the user to be able to override
738 * some policy settings by what he provided in there. For example,
739 * if he set the endpoint to "https://myendpoint.com:4443/mypath" it
740 * might be quite convenient to override the policy to follow the info
741 * that was given for at least server, port and the url path.
742 */
743
744 _port = port = h->policy->port;
745 _prot = prot = NULL;
746 _ipath = ipath = "";
747 _ads = ads = ep;
748
749 if (strchr(ep, ':') &&
750 !lws_parse_uri(ep, &_prot, &_ads, &_port, &_ipath)) {
751 lwsl_debug("%s: using uri parse results '%s' '%s' %d '%s'\n",
752 __func__, _prot, _ads, _port, _ipath);
753 prot = _prot;
754 ads = _ads;
755 port = _port;
756 ipath = _ipath;
757 }
758
759 memset(&i, 0, sizeof i); /* otherwise uninitialized garbage */
760 i.context = h->context;
761 tls = !!(h->policy->flags & LWSSSPOLF_TLS);
762
763 if (prot && (!strcmp(prot, "http") || !strcmp(prot, "ws") ||
764 !strcmp(prot, "mqtt")))
765 tls = 0;
766
767 if (tls) {
768 lwsl_info("%s: using tls\n", __func__);
769 i.ssl_connection = LCCSCF_USE_SSL;
770
771 if (!h->policy->trust.store)
772 lwsl_info("%s: using platform trust store\n", __func__);
773 else {
774
775 i.vhost = lws_get_vhost_by_name(h->context,
776 h->policy->trust.store->name);
777 if (!i.vhost) {
778 lwsl_err("%s: missing vh for policy %s\n",
779 __func__,
780 h->policy->trust.store->name);
781
782 return -1;
783 }
784 }
785 }
786
787 if (h->policy->flags & LWSSSPOLF_WAKE_SUSPEND__VALIDITY)
788 i.ssl_connection |= LCCSCF_WAKE_SUSPEND__VALIDITY;
789
790 /* translate policy attributes to IP ToS flags */
791
792 if (h->policy->flags & LWSSSPOLF_ATTR_LOW_LATENCY)
793 i.ssl_connection |= LCCSCF_IP_LOW_LATENCY;
794 if (h->policy->flags & LWSSSPOLF_ATTR_HIGH_THROUGHPUT)
795 i.ssl_connection |= LCCSCF_IP_HIGH_THROUGHPUT;
796 if (h->policy->flags & LWSSSPOLF_ATTR_HIGH_RELIABILITY)
797 i.ssl_connection |= LCCSCF_IP_HIGH_RELIABILITY;
798 if (h->policy->flags & LWSSSPOLF_ATTR_LOW_COST)
799 i.ssl_connection |= LCCSCF_IP_LOW_COST;
800 if (h->policy->flags & LWSSSPOLF_PERF) /* collect conmon stats on this */
801 i.ssl_connection |= LCCSCF_CONMON;
802
803 /* mark the connection with the streamtype priority from the policy */
804
805 i.priority = h->policy->priority;
806
807 i.ssl_connection |= LCCSCF_SECSTREAM_CLIENT;
808
809 if (conn_if_sspc_onw) {
810 i.ssl_connection |= LCCSCF_SECSTREAM_PROXY_ONWARD;
811 h->conn_if_sspc_onw = conn_if_sspc_onw;
812 }
813
814
815 i.address = ads;
816 i.port = port;
817 i.host = i.address;
818 i.origin = i.address;
819 i.opaque_user_data = h;
820 i.seq = h->seq;
821 i.retry_and_idle_policy = h->policy->retry_bo;
822 i.sys_tls_client_cert = h->policy->client_cert;
823
824 i.path = ipath;
825 /* if this is not "", munge should use it instead of policy
826 * url path
827 */
828
829 ssp = ss_pcols[(int)h->policy->protocol];
830 if (!ssp) {
831 lwsl_err("%s: unsupported protocol\n", __func__);
832
833 return LWSSSSRET_TX_DONT_SEND;
834 }
835 i.alpn = ssp->alpn;
836
837 /*
838 * For http, we can get the method from the http object, override in
839 * the protocol-specific munge callback below if not http
840 */
841 i.method = h->policy->u.http.method;
842 i.protocol = ssp->protocol->name; /* lws protocol name */
843 i.local_protocol_name = i.protocol;
844
845 path = lws_malloc(h->context->max_http_header_data, __func__);
846 if (!path) {
847 lwsl_warn("%s: OOM on path prealloc\n", __func__);
848 return LWSSSSRET_TX_DONT_SEND;
849 }
850
851 if (ssp->munge) /* eg, raw doesn't use; endpoint strexp already done */
852 ssp->munge(h, path, h->context->max_http_header_data, &i, &ct);
853
854 i.pwsi = &h->wsi;
855
856 #if defined(LWS_WITH_SSPLUGINS)
857 if (h->policy->plugins[0] && h->policy->plugins[0]->munge)
858 h->policy->plugins[0]->munge(h, path, h->context->max_http_header_data);
859 #endif
860
861 lwsl_info("%s: connecting %s, '%s' '%s' %s\n", __func__, i.method,
862 i.alpn, i.address, i.path);
863
864 #if defined(LWS_WITH_SYS_METRICS)
865 /* possibly already hanging connect retry... */
866 if (!h->cal_txn.mt)
867 lws_metrics_caliper_bind(h->cal_txn, h->context->mth_ss_conn);
868
869 if (h->policy->streamtype)
870 lws_metrics_tag_add(&h->cal_txn.mtags_owner, "ss",
871 h->policy->streamtype);
872 #endif
873
874 h->txn_ok = 0;
875 r = lws_ss_event_helper(h, LWSSSCS_CONNECTING);
876 if (r) {
877 lws_free(path);
878 return r;
879 }
880
881 h->inside_connect = 1;
882 h->pending_ret = LWSSSSRET_OK;
883 wsi = lws_client_connect_via_info(&i);
884 h->inside_connect = 0;
885 lws_free(path);
886 if (!wsi) {
887 /*
888 * We already found that we could not connect, without even
889 * having to go around the event loop
890 */
891
892 if (h->pending_ret)
893 return h->pending_ret;
894
895 if (h->prev_ss_state != LWSSSCS_UNREACHABLE &&
896 h->prev_ss_state != LWSSSCS_ALL_RETRIES_FAILED) {
897 /*
898 * blocking DNS failure can get to unreachable via
899 * CCE, and unreachable can get to ALL_RETRIES_FAILED
900 */
901 r = lws_ss_event_helper(h, LWSSSCS_UNREACHABLE);
902 if (r)
903 return r;
904
905 r = lws_ss_backoff(h);
906 if (r)
907 return r;
908 }
909
910 return LWSSSSRET_TX_DONT_SEND;
911 }
912
913 return LWSSSSRET_OK;
914 }
915
916 lws_ss_state_return_t
lws_ss_client_connect(lws_ss_handle_t * h)917 lws_ss_client_connect(lws_ss_handle_t *h)
918 {
919 lws_ss_state_return_t r;
920
921 lws_service_assert_loop_thread(h->context, h->tsi);
922
923 r = _lws_ss_client_connect(h, 0, 0);
924
925 return r;
926 }
927
928 /*
929 * Public API
930 */
931
932 /*
933 * Create either a stream or a sink
934 */
935
936 int
lws_ss_create(struct lws_context * context,int tsi,const lws_ss_info_t * ssi,void * opaque_user_data,lws_ss_handle_t ** ppss,struct lws_sequencer * seq_owner,const char ** ppayload_fmt)937 lws_ss_create(struct lws_context *context, int tsi, const lws_ss_info_t *ssi,
938 void *opaque_user_data, lws_ss_handle_t **ppss,
939 struct lws_sequencer *seq_owner, const char **ppayload_fmt)
940 {
941 struct lws_context_per_thread *pt = &context->pt[tsi];
942 const lws_ss_policy_t *pol;
943 lws_ss_state_return_t r;
944 lws_ss_metadata_t *smd;
945 lws_ss_handle_t *h;
946 size_t size;
947 void **v;
948 char *p;
949 int n;
950
951 lws_service_assert_loop_thread(context, tsi);
952
953 #if defined(LWS_WITH_SECURE_STREAMS_CPP)
954 pol = ssi->policy;
955 if (!pol) {
956 #endif
957
958 #if defined(LWS_WITH_SYS_FAULT_INJECTION)
959 lws_fi_ctx_t temp_fic;
960
961 /*
962 * We have to do a temp inherit from context to find out
963 * early if we are supposed to inject a fault concealing
964 * the policy
965 */
966
967 memset(&temp_fic, 0, sizeof(temp_fic));
968 lws_xos_init(&temp_fic.xos, lws_xos(&context->fic.xos));
969 lws_fi_inherit_copy(&temp_fic, &context->fic, "ss", ssi->streamtype);
970
971 if (lws_fi(&temp_fic, "ss_no_streamtype_policy"))
972 pol = NULL;
973 else
974 pol = lws_ss_policy_lookup(context, ssi->streamtype);
975
976 lws_fi_destroy(&temp_fic);
977 #else
978 pol = lws_ss_policy_lookup(context, ssi->streamtype);
979 #endif
980 if (!pol) {
981 lwsl_cx_info(context, "unknown stream type %s",
982 ssi->streamtype);
983 return 1;
984 }
985 #if defined(LWS_WITH_SECURE_STREAMS_CPP)
986 }
987 #endif
988
989 #if 0
990 if (ssi->flags & LWSSSINFLAGS_REGISTER_SINK) {
991 /*
992 * This can register a secure streams sink as well as normal
993 * secure streams connections. If that's what's happening,
994 * confirm the policy agrees that this streamtype should be
995 * directed to a sink.
996 */
997 if (!(pol->flags & LWSSSPOLF_LOCAL_SINK)) {
998 /*
999 * Caller wanted to create a sink for this streamtype,
1000 * but the policy does not agree the streamtype should
1001 * be routed to a local sink.
1002 */
1003 lwsl_err("%s: %s policy does not allow local sink\n",
1004 __func__, ssi->streamtype);
1005
1006 return 1;
1007 }
1008 } else {
1009
1010 if (!(pol->flags & LWSSSPOLF_LOCAL_SINK)) {
1011
1012 }
1013 // lws_dll2_foreach_safe(&pt->ss_owner, NULL, lws_ss_destroy_dll);
1014 }
1015 #endif
1016
1017 /*
1018 * We overallocate and point to things in the overallocation...
1019 *
1020 * 1) the user_alloc from the stream info
1021 * 2) network auth plugin instantiation data
1022 * 3) stream auth plugin instantiation data
1023 * 4) as many metadata pointer structs as the policy tells
1024 * 5) the streamtype name (length is not aligned)
1025 *
1026 * ... when we come to destroy it, just one free to do.
1027 */
1028
1029 size = sizeof(*h) + ssi->user_alloc +
1030 (ssi->streamtype ? strlen(ssi->streamtype): 0) + 1;
1031 #if defined(LWS_WITH_SSPLUGINS)
1032 if (pol->plugins[0])
1033 size += pol->plugins[0]->alloc;
1034 if (pol->plugins[1])
1035 size += pol->plugins[1]->alloc;
1036 #endif
1037 size += pol->metadata_count * sizeof(lws_ss_metadata_t);
1038
1039 h = lws_zalloc(size, __func__);
1040 if (!h)
1041 return 2;
1042
1043 h->lc.log_cx = context->log_cx;
1044
1045 if (ssi->sss_protocol_version)
1046 __lws_lc_tag(context, &context->lcg[LWSLCG_WSI_SS_CLIENT],
1047 &h->lc, "%s|v%u|%u",
1048 ssi->streamtype ? ssi->streamtype : "nostreamtype",
1049 (unsigned int)ssi->sss_protocol_version,
1050 (unsigned int)ssi->client_pid);
1051 else
1052 __lws_lc_tag(context, &context->lcg[LWSLCG_WSI_SS_CLIENT],
1053 &h->lc, "%s",
1054 ssi->streamtype ? ssi->streamtype : "nostreamtype");
1055
1056 #if defined(LWS_WITH_SYS_FAULT_INJECTION)
1057 h->fic.name = "ss";
1058 lws_xos_init(&h->fic.xos, lws_xos(&context->fic.xos));
1059 if (ssi->fic.fi_owner.count)
1060 lws_fi_import(&h->fic, &ssi->fic);
1061
1062 lws_fi_inherit_copy(&h->fic, &context->fic, "ss", ssi->streamtype);
1063 #endif
1064
1065 h->info = *ssi;
1066 h->policy = pol;
1067 h->context = context;
1068 h->tsi = (uint8_t)tsi;
1069 h->seq = seq_owner;
1070
1071 if (h->info.flags & LWSSSINFLAGS_PROXIED)
1072 h->proxy_onward = 1;
1073
1074 /* start of overallocated area */
1075 p = (char *)&h[1];
1076
1077 /* set the handle pointer in the user data struct */
1078 v = (void **)(p + ssi->handle_offset);
1079 *v = h;
1080
1081 /* set the opaque user data in the user data struct */
1082 v = (void **)(p + ssi->opaque_user_data_offset);
1083 *v = opaque_user_data;
1084
1085 p += ssi->user_alloc;
1086
1087 #if defined(LWS_WITH_SSPLUGINS)
1088 if (pol->plugins[0]) {
1089 h->nauthi = p;
1090 p += pol->plugins[0]->alloc;
1091 }
1092 if (pol->plugins[1]) {
1093 h->sauthi = p;
1094 p += pol->plugins[1]->alloc;
1095 }
1096 #endif
1097
1098 if (pol->metadata_count) {
1099 h->metadata = (lws_ss_metadata_t *)p;
1100 p += pol->metadata_count * sizeof(lws_ss_metadata_t);
1101
1102 lwsl_cx_info(context, "%s metadata count %d",
1103 pol->streamtype, pol->metadata_count);
1104 }
1105
1106 smd = pol->metadata;
1107 for (n = 0; n < pol->metadata_count; n++) {
1108 h->metadata[n].name = smd->name;
1109 if (n + 1 == pol->metadata_count)
1110 h->metadata[n].next = NULL;
1111 else
1112 h->metadata[n].next = &h->metadata[n + 1];
1113 smd = smd->next;
1114 }
1115
1116 if (ssi->streamtype)
1117 memcpy(p, ssi->streamtype, strlen(ssi->streamtype) + 1);
1118 /* don't mark accepted ss as being the server */
1119 if (ssi->flags & LWSSSINFLAGS_SERVER)
1120 h->info.flags &= (uint8_t)~LWSSSINFLAGS_SERVER;
1121 h->info.streamtype = p;
1122
1123 lws_pt_lock(pt, __func__);
1124 lws_dll2_add_head(&h->list, &pt->ss_owner);
1125 lws_pt_unlock(pt);
1126
1127 if (ppss)
1128 *ppss = h;
1129
1130 if (ppayload_fmt)
1131 *ppayload_fmt = pol->payload_fmt;
1132
1133 if (ssi->flags & LWSSSINFLAGS_SERVER)
1134 /*
1135 * return early for accepted connection flow
1136 */
1137 return 0;
1138
1139 #if defined(LWS_WITH_SYS_SMD)
1140 /*
1141 * For a local Secure Streams connection
1142 */
1143 if (!(ssi->flags & LWSSSINFLAGS_PROXIED) &&
1144 pol == &pol_smd) {
1145
1146 /*
1147 * So he has asked to be wired up to SMD over a SS link.
1148 * Register him as an smd participant in his own right.
1149 *
1150 * Just for this case, ssi->manual_initial_tx_credit is used
1151 * to set the rx class mask (this is part of the SS serialization
1152 * format as well)
1153 */
1154 h->u.smd.smd_peer = lws_smd_register(context, h, 0,
1155 (lws_smd_class_t)ssi->manual_initial_tx_credit,
1156 lws_smd_ss_cb);
1157 if (!h->u.smd.smd_peer || lws_fi(&h->fic, "ss_create_smd"))
1158 goto fail_creation;
1159 lwsl_cx_info(context, "registered SS SMD");
1160 }
1161 #endif
1162
1163 #if defined(LWS_WITH_SERVER)
1164 if (h->policy->flags & LWSSSPOLF_SERVER) {
1165 const struct lws_protocols *pprot[3], **ppp = &pprot[0];
1166 struct lws_context_creation_info i;
1167 struct lws_vhost *vho = NULL;
1168
1169 lwsl_cx_info(context, "creating server");
1170
1171 if (h->policy->endpoint &&
1172 h->policy->endpoint[0] == '!') {
1173 /*
1174 * There's already a vhost existing that we want to
1175 * bind to, we don't have to specify and create one.
1176 *
1177 * The vhost must enable any protocols that we want.
1178 */
1179
1180 vho = lws_get_vhost_by_name(context,
1181 &h->policy->endpoint[1]);
1182 if (!vho || lws_fi(&h->fic, "ss_create_vhost")) {
1183 lwsl_err("%s: no vhost %s\n", __func__,
1184 &h->policy->endpoint[1]);
1185 goto fail_creation;
1186 }
1187
1188 goto extant;
1189 }
1190
1191 /*
1192 * This streamtype represents a server, we're being asked to
1193 * instantiate a corresponding vhost for it
1194 */
1195
1196 memset(&i, 0, sizeof i);
1197
1198 i.iface = h->policy->endpoint;
1199 i.vhost_name = h->policy->streamtype;
1200 i.port = h->policy->port;
1201
1202 if (i.iface && i.iface[0] == '+') {
1203 i.iface++;
1204 i.options |= LWS_SERVER_OPTION_UNIX_SOCK;
1205 }
1206
1207 if (!ss_pcols[h->policy->protocol] ||
1208 lws_fi(&h->fic, "ss_create_pcol")) {
1209 lwsl_err("%s: unsupp protocol", __func__);
1210 goto fail_creation;
1211 }
1212
1213 *ppp++ = ss_pcols[h->policy->protocol]->protocol;
1214 #if defined(LWS_ROLE_WS)
1215 if (h->policy->u.http.u.ws.subprotocol)
1216 /*
1217 * He names a ws subprotocol, ie, we want to support
1218 * ss-ws protocol in this vhost
1219 */
1220 *ppp++ = &protocol_secstream_ws;
1221 #endif
1222 *ppp = NULL;
1223 i.pprotocols = pprot;
1224
1225 #if defined(LWS_WITH_TLS)
1226 if (h->policy->flags & LWSSSPOLF_TLS) {
1227 i.options |= LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT;
1228 i.server_ssl_cert_mem =
1229 h->policy->trust.server.cert->ca_der;
1230 i.server_ssl_cert_mem_len = (unsigned int)
1231 h->policy->trust.server.cert->ca_der_len;
1232 i.server_ssl_private_key_mem =
1233 h->policy->trust.server.key->ca_der;
1234 i.server_ssl_private_key_mem_len = (unsigned int)
1235 h->policy->trust.server.key->ca_der_len;
1236 }
1237 #endif
1238
1239 if (!lws_fi(&h->fic, "ss_srv_vh_fail"))
1240 vho = lws_create_vhost(context, &i);
1241 else
1242 vho = NULL;
1243 if (!vho) {
1244 lwsl_cx_err(context, "failed to create vh");
1245 goto fail_creation;
1246 }
1247
1248 extant:
1249
1250 /*
1251 * Mark this vhost as having to apply ss server semantics to
1252 * any incoming accepted connection
1253 */
1254 vho->ss_handle = h;
1255
1256 r = lws_ss_event_helper(h, LWSSSCS_CREATING);
1257 lwsl_cx_info(context, "CREATING returned status %d", (int)r);
1258 if (r == LWSSSSRET_DESTROY_ME ||
1259 lws_fi(&h->fic, "ss_create_destroy_me"))
1260 goto fail_creation;
1261
1262 lwsl_cx_notice(context, "created server %s",
1263 h->policy->streamtype);
1264
1265 return 0;
1266 }
1267 #endif
1268
1269 #if defined(LWS_WITH_SECURE_STREAMS_STATIC_POLICY_ONLY)
1270
1271 /*
1272 * For static policy case, dynamically ref / instantiate the related
1273 * trust store and vhost. We do it by logical ss rather than connection
1274 * because we don't want to expose the latency of creating the x.509
1275 * trust store at the first connection.
1276 *
1277 * But it might be given the tls linkup takes time anyway, it can move
1278 * to the ss connect code instead.
1279 */
1280
1281 if (!lws_ss_policy_ref_trust_store(context, h->policy, 1 /* do the ref */) ||
1282 lws_fi(&h->fic, "ss_create_no_ts")) {
1283 lwsl_err("%s: unable to get vhost / trust store\n", __func__);
1284 goto fail_creation;
1285 }
1286 #else
1287 #if defined(LWS_WITH_SECURE_STREAMS_CPP)
1288 if (!ssi->streamtype &&
1289 !lws_ss_policy_ref_trust_store(context, h->policy, 1 /* do the ref */)) {
1290 lwsl_err("%s: unable to get vhost / trust store\n", __func__);
1291 goto fail_creation;
1292 }
1293 #endif
1294 #endif
1295
1296 r = lws_ss_event_helper(h, LWSSSCS_CREATING);
1297 lwsl_ss_info(h, "CREATING returned status %d", (int)r);
1298 if (r == LWSSSSRET_DESTROY_ME ||
1299 lws_fi(&h->fic, "ss_create_destroy_me"))
1300 goto fail_creation;
1301
1302 #if defined(LWS_WITH_SYS_SMD)
1303 if (!(ssi->flags & LWSSSINFLAGS_PROXIED) &&
1304 pol == &pol_smd) {
1305 r = lws_ss_event_helper(h, LWSSSCS_CONNECTING);
1306 if (r || lws_fi(&h->fic, "ss_create_smd_1"))
1307 goto fail_creation;
1308 r = lws_ss_event_helper(h, LWSSSCS_CONNECTED);
1309 if (r || lws_fi(&h->fic, "ss_create_smd_2"))
1310 goto fail_creation;
1311 }
1312 #endif
1313
1314 if (!(ssi->flags & LWSSSINFLAGS_REGISTER_SINK) &&
1315 ((h->policy->flags & LWSSSPOLF_NAILED_UP)
1316 #if defined(LWS_WITH_SYS_SMD)
1317 || ((h->policy == &pol_smd) //&&
1318 //(ssi->flags & LWSSSINFLAGS_PROXIED))
1319 )
1320 #endif
1321 )) {
1322 r = _lws_ss_client_connect(h, 0, 0);
1323 if (lws_fi(&h->fic, "ss_create_conn"))
1324 r = LWSSSSRET_DESTROY_ME;
1325 switch (r) {
1326 case LWSSSSRET_OK:
1327 break;
1328 case LWSSSSRET_TX_DONT_SEND:
1329 case LWSSSSRET_DISCONNECT_ME:
1330 if (lws_ss_backoff(h) == LWSSSSRET_DESTROY_ME)
1331 goto fail_creation;
1332 break;
1333 case LWSSSSRET_DESTROY_ME:
1334 goto fail_creation;
1335 }
1336 }
1337
1338 return 0;
1339
1340 fail_creation:
1341
1342 if (ppss)
1343 *ppss = NULL;
1344
1345 lws_ss_destroy(&h);
1346
1347 return 1;
1348 }
1349
1350 void *
lws_ss_to_user_object(struct lws_ss_handle * h)1351 lws_ss_to_user_object(struct lws_ss_handle *h)
1352 {
1353 return (void *)&h[1];
1354 }
1355
1356 void
lws_ss_destroy(lws_ss_handle_t ** ppss)1357 lws_ss_destroy(lws_ss_handle_t **ppss)
1358 {
1359 struct lws_context_per_thread *pt;
1360 #if defined(LWS_WITH_SERVER)
1361 struct lws_vhost *v = NULL;
1362 #endif
1363 lws_ss_handle_t *h = *ppss;
1364 lws_ss_metadata_t *pmd;
1365
1366 if (!h)
1367 return;
1368
1369 lws_service_assert_loop_thread(h->context, h->tsi);
1370
1371 if (h == h->h_in_svc) {
1372 lwsl_err("%s: illegal destroy, return LWSSSSRET_DESTROY_ME instead\n",
1373 __func__);
1374 assert(0);
1375 return;
1376 }
1377
1378 if (h->destroying) {
1379 lwsl_info("%s: reentrant destroy\n", __func__);
1380 return;
1381 }
1382 h->destroying = 1;
1383
1384 #if defined(LWS_WITH_CONMON)
1385 if (h->conmon_json)
1386 lws_free_set_NULL(h->conmon_json);
1387 #endif
1388
1389 if (h->wsi) {
1390 /*
1391 * Don't let the wsi point to us any more,
1392 * we (the ss object bound to the wsi) are going away now
1393 */
1394 lws_set_opaque_user_data(h->wsi, NULL);
1395 lws_set_timeout(h->wsi, 1, LWS_TO_KILL_SYNC);
1396 }
1397
1398 /*
1399 * if we bound an smd registration to the SS, unregister it
1400 */
1401
1402 #if defined(LWS_WITH_SYS_SMD)
1403 if (h->policy == &pol_smd) {
1404 lws_sul_cancel(&h->u.smd.sul_write);
1405
1406 if (h->u.smd.smd_peer) {
1407 lws_smd_unregister(h->u.smd.smd_peer);
1408 h->u.smd.smd_peer = NULL;
1409 }
1410 }
1411 #endif
1412
1413 pt = &h->context->pt[h->tsi];
1414
1415 lws_pt_lock(pt, __func__);
1416 *ppss = NULL;
1417 lws_dll2_remove(&h->list);
1418 #if defined(LWS_WITH_SERVER)
1419 lws_dll2_remove(&h->cli_list);
1420 #endif
1421 lws_dll2_remove(&h->to_list);
1422 lws_sul_cancel(&h->sul_timeout);
1423
1424 /*
1425 * for lss, DESTROYING deletes the C++ lss object, making the
1426 * self-defined h->policy radioactive
1427 */
1428
1429 #if defined(LWS_WITH_SERVER)
1430 if (h->policy && (h->policy->flags & LWSSSPOLF_SERVER))
1431 v = lws_get_vhost_by_name(h->context, h->policy->streamtype);
1432 #endif
1433
1434 /*
1435 * Since we also come here to unpick create, it's possible we failed
1436 * the creation before issuing any states, even CREATING. We should
1437 * only issue cleanup states on destroy if we previously got as far as
1438 * issuing CREATING.
1439 */
1440
1441 if (h->prev_ss_state) {
1442 if (h->ss_dangling_connected)
1443 (void)lws_ss_event_helper(h, LWSSSCS_DISCONNECTED);
1444
1445 (void)lws_ss_event_helper(h, LWSSSCS_DESTROYING);
1446 }
1447
1448 lws_pt_unlock(pt);
1449
1450 /* in proxy case, metadata value on heap may need cleaning up */
1451
1452 pmd = h->metadata;
1453 while (pmd) {
1454 lwsl_info("%s: pmd %p\n", __func__, pmd);
1455 if (pmd->value_on_lws_heap)
1456 lws_free_set_NULL(pmd->value__may_own_heap);
1457
1458 pmd = pmd->next;
1459 }
1460
1461 #if defined(LWS_WITH_SS_DIRECT_PROTOCOL_STR)
1462 {
1463
1464 lws_ss_metadata_t *imd;
1465
1466 pmd = h->instant_metadata;
1467
1468 while (pmd) {
1469 imd = pmd;
1470 pmd = pmd->next;
1471
1472 lwsl_info("%s: instant md %p\n", __func__, imd);
1473 lws_free(imd);
1474 }
1475 h->instant_metadata = NULL;
1476
1477 if (h->imd_ac)
1478 lwsac_free(&h->imd_ac);
1479 }
1480 #endif
1481
1482 lws_sul_cancel(&h->sul);
1483
1484 #if defined(LWS_WITH_SECURE_STREAMS_STATIC_POLICY_ONLY)
1485
1486 /*
1487 * For static policy case, dynamically ref / instantiate the related
1488 * trust store and vhost. We do it by logical ss rather than connection
1489 * because we don't want to expose the latency of creating the x.509
1490 * trust store at the first connection.
1491 *
1492 * But it might be given the tls linkup takes time anyway, it can move
1493 * to the ss connect code instead.
1494 */
1495
1496 if (h->policy)
1497 lws_ss_policy_unref_trust_store(h->context, h->policy);
1498 #else
1499 #if defined(LWS_WITH_SECURE_STREAMS_CPP)
1500 if (!h->info.streamtype || !*(h->info.streamtype))
1501 lws_ss_policy_unref_trust_store(h->context, h->policy);
1502 #endif
1503 #endif
1504
1505 #if defined(LWS_WITH_SERVER)
1506 if (v)
1507 /*
1508 * For server, the policy describes a vhost that implements the
1509 * server, when we take down the ss, we take down the related
1510 * vhost (if it got that far)
1511 */
1512 lws_vhost_destroy(v);
1513 #endif
1514
1515 #if defined(LWS_WITH_SYS_FAULT_INJECTION)
1516 lws_fi_destroy(&h->fic);
1517 #endif
1518
1519 #if defined(LWS_WITH_SYS_METRICS)
1520 /*
1521 * If any hanging caliper measurement, dump it, and free any tags
1522 */
1523 lws_metrics_caliper_report_hist(h->cal_txn, (struct lws *)NULL);
1524 #endif
1525
1526 lws_sul_cancel(&h->sul_timeout);
1527
1528 /* confirm no sul left scheduled in handle or user allocation object */
1529 lws_sul_debug_zombies(h->context, h, sizeof(*h) + h->info.user_alloc,
1530 __func__);
1531
1532 __lws_lc_untag(h->context, &h->lc);
1533
1534 lws_explicit_bzero((void *)h, sizeof(*h) + h->info.user_alloc);
1535
1536 lws_free_set_NULL(h);
1537 }
1538
1539 #if defined(LWS_WITH_SERVER)
1540 void
lws_ss_server_ack(struct lws_ss_handle * h,int nack)1541 lws_ss_server_ack(struct lws_ss_handle *h, int nack)
1542 {
1543 h->txn_resp = nack;
1544 h->txn_resp_set = 1;
1545 }
1546
1547 void
lws_ss_server_foreach_client(struct lws_ss_handle * h,lws_sssfec_cb cb,void * arg)1548 lws_ss_server_foreach_client(struct lws_ss_handle *h, lws_sssfec_cb cb,
1549 void *arg)
1550 {
1551 lws_start_foreach_dll_safe(struct lws_dll2 *, d, d1, h->src_list.head) {
1552 struct lws_ss_handle *h =
1553 lws_container_of(d, struct lws_ss_handle, cli_list);
1554
1555 cb(h, arg);
1556
1557 } lws_end_foreach_dll_safe(d, d1);
1558 }
1559 #endif
1560
1561 lws_ss_state_return_t
lws_ss_request_tx(lws_ss_handle_t * h)1562 lws_ss_request_tx(lws_ss_handle_t *h)
1563 {
1564 lws_ss_state_return_t r;
1565
1566 r = _lws_ss_request_tx(h);
1567
1568 return r;
1569 }
1570
1571 lws_ss_state_return_t
_lws_ss_request_tx(lws_ss_handle_t * h)1572 _lws_ss_request_tx(lws_ss_handle_t *h)
1573 {
1574 lws_ss_state_return_t r;
1575
1576 // lwsl_notice("%s: h %p, wsi %p\n", __func__, h, h->wsi);
1577
1578 lws_service_assert_loop_thread(h->context, h->tsi);
1579
1580 if (h->wsi) {
1581 lws_callback_on_writable(h->wsi);
1582
1583 return LWSSSSRET_OK;
1584 }
1585
1586 if (!h->policy) {
1587 /* avoid crash */
1588 lwsl_err("%s: null policy\n", __func__);
1589 return LWSSSSRET_OK;
1590 }
1591
1592 if (h->policy->flags & LWSSSPOLF_SERVER)
1593 return LWSSSSRET_OK;
1594
1595 /*
1596 * there's currently no wsi / connection associated with the ss handle
1597 */
1598
1599 #if defined(LWS_WITH_SYS_SMD)
1600 if (h->policy == &pol_smd) {
1601 /*
1602 * He's an _lws_smd... and no wsi... since we're just going
1603 * to queue it, we could call his tx() right here, but rather
1604 * than surprise him let's set a sul to do it next time around
1605 * the event loop
1606 */
1607
1608 lws_sul_schedule(h->context, 0, &h->u.smd.sul_write,
1609 lws_ss_smd_tx_cb, 1);
1610
1611 return LWSSSSRET_OK;
1612 }
1613 #endif
1614
1615 if (h->seqstate != SSSEQ_IDLE &&
1616 h->seqstate != SSSEQ_DO_RETRY)
1617 return LWSSSSRET_OK;
1618
1619 h->seqstate = SSSEQ_TRY_CONNECT;
1620 r = lws_ss_event_helper(h, LWSSSCS_POLL);
1621 if (r)
1622 return r;
1623
1624 /*
1625 * Retries operate via lws_ss_request_tx(), explicitly ask for a
1626 * reconnection to clear the retry limit
1627 */
1628 r = _lws_ss_client_connect(h, 1, 0);
1629 if (r == LWSSSSRET_DESTROY_ME)
1630 return r;
1631
1632 if (r)
1633 return lws_ss_backoff(h);
1634
1635 return LWSSSSRET_OK;
1636 }
1637
1638 lws_ss_state_return_t
lws_ss_request_tx_len(lws_ss_handle_t * h,unsigned long len)1639 lws_ss_request_tx_len(lws_ss_handle_t *h, unsigned long len)
1640 {
1641 lws_service_assert_loop_thread(h->context, h->tsi);
1642
1643 if (h->wsi && h->policy &&
1644 (h->policy->protocol == LWSSSP_H1 ||
1645 h->policy->protocol == LWSSSP_H2 ||
1646 h->policy->protocol == LWSSSP_WS))
1647 h->wsi->http.writeable_len = len;
1648 else
1649 h->writeable_len = len;
1650
1651 return lws_ss_request_tx(h);
1652 }
1653
1654 /*
1655 * private helpers
1656 */
1657
1658 /* used on context destroy when iterating listed lws_ss on a pt */
1659
1660 int
lws_ss_destroy_dll(struct lws_dll2 * d,void * user)1661 lws_ss_destroy_dll(struct lws_dll2 *d, void *user)
1662 {
1663 lws_ss_handle_t *h = lws_container_of(d, lws_ss_handle_t, list);
1664
1665 lws_ss_destroy(&h);
1666
1667 return 0;
1668 }
1669
1670 int
lws_ss_cancel_notify_dll(struct lws_dll2 * d,void * user)1671 lws_ss_cancel_notify_dll(struct lws_dll2 *d, void *user)
1672 {
1673 lws_ss_handle_t *h = lws_container_of(d, lws_ss_handle_t, list);
1674
1675 if (lws_ss_event_helper(h, LWSSSCS_EVENT_WAIT_CANCELLED))
1676 lwsl_warn("%s: cancel event ignores return\n", __func__);
1677
1678 return 0;
1679 }
1680
1681 struct lws_sequencer *
lws_ss_get_sequencer(lws_ss_handle_t * h)1682 lws_ss_get_sequencer(lws_ss_handle_t *h)
1683 {
1684 return h->seq;
1685 }
1686
1687 struct lws_context *
lws_ss_get_context(struct lws_ss_handle * h)1688 lws_ss_get_context(struct lws_ss_handle *h)
1689 {
1690 return h->context;
1691 }
1692
1693 const char *
lws_ss_rideshare(struct lws_ss_handle * h)1694 lws_ss_rideshare(struct lws_ss_handle *h)
1695 {
1696 if (!h->rideshare)
1697 return h->policy->streamtype;
1698
1699 return h->rideshare->streamtype;
1700 }
1701
1702 int
lws_ss_add_peer_tx_credit(struct lws_ss_handle * h,int32_t bump)1703 lws_ss_add_peer_tx_credit(struct lws_ss_handle *h, int32_t bump)
1704 {
1705 const struct ss_pcols *ssp;
1706
1707 lws_service_assert_loop_thread(h->context, h->tsi);
1708
1709 ssp = ss_pcols[(int)h->policy->protocol];
1710
1711 if (h->wsi && ssp && ssp->tx_cr_add)
1712 return ssp->tx_cr_add(h, bump);
1713
1714 return 0;
1715 }
1716
1717 int
lws_ss_get_est_peer_tx_credit(struct lws_ss_handle * h)1718 lws_ss_get_est_peer_tx_credit(struct lws_ss_handle *h)
1719 {
1720 const struct ss_pcols *ssp;
1721
1722 lws_service_assert_loop_thread(h->context, h->tsi);
1723
1724 ssp = ss_pcols[(int)h->policy->protocol];
1725
1726 if (h->wsi && ssp && ssp->tx_cr_add)
1727 return ssp->tx_cr_est(h);
1728
1729 return 0;
1730 }
1731
1732 /*
1733 * protocol-independent handler for ss timeout
1734 */
1735
1736 static void
lws_ss_to_cb(lws_sorted_usec_list_t * sul)1737 lws_ss_to_cb(lws_sorted_usec_list_t *sul)
1738 {
1739 lws_ss_handle_t *h = lws_container_of(sul, lws_ss_handle_t, sul_timeout);
1740 lws_ss_state_return_t r;
1741
1742 lwsl_info("%s: %s timeout fired\n", __func__, lws_ss_tag(h));
1743
1744 r = lws_ss_event_helper(h, LWSSSCS_TIMEOUT);
1745 if (r != LWSSSSRET_DISCONNECT_ME && r != LWSSSSRET_DESTROY_ME)
1746 return;
1747
1748 if (h->wsi)
1749 lws_set_timeout(h->wsi, 1, LWS_TO_KILL_ASYNC);
1750
1751 _lws_ss_handle_state_ret_CAN_DESTROY_HANDLE(r, h->wsi, &h);
1752 }
1753
1754 void
lws_ss_start_timeout(struct lws_ss_handle * h,unsigned int timeout_ms)1755 lws_ss_start_timeout(struct lws_ss_handle *h, unsigned int timeout_ms)
1756 {
1757 lws_service_assert_loop_thread(h->context, h->tsi);
1758
1759 if (!timeout_ms && !h->policy->timeout_ms)
1760 return;
1761
1762 lws_sul_schedule(h->context, 0, &h->sul_timeout, lws_ss_to_cb,
1763 (timeout_ms ? timeout_ms : h->policy->timeout_ms) *
1764 LWS_US_PER_MS);
1765 }
1766
1767 void
lws_ss_cancel_timeout(struct lws_ss_handle * h)1768 lws_ss_cancel_timeout(struct lws_ss_handle *h)
1769 {
1770 lws_service_assert_loop_thread(h->context, h->tsi);
1771 lws_sul_cancel(&h->sul_timeout);
1772 }
1773
1774 void
lws_ss_change_handlers(struct lws_ss_handle * h,lws_ss_state_return_t (* rx)(void * userobj,const uint8_t * buf,size_t len,int flags),lws_ss_state_return_t (* tx)(void * userobj,lws_ss_tx_ordinal_t ord,uint8_t * buf,size_t * len,int * flags),lws_ss_state_return_t (* state)(void * userobj,void * h_src,lws_ss_constate_t state,lws_ss_tx_ordinal_t ack))1775 lws_ss_change_handlers(struct lws_ss_handle *h,
1776 lws_ss_state_return_t (*rx)(void *userobj, const uint8_t *buf,
1777 size_t len, int flags),
1778 lws_ss_state_return_t (*tx)(void *userobj, lws_ss_tx_ordinal_t ord,
1779 uint8_t *buf, size_t *len, int *flags),
1780 lws_ss_state_return_t (*state)(void *userobj, void *h_src /* ss handle type */,
1781 lws_ss_constate_t state,
1782 lws_ss_tx_ordinal_t ack))
1783 {
1784 if (rx)
1785 h->info.rx = rx;
1786 if (tx)
1787 h->info.tx = tx;
1788 if (state)
1789 h->info.state = state;
1790 }
1791
1792 const char *
lws_ss_tag(struct lws_ss_handle * h)1793 lws_ss_tag(struct lws_ss_handle *h)
1794 {
1795 if (!h)
1796 return "[null ss]";
1797 return lws_lc_tag(&h->lc);
1798 }
1799
1800 struct lws_log_cx *
lwsl_ss_get_cx(struct lws_ss_handle * ss)1801 lwsl_ss_get_cx(struct lws_ss_handle *ss)
1802 {
1803 if (!ss)
1804 return NULL;
1805
1806 return ss->lc.log_cx;
1807 }
1808
1809 void
lws_log_prepend_ss(struct lws_log_cx * cx,void * obj,char ** p,char * e)1810 lws_log_prepend_ss(struct lws_log_cx *cx, void *obj, char **p, char *e)
1811 {
1812 struct lws_ss_handle *h = (struct lws_ss_handle *)obj;
1813
1814 *p += lws_snprintf(*p, lws_ptr_diff_size_t(e, (*p)), "%s: ",
1815 lws_ss_tag(h));
1816 }
1817
1818 #if defined(_DEBUG)
1819 void
lws_ss_assert_extant(struct lws_context * cx,int tsi,struct lws_ss_handle * h)1820 lws_ss_assert_extant(struct lws_context *cx, int tsi, struct lws_ss_handle *h)
1821 {
1822 struct lws_context_per_thread *pt = &cx->pt[tsi];
1823
1824 lws_start_foreach_dll_safe(struct lws_dll2 *, d, d1, pt->ss_owner.head) {
1825 struct lws_ss_handle *h1 = lws_container_of(d,
1826 struct lws_ss_handle, list);
1827
1828 if (h == h1)
1829 return; /* okay */
1830
1831 } lws_end_foreach_dll_safe(d, d1);
1832
1833 /*
1834 * The ss handle is not listed in the pt ss handle owner...
1835 */
1836
1837 assert(0);
1838 }
1839 #endif
1840