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 int f = 0, txr;
35 size_t buflen;
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 if (h->info.rx(ss_to_userobj(h), (const uint8_t *)pmqpp->payload,
97 len, f) < 0)
98 return -1;
99
100 return 0; /* don't passthru */
101
102 case LWS_CALLBACK_MQTT_SUBSCRIBED:
103 wsi->mqtt->done_subscribe = 1;
104 lws_callback_on_writable(wsi);
105 break;
106
107 case LWS_CALLBACK_MQTT_ACK:
108 lws_ss_event_helper(h, LWSSSCS_QOS_ACK_REMOTE);
109 break;
110
111 case LWS_CALLBACK_MQTT_CLIENT_WRITEABLE:
112 if (!h)
113 return 0;
114 lwsl_notice("%s: ss %p: WRITEABLE\n", __func__, h);
115
116 if (h->seqstate != SSSEQ_CONNECTED) {
117 lwsl_warn("%s: seqstate %d\n", __func__, h->seqstate);
118 break;
119 }
120
121 if (h->policy->u.mqtt.subscribe && !wsi->mqtt->done_subscribe) {
122
123 /*
124 * The policy says to subscribe to something, and we
125 * haven't done it yet
126 */
127
128 lwsl_warn("%s: subscribing %s\n", __func__, h->policy->u.mqtt.subscribe);
129
130 memset(&h->u.mqtt.sub_top, 0, sizeof(h->u.mqtt.sub_top));
131 h->u.mqtt.sub_top.name = h->policy->u.mqtt.subscribe;
132 h->u.mqtt.sub_top.qos = h->policy->u.mqtt.qos;
133 memset(&h->u.mqtt.sub_info, 0, sizeof(h->u.mqtt.sub_info));
134 h->u.mqtt.sub_info.num_topics = 1;
135 h->u.mqtt.sub_info.topic = &h->u.mqtt.sub_top;
136
137 if (lws_mqtt_client_send_subcribe(wsi, &h->u.mqtt.sub_info)) {
138 lwsl_notice("%s: unable to subscribe", __func__);
139 return -1;
140 }
141
142 return 0;
143 }
144
145
146 buflen = sizeof(buf) - LWS_PRE;
147 txr = h->info.tx(ss_to_userobj(h), h->txord++, buf + LWS_PRE,
148 &buflen, &f);
149 if (txr < 0) {
150 lwsl_debug("%s: tx handler asked to close\n", __func__);
151 return -1;
152 }
153 if (txr > 0)
154 /* don't want to send anything */
155 return 0;
156
157 memset(&mqpp, 0, sizeof(mqpp));
158 mqpp.topic = (char *)h->policy->u.mqtt.topic;
159 mqpp.topic_len = strlen(mqpp.topic);
160 mqpp.packet_id = h->txord - 1;
161 mqpp.payload = buf + LWS_PRE;
162 if (h->writeable_len)
163 mqpp.payload_len = h->writeable_len;
164 else
165 mqpp.payload_len = buflen;
166
167 lwsl_notice("%s: payload len %d\n", __func__, (int)mqpp.payload_len);
168
169 mqpp.qos = h->policy->u.mqtt.qos;
170
171 if (lws_mqtt_client_send_publish(wsi, &mqpp,
172 (const char *)buf + LWS_PRE, buflen,
173 f & LWSSS_FLAG_EOM)) {
174 lwsl_notice("%s: failed to publish\n", __func__);
175
176 return -1;
177 }
178
179 return 0;
180
181 default:
182 break;
183 }
184
185 return lws_callback_http_dummy(wsi, reason, user, in, len);
186 }
187
188 const struct lws_protocols protocol_secstream_mqtt = {
189 "lws-secstream-mqtt",
190 secstream_mqtt,
191 0,
192 0,
193 };
194 /*
195 * Munge connect info according to protocol-specific considerations... this
196 * usually means interpreting aux in a protocol-specific way and using the
197 * pieces at connection setup time, eg, http url pieces.
198 *
199 * len bytes of buf can be used for things with scope until after the actual
200 * connect.
201 *
202 * For ws, protocol aux is <url path>;<ws subprotocol name>
203 */
204
205 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)206 secstream_connect_munge_mqtt(lws_ss_handle_t *h, char *buf, size_t len,
207 struct lws_client_connect_info *i,
208 union lws_ss_contemp *ct)
209 {
210 memset(&ct->ccp, 0, sizeof(ct->ccp));
211
212 ct->ccp.client_id = "lwsMqttClient";
213 ct->ccp.keep_alive = h->policy->u.mqtt.keep_alive;
214 ct->ccp.clean_start = h->policy->u.mqtt.clean_start;
215 ct->ccp.will_param.topic = h->policy->u.mqtt.will_topic;
216 ct->ccp.will_param.message = h->policy->u.mqtt.will_message;
217 ct->ccp.will_param.qos = h->policy->u.mqtt.will_qos;
218 ct->ccp.will_param.retain = h->policy->u.mqtt.will_retain;
219
220 lwsl_notice("%s\n", __func__);
221
222 h->u.mqtt.topic_qos.name = h->policy->u.mqtt.subscribe;
223 h->u.mqtt.topic_qos.qos = h->policy->u.mqtt.qos;
224
225 i->method = "MQTT";
226 i->mqtt_cp = &ct->ccp;
227
228 i->alpn = "x-amzn-mqtt-ca";
229
230 /* share connections where possible */
231 i->ssl_connection |= LCCSCF_PIPELINE;
232
233 /*
234 if (!h->policy->u.http.url)
235 return 0;
236
237 // protocol aux is the path part ; ws subprotocol name
238
239 i->path = NULL;
240 lws_snprintf(buf, len, "/%s", h->policy->u.mqtt.topic);
241
242 // i->protocol = h->policy->u.mqtt.u.ws.subprotocol;
243
244 lwsl_notice("%s: url %s, ws subprotocol %s\n", __func__, buf, i->protocol);
245 */
246 return 0;
247 }
248
249 const struct ss_pcols ss_pcol_mqtt = {
250 "MQTT",
251 "x-amzn-mqtt-ca", //"mqtt/3.1.1",
252 "lws-secstream-mqtt",
253 secstream_connect_munge_mqtt
254 };
255