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 #include <private-lib-core.h>
26
27 static int
secstream_mqtt(struct lws * wsi,enum lws_callback_reasons reason,void * user,void * in,size_t len)28 secstream_mqtt(struct lws *wsi, enum lws_callback_reasons reason, void *user,
29 void *in, size_t len)
30 {
31 lws_ss_handle_t *h = (lws_ss_handle_t *)lws_get_opaque_user_data(wsi);
32 lws_mqtt_publish_param_t mqpp, *pmqpp;
33 uint8_t buf[LWS_PRE + 1400];
34 size_t buflen;
35 int f = 0;
36
37 switch (reason) {
38
39 /* because we are protocols[0] ... */
40 case LWS_CALLBACK_CLIENT_CONNECTION_ERROR:
41 lwsl_info("%s: CLIENT_CONNECTION_ERROR: %s\n", __func__,
42 in ? (char *)in : "(null)");
43 if (!h)
44 break;
45 lws_ss_event_helper(h, LWSSSCS_UNREACHABLE);
46 h->wsi = NULL;
47 lws_ss_backoff(h);
48 break;
49
50 case LWS_CALLBACK_MQTT_CLIENT_CLOSED:
51 if (!h)
52 break;
53 f = lws_ss_event_helper(h, LWSSSCS_DISCONNECTED);
54 if (h->wsi)
55 lws_set_opaque_user_data(h->wsi, NULL);
56 h->wsi = NULL;
57 if (f) {
58 lws_ss_destroy(&h);
59 break;
60 }
61
62 if (h->policy && !(h->policy->flags & LWSSSPOLF_OPPORTUNISTIC) &&
63 !h->txn_ok && !wsi->context->being_destroyed)
64 lws_ss_backoff(h);
65 break;
66
67 case LWS_CALLBACK_MQTT_CLIENT_ESTABLISHED:
68 /*
69 * Make sure the handle wsi points to the stream wsi not the
70 * original nwsi, in the case it was migrated
71 */
72 h->wsi = wsi;
73 h->retry = 0;
74 h->seqstate = SSSEQ_CONNECTED;
75 lws_ss_set_timeout_us(h, LWS_SET_TIMER_USEC_CANCEL);
76 lws_ss_event_helper(h, LWSSSCS_CONNECTED);
77 if (h->policy->u.mqtt.topic)
78 lws_callback_on_writable(wsi);
79 break;
80
81 case LWS_CALLBACK_MQTT_CLIENT_RX:
82 // lwsl_user("LWS_CALLBACK_CLIENT_RECEIVE: read %d\n", (int)len);
83 if (!h)
84 return 0;
85
86 pmqpp = (lws_mqtt_publish_param_t *)in;
87
88 f = 0;
89 if (!pmqpp->payload_pos)
90 f |= LWSSS_FLAG_SOM;
91 if (pmqpp->payload_pos + len == pmqpp->payload_len)
92 f |= LWSSS_FLAG_EOM;
93
94 h->subseq = 1;
95
96 h->info.rx(ss_to_userobj(h), (const uint8_t *)pmqpp->payload,
97 len, f);
98
99 return 0; /* don't passthru */
100
101 case LWS_CALLBACK_MQTT_SUBSCRIBED:
102 wsi->mqtt->done_subscribe = 1;
103 lws_callback_on_writable(wsi);
104 break;
105
106 case LWS_CALLBACK_MQTT_ACK:
107 lws_ss_event_helper(h, LWSSSCS_QOS_ACK_REMOTE);
108 break;
109
110 case LWS_CALLBACK_MQTT_CLIENT_WRITEABLE:
111 if (!h)
112 return 0;
113 lwsl_notice("%s: ss %p: WRITEABLE\n", __func__, h);
114
115 if (h->seqstate != SSSEQ_CONNECTED) {
116 lwsl_warn("%s: seqstate %d\n", __func__, h->seqstate);
117 break;
118 }
119
120 if (h->policy->u.mqtt.subscribe && !wsi->mqtt->done_subscribe) {
121
122 /*
123 * The policy says to subscribe to something, and we
124 * haven't done it yet
125 */
126
127 lwsl_warn("%s: subscribing %s\n", __func__, h->policy->u.mqtt.subscribe);
128
129 memset(&h->u.mqtt.sub_top, 0, sizeof(h->u.mqtt.sub_top));
130 h->u.mqtt.sub_top.name = h->policy->u.mqtt.subscribe;
131 h->u.mqtt.sub_top.qos = h->policy->u.mqtt.qos;
132 memset(&h->u.mqtt.sub_info, 0, sizeof(h->u.mqtt.sub_info));
133 h->u.mqtt.sub_info.num_topics = 1;
134 h->u.mqtt.sub_info.topic = &h->u.mqtt.sub_top;
135
136 if (lws_mqtt_client_send_subcribe(wsi, &h->u.mqtt.sub_info)) {
137 lwsl_notice("%s: unable to subscribe", __func__);
138 return -1;
139 }
140
141 return 0;
142 }
143
144
145 buflen = sizeof(buf) - LWS_PRE;
146 if (h->info.tx(ss_to_userobj(h), h->txord++, buf + LWS_PRE,
147 &buflen, &f))
148 /* don't want to send anything */
149 return 0;
150
151 memset(&mqpp, 0, sizeof(mqpp));
152 mqpp.topic = (char *)h->policy->u.mqtt.topic;
153 mqpp.topic_len = strlen(mqpp.topic);
154 mqpp.packet_id = h->txord - 1;
155 mqpp.payload = buf + LWS_PRE;
156 if (h->writeable_len)
157 mqpp.payload_len = h->writeable_len;
158 else
159 mqpp.payload_len = buflen;
160
161 lwsl_notice("%s: payload len %d\n", __func__, (int)mqpp.payload_len);
162
163 mqpp.qos = h->policy->u.mqtt.qos;
164
165 if (lws_mqtt_client_send_publish(wsi, &mqpp,
166 (const char *)buf + LWS_PRE, buflen,
167 f & LWSSS_FLAG_EOM)) {
168 lwsl_notice("%s: failed to publish\n", __func__);
169
170 return -1;
171 }
172
173 return 0;
174
175 default:
176 break;
177 }
178
179 return lws_callback_http_dummy(wsi, reason, user, in, len);
180 }
181
182 const struct lws_protocols protocol_secstream_mqtt = {
183 "lws-secstream-mqtt",
184 secstream_mqtt,
185 0,
186 0,
187 };
188 /*
189 * Munge connect info according to protocol-specific considerations... this
190 * usually means interpreting aux in a protocol-specific way and using the
191 * pieces at connection setup time, eg, http url pieces.
192 *
193 * len bytes of buf can be used for things with scope until after the actual
194 * connect.
195 *
196 * For ws, protocol aux is <url path>;<ws subprotocol name>
197 */
198
199 static int
secstream_connect_munge_mqtt(lws_ss_handle_t * h,char * buf,size_t len,struct lws_client_connect_info * i,union lws_ss_contemp * ct)200 secstream_connect_munge_mqtt(lws_ss_handle_t *h, char *buf, size_t len,
201 struct lws_client_connect_info *i,
202 union lws_ss_contemp *ct)
203 {
204 memset(&ct->ccp, 0, sizeof(ct->ccp));
205
206 ct->ccp.client_id = "lwsMqttClient";
207 ct->ccp.keep_alive = h->policy->u.mqtt.keep_alive;
208 ct->ccp.clean_start = h->policy->u.mqtt.clean_start;
209 ct->ccp.will_param.topic = h->policy->u.mqtt.will_topic;
210 ct->ccp.will_param.message = h->policy->u.mqtt.will_message;
211 ct->ccp.will_param.qos = h->policy->u.mqtt.will_qos;
212 ct->ccp.will_param.retain = h->policy->u.mqtt.will_retain;
213
214 lwsl_notice("%s\n", __func__);
215
216 h->u.mqtt.topic_qos.name = h->policy->u.mqtt.subscribe;
217 h->u.mqtt.topic_qos.qos = h->policy->u.mqtt.qos;
218
219 i->method = "MQTT";
220 i->mqtt_cp = &ct->ccp;
221
222 i->alpn = "x-amzn-mqtt-ca";
223
224 /* share connections where possible */
225 i->ssl_connection |= LCCSCF_PIPELINE;
226
227 /*
228 if (!h->policy->u.http.url)
229 return 0;
230
231 // protocol aux is the path part ; ws subprotocol name
232
233 i->path = NULL;
234 lws_snprintf(buf, len, "/%s", h->policy->u.mqtt.topic);
235
236 // i->protocol = h->policy->u.mqtt.u.ws.subprotocol;
237
238 lwsl_notice("%s: url %s, ws subprotocol %s\n", __func__, buf, i->protocol);
239 */
240 return 0;
241 }
242
243 const struct ss_pcols ss_pcol_mqtt = {
244 "MQTT",
245 "x-amzn-mqtt-ca", //"mqtt/3.1.1",
246 "lws-secstream-mqtt",
247 secstream_connect_munge_mqtt
248 };
249