Browse code

modules/websocket: tidied up some of the WS connection code

Peter Dunkley authored on 23/06/2012 10:31:50
Showing 4 changed files
... ...
@@ -110,7 +110,7 @@ void wsconn_destroy(void)
110 110
 			ws_connection_t *wsc = wsconn_hash[h];
111 111
 			while (wsc)
112 112
 			{
113
-				ws_connection_t *next = wsc->next;
113
+				ws_connection_t *next = wsc->id_next;
114 114
 				_wsconn_rm(wsc);
115 115
 				wsc = next;
116 116
 			}
... ...
@@ -136,17 +136,13 @@ void wsconn_destroy(void)
136 136
 	}
137 137
 }
138 138
 
139
-int wsconn_add(struct tcp_connection *con)
139
+int wsconn_add(int id)
140 140
 {
141 141
 	int cur_cons, max_cons;
142
+	int id_hash = tcp_id_hash(id);
142 143
 	ws_connection_t *wsc;
143 144
 
144
-	if (!con)
145
-	{
146
-		LM_ERR("wsconn_add: null pointer\n");
147
-		return -1;
148
-	}
149
-
145
+	/* Allocate and fill in new WebSocket connection */
150 146
 	wsc = shm_malloc(sizeof(ws_connection_t));
151 147
 	if (wsc == NULL)
152 148
 	{
... ...
@@ -155,20 +151,14 @@ int wsconn_add(struct tcp_connection *con)
155 151
 	}
156 152
 	memset(wsc, 0, sizeof(ws_connection_t));
157 153
 
158
-	wsc->con = con;
159
-	wsc->id_hash = con->id_hash;
154
+	wsc->id = id;
155
+	wsc->id_hash = id_hash;
160 156
 	wsc->last_used = (int)time(NULL);
161 157
 	wsc->state = WS_S_OPEN;
162 158
 
163
-	/* Make sure Kamailio core sends future messages on this connection
164
-	   directly to this module */
165
-	con->flags |= F_CONN_WS;
166
-
159
+	/* Add to WebSocket connection table */
167 160
 	lock_get(wsconn_lock);
168
-	wsc->next = wsconn_hash[wsc->id_hash];
169
-	wsc->prev = NULL;
170
-	if (wsconn_hash[wsc->id_hash]) wsconn_hash[wsc->id_hash]->prev = wsc;
171
-	wsconn_hash[wsc->id_hash] = wsc;
161
+	wsconn_listadd(wsconn_hash[wsc->id_hash], wsc, id_next, id_prev);
172 162
 	lock_release(wsconn_lock);
173 163
 
174 164
 	/* Update connection statistics */
... ...
@@ -185,10 +175,7 @@ int wsconn_add(struct tcp_connection *con)
185 175
 
186 176
 static inline void _wsconn_rm(ws_connection_t *wsc)
187 177
 {
188
-	if (wsconn_hash[wsc->id_hash] == wsc)
189
-		wsconn_hash[wsc->id_hash] = wsc->next;
190
-	if (wsc->next) wsc->next->prev = wsc->prev;
191
-	if (wsc->prev) wsc->prev->next = wsc->next;
178
+	wsconn_listrm(wsconn_hash[wsc->id_hash], wsc, id_next, id_prev);
192 179
 	shm_free(wsc);
193 180
 	wsc = NULL;
194 181
 	update_stat(ws_current_connections, -1);
... ...
@@ -223,27 +210,31 @@ int wsconn_update(ws_connection_t *wsc)
223 210
 
224 211
 void wsconn_close_now(ws_connection_t *wsc)
225 212
 {
226
-	wsc->con->send_flags.f |= SND_F_CON_CLOSE;
227
-	wsc->con->state = S_CONN_BAD;
228
-	wsc->con->timeout = get_ticks_raw();
213
+	struct tcp_connection *con = tcpconn_get(wsc->id, 0, 0, 0, 0);
214
+
215
+	if (con == NULL)
216
+	{
217
+		LM_ERR("getting TCP/TLS connection\n");
218
+		return;
219
+	}
220
+
221
+	con->send_flags.f |= SND_F_CON_CLOSE;
222
+	con->state = S_CONN_BAD;
223
+	con->timeout = get_ticks_raw();
224
+
229 225
 	if (wsconn_rm(wsc) < 0)
230 226
 		LM_ERR("removing WebSocket connection\n");
231 227
 }
232 228
 
233
-ws_connection_t *wsconn_find(struct tcp_connection *con)
229
+ws_connection_t *wsconn_get(int id)
234 230
 {
231
+	int id_hash = tcp_id_hash(id);
235 232
 	ws_connection_t *wsc;
236 233
 
237
-	if (!con)
238
-	{
239
-		LM_ERR("wsconn_find: null pointer\n");
240
-		return NULL;
241
-	}
242
-
243 234
 	lock_get(wsconn_lock);
244
-	for (wsc = wsconn_hash[con->id_hash]; wsc; wsc = wsc->next)
235
+	for (wsc = wsconn_hash[id_hash]; wsc; wsc = wsc->id_next)
245 236
 	{
246
-		if (wsc->con->id == con->id)
237
+		if (wsc->id == id)
247 238
 		{
248 239
 			lock_release(wsconn_lock);
249 240
 			return wsc;
... ...
@@ -271,18 +262,21 @@ struct mi_root *ws_mi_dump(struct mi_root *cmd, void *param)
271 262
 		wsc = wsconn_hash[h];
272 263
 		while(wsc)
273 264
 		{
274
-			if (wsc->con)
265
+			struct tcp_connection *con =
266
+					tcpconn_get(wsc->id, 0, 0, 0, 0);
267
+
268
+			if (con)
275 269
 			{
276
-				src_proto = (wsc->con->rcv.proto== PROTO_TCP)
277
-						? "tcp" : "tls";
270
+				src_proto = (con->rcv.proto== PROTO_TCP)
271
+						? "ws" : "wss";
278 272
 				memset(src_ip, 0, IP6_MAX_STR_SIZE + 1);
279
-				ip_addr2sbuf(&wsc->con->rcv.src_ip, src_ip,
273
+				ip_addr2sbuf(&con->rcv.src_ip, src_ip,
280 274
 						IP6_MAX_STR_SIZE);
281 275
 
282
-				dst_proto = (wsc->con->rcv.proto == PROTO_TCP)
283
-						? "tcp" : "tls";
276
+				dst_proto = (con->rcv.proto == PROTO_TCP)
277
+						? "ws" : "wss";
284 278
 				memset(dst_ip, 0, IP6_MAX_STR_SIZE + 1);
285
-				ip_addr2sbuf(&wsc->con->rcv.dst_ip, src_ip,
279
+				ip_addr2sbuf(&con->rcv.dst_ip, src_ip,
286 280
 						IP6_MAX_STR_SIZE);
287 281
 
288 282
 				interval = (int)time(NULL) - wsc->last_used;
... ...
@@ -291,13 +285,13 @@ struct mi_root *ws_mi_dump(struct mi_root *cmd, void *param)
291 285
 						"%d: %s:%s:%hu -> %s:%s:%hu "
292 286
 						"(state: %s, "
293 287
 						"last used %ds ago)",
294
-						wsc->con->id,
288
+						wsc->id,
295 289
 						src_proto,
296 290
 						strlen(src_ip) ? src_ip : "*",
297
-						wsc->con->rcv.src_port,
291
+						con->rcv.src_port,
298 292
 						dst_proto,
299 293
 						strlen(dst_ip) ? dst_ip : "*",
300
-						wsc->con->rcv.dst_port,
294
+						con->rcv.dst_port,
301 295
 						wsconn_state_str[wsc->state],
302 296
 						interval) == 0)
303 297
 					return 0;
... ...
@@ -309,7 +303,7 @@ struct mi_root *ws_mi_dump(struct mi_root *cmd, void *param)
309 303
 				}
310 304
 			}
311 305
 
312
-			wsc = wsc->next;
306
+			wsc = wsc->id_next;
313 307
 		}
314 308
 	}
315 309
 	lock_release(wsconn_lock);
... ...
@@ -25,7 +25,6 @@
25 25
 #define _WS_CONN_H
26 26
 
27 27
 #include "../../locking.h"
28
-#include "../../tcp_conn.h"
29 28
 #include "../../lib/kmi/tree.h"
30 29
 
31 30
 typedef enum
... ...
@@ -38,17 +37,18 @@ typedef enum
38 37
 
39 38
 typedef struct ws_connection
40 39
 {
41
-	struct tcp_connection *con;
42
-
43 40
 	ws_conn_state_t state;
44
-	int id;
45
-	unsigned id_hash;
46 41
 	int last_used;
47 42
 
48
-	struct ws_connection *prev;
49
-	struct ws_connection *next;
43
+	int id;			/* id and id_hash are identical to the values */
44
+	unsigned id_hash;	/* for the corresponding TCP/TLS connection */
45
+	struct ws_connection *id_prev;
46
+	struct ws_connection *id_next;
50 47
 } ws_connection_t;
51 48
 
49
+#define wsconn_listadd	tcpconn_listadd
50
+#define wsconn_listrm	tcpconn_listrm
51
+
52 52
 extern char *wsconn_state_str[];
53 53
 
54 54
 extern stat_var *ws_current_connections;
... ...
@@ -56,11 +56,11 @@ extern stat_var *ws_max_concurrent_connections;
56 56
 
57 57
 int wsconn_init(void);
58 58
 void wsconn_destroy(void);
59
-int wsconn_add(struct tcp_connection *con);
59
+int wsconn_add(int id);
60 60
 int wsconn_rm(ws_connection_t *wsc);
61 61
 int wsconn_update(ws_connection_t *wsc);
62 62
 void wsconn_close_now(ws_connection_t *wsc);
63
-ws_connection_t *wsconn_find(struct tcp_connection *con);
63
+ws_connection_t *wsconn_get(int id);
64 64
 struct mi_root *ws_mi_dump(struct mi_root *cmd, void *param);
65 65
 
66 66
 #endif /* _WS_CONN_H */
... ...
@@ -87,9 +87,6 @@ typedef enum
87 87
 #define OPCODE_PONG		(0xa)
88 88
 /* 0xb - 0xf are reserved for further control frames */
89 89
 
90
-static int close_connection(ws_connection_t *wsc, ws_close_type_t type,
91
-				short int status, str reason);
92
-
93 90
 stat_var *ws_failed_connections;
94 91
 stat_var *ws_local_closed_connections;
95 92
 stat_var *ws_received_frames;
... ...
@@ -109,177 +106,12 @@ static str str_status_bad_param = str_init("Bad connection ID parameter");
109 106
 static str str_status_error_closing = str_init("Error closing connection");
110 107
 static str str_status_error_sending = str_init("Error sending frame");
111 108
 
112
-static int decode_and_validate_ws_frame(ws_frame_t *frame,
113
-					tcp_event_info_t *tcpinfo)
114
-{
115
-	unsigned int i, len = tcpinfo->len;
116
-	int mask_start, j;
117
-	char *buf = tcpinfo->buf;
118
-
119
-	LM_INFO("decoding WebSocket frame\n");
120
-
121
-	if ((frame->wsc = wsconn_find(tcpinfo->con)) == NULL)
122
-	{
123
-		LM_WARN("WebSocket connection not found\n");
124
-		return -1;
125
-	}
126
-
127
-	wsconn_update(frame->wsc);
128
-
129
-	/* Decode and validate first 9 bits */
130
-	if (len < 2)
131
-	{
132
-		LM_WARN("message is too short\n");
133
-		if (close_connection(frame->wsc, LOCAL_CLOSE, 1002,
134
-					str_status_protocol_error) < 0)
135
-			LM_ERR("closing connection\n");
136
-		return -1;
137
-	}
138
-	frame->fin = (buf[0] & 0xff) & BYTE0_MASK_FIN;
139
-	frame->rsv1 = (buf[0] & 0xff) & BYTE0_MASK_RSV1;
140
-	frame->rsv2 = (buf[0] & 0xff) & BYTE0_MASK_RSV2;
141
-	frame->rsv3 = (buf[0] & 0xff) & BYTE0_MASK_RSV3;
142
-	frame->opcode = (buf[0] & 0xff) & BYTE0_MASK_OPCODE;
143
-	frame->mask = (buf[1] & 0xff) & BYTE1_MASK_MASK;
144
-	
145
-	if (!frame->fin)
146
-	{
147
-		LM_WARN("WebSocket fragmentation not supported in the sip "
148
-			"sub-protocol\n");
149
-		if (close_connection(frame->wsc, LOCAL_CLOSE, 1002,
150
-					str_status_protocol_error) < 0)
151
-			LM_ERR("closing connection\n");
152
-		return -1;
153
-	}
154
-
155
-	if (frame->rsv1 || frame->rsv2 || frame->rsv3)
156
-	{
157
-		LM_WARN("WebSocket reserved fields with non-zero values\n");
158
-		if (close_connection(frame->wsc, LOCAL_CLOSE, 1002,
159
-					str_status_protocol_error) < 0)
160
-			LM_ERR("closing connection\n");
161
-		return -1;
162
-	}
163
-
164
-	switch(frame->opcode)
165
-	{
166
-	case OPCODE_TEXT_FRAME:
167
-	case OPCODE_BINARY_FRAME:
168
-		LM_INFO("supported non-control frame: 0x%x\n",
169
-			(unsigned char) frame->opcode);
170
-		break;
171
-
172
-	case OPCODE_CLOSE:
173
-	case OPCODE_PING:
174
-	case OPCODE_PONG:
175
-		LM_INFO("supported control frame: 0x%x\n",
176
-			(unsigned char) frame->opcode);
177
-		break;
178
-
179
-	default:
180
-		LM_WARN("unsupported opcode: 0x%x\n",
181
-			(unsigned char) frame->opcode);
182
-		if (close_connection(frame->wsc, LOCAL_CLOSE, 1008,
183
-					str_status_unsupported_opcode) < 0)
184
-			LM_ERR("closing connection\n");
185
-		return -1;
186
-	}
187
-
188
-	if (!frame->mask)
189
-	{
190
-		LM_WARN("this is a server - all received messages must be "
191
-			"masked\n");
192
-		if (close_connection(frame->wsc, LOCAL_CLOSE, 1002,
193
-					str_status_protocol_error) < 0)
194
-			LM_ERR("closing connection\n");
195
-		return -1;
196
-	}
197
-
198
-	/* Decode and validate length */
199
-	frame->payload_len = (buf[1] & 0xff) & BYTE1_MASK_PAYLOAD_LEN;
200
-	if (frame->payload_len == 126)
201
-	{
202
-		if (len < 4)
203
-		{
204
-			LM_WARN("message is too short\n");
205
-			if (close_connection(frame->wsc, LOCAL_CLOSE, 1002,
206
-						str_status_protocol_error) < 0)
207
-				LM_ERR("closing connection\n");
208
-			return -1;
209
-		}
210
-		mask_start = 4;
211
-
212
-		frame->payload_len = 	  ((buf[2] & 0xff) <<  8)
213
-					| ((buf[3] & 0xff) <<  0);
214
-	}
215
-	else if (frame->payload_len == 127)
216
-	{
217
-		if (len < 10)
218
-		{
219
-			LM_WARN("message is too short\n");
220
-			if (close_connection(frame->wsc, LOCAL_CLOSE, 1002,
221
-						str_status_protocol_error) < 0)
222
-				LM_ERR("closing connection\n");
223
-			return -1;
224
-		}
225
-		mask_start = 10;
226
-
227
-		if ((buf[2] & 0xff) != 0 || (buf[3] & 0xff) != 0
228
-			|| (buf[4] & 0xff) != 0 || (buf[5] & 0xff) != 0)
229
-		{
230
-			LM_WARN("message is too long\n");
231
-			if (close_connection(frame->wsc, LOCAL_CLOSE, 1009,
232
-						str_status_message_too_big) < 0)
233
-				LM_ERR("closing connection\n");
234
-			return -1;
235
-		}
236
-
237
-		/* Only decoding the last four bytes of the length...
238
-		   This limits the size of WebSocket messages that can be
239
-		   handled to 2^32 = which should be plenty for SIP! */
240
-	 	frame->payload_len =	  ((buf[6] & 0xff) << 24)
241
-					| ((buf[7] & 0xff) << 16)
242
-					| ((buf[8] & 0xff) <<  8)
243
-					| ((buf[9] & 0xff) <<  0);
244
-	}
245
-	else
246
-		mask_start = 2;
247
-
248
-	/* Decode mask */
249
-	frame->masking_key[0] = (buf[mask_start + 0] & 0xff);
250
-	frame->masking_key[1] = (buf[mask_start + 1] & 0xff);
251
-	frame->masking_key[2] = (buf[mask_start + 2] & 0xff);
252
-	frame->masking_key[3] = (buf[mask_start + 3] & 0xff);
253
-
254
-	/* Decode and unmask payload */
255
-	if (len != frame->payload_len + mask_start + 4)
256
-	{
257
-		LM_WARN("message not complete frame size %u but received %u\n",
258
-			frame->payload_len + mask_start + 4, len);
259
-		if (close_connection(frame->wsc, LOCAL_CLOSE, 1002,
260
-					str_status_protocol_error) < 0)
261
-			LM_ERR("closing connection\n");
262
-		return -1;
263
-	}
264
-	frame->payload_data = &buf[mask_start + 4];
265
-	for (i = 0; i < frame->payload_len; i++)
266
-	{
267
-		j = i % 4;
268
-		frame->payload_data[i]
269
-			= frame->payload_data[i] ^ frame->masking_key[j];
270
-	}
271
-
272
-	LM_INFO("Rx (decoded): %.*s\n",
273
-		(int) frame->payload_len, frame->payload_data);
274
-
275
-	return frame->opcode;
276
-}
277
-
278 109
 static int encode_and_send_ws_frame(ws_frame_t *frame, conn_close_t conn_close)
279 110
 {
280 111
 	int pos = 0, extended_length;
281 112
 	unsigned int frame_length;
282 113
 	char *send_buf;
114
+	struct tcp_connection *con;
283 115
 	struct dest_info dst;
284 116
 
285 117
 	LM_INFO("encoding WebSocket frame\n");
... ...
@@ -372,7 +204,12 @@ static int encode_and_send_ws_frame(ws_frame_t *frame, conn_close_t conn_close)
372 204
 	}
373 205
 	memcpy(&send_buf[pos], frame->payload_data, frame->payload_len);
374 206
 
375
-	init_dst_from_rcv(&dst, &frame->wsc->con->rcv);
207
+	if ((con = tcpconn_get(frame->wsc->id, 0, 0, 0, 0)) == NULL)
208
+	{
209
+		LM_ERR("getting TCP/TLS connection\n");
210
+		return -1;
211
+	}
212
+	init_dst_from_rcv(&dst, &con->rcv);
376 213
 	if (conn_close == CONN_CLOSE_DO)
377 214
 	{
378 215
 		dst.send_flags.f |= SND_F_CON_CLOSE;
... ...
@@ -450,6 +287,172 @@ static int close_connection(ws_connection_t *wsc, ws_close_type_t type,
450 287
 	return 0;
451 288
 }
452 289
 
290
+static int decode_and_validate_ws_frame(ws_frame_t *frame,
291
+					tcp_event_info_t *tcpinfo)
292
+{
293
+	unsigned int i, len = tcpinfo->len;
294
+	int mask_start, j;
295
+	char *buf = tcpinfo->buf;
296
+
297
+	LM_INFO("decoding WebSocket frame\n");
298
+
299
+	if ((frame->wsc = wsconn_get(tcpinfo->con->id)) == NULL)
300
+	{
301
+		LM_WARN("WebSocket connection not found\n");
302
+		return -1;
303
+	}
304
+
305
+	wsconn_update(frame->wsc);
306
+
307
+	/* Decode and validate first 9 bits */
308
+	if (len < 2)
309
+	{
310
+		LM_WARN("message is too short\n");
311
+		if (close_connection(frame->wsc, LOCAL_CLOSE, 1002,
312
+					str_status_protocol_error) < 0)
313
+			LM_ERR("closing connection\n");
314
+		return -1;
315
+	}
316
+	frame->fin = (buf[0] & 0xff) & BYTE0_MASK_FIN;
317
+	frame->rsv1 = (buf[0] & 0xff) & BYTE0_MASK_RSV1;
318
+	frame->rsv2 = (buf[0] & 0xff) & BYTE0_MASK_RSV2;
319
+	frame->rsv3 = (buf[0] & 0xff) & BYTE0_MASK_RSV3;
320
+	frame->opcode = (buf[0] & 0xff) & BYTE0_MASK_OPCODE;
321
+	frame->mask = (buf[1] & 0xff) & BYTE1_MASK_MASK;
322
+	
323
+	if (!frame->fin)
324
+	{
325
+		LM_WARN("WebSocket fragmentation not supported in the sip "
326
+			"sub-protocol\n");
327
+		if (close_connection(frame->wsc, LOCAL_CLOSE, 1002,
328
+					str_status_protocol_error) < 0)
329
+			LM_ERR("closing connection\n");
330
+		return -1;
331
+	}
332
+
333
+	if (frame->rsv1 || frame->rsv2 || frame->rsv3)
334
+	{
335
+		LM_WARN("WebSocket reserved fields with non-zero values\n");
336
+		if (close_connection(frame->wsc, LOCAL_CLOSE, 1002,
337
+					str_status_protocol_error) < 0)
338
+			LM_ERR("closing connection\n");
339
+		return -1;
340
+	}
341
+
342
+	switch(frame->opcode)
343
+	{
344
+	case OPCODE_TEXT_FRAME:
345
+	case OPCODE_BINARY_FRAME:
346
+		LM_INFO("supported non-control frame: 0x%x\n",
347
+			(unsigned char) frame->opcode);
348
+		break;
349
+
350
+	case OPCODE_CLOSE:
351
+	case OPCODE_PING:
352
+	case OPCODE_PONG:
353
+		LM_INFO("supported control frame: 0x%x\n",
354
+			(unsigned char) frame->opcode);
355
+		break;
356
+
357
+	default:
358
+		LM_WARN("unsupported opcode: 0x%x\n",
359
+			(unsigned char) frame->opcode);
360
+		if (close_connection(frame->wsc, LOCAL_CLOSE, 1008,
361
+					str_status_unsupported_opcode) < 0)
362
+			LM_ERR("closing connection\n");
363
+		return -1;
364
+	}
365
+
366
+	if (!frame->mask)
367
+	{
368
+		LM_WARN("this is a server - all received messages must be "
369
+			"masked\n");
370
+		if (close_connection(frame->wsc, LOCAL_CLOSE, 1002,
371
+					str_status_protocol_error) < 0)
372
+			LM_ERR("closing connection\n");
373
+		return -1;
374
+	}
375
+
376
+	/* Decode and validate length */
377
+	frame->payload_len = (buf[1] & 0xff) & BYTE1_MASK_PAYLOAD_LEN;
378
+	if (frame->payload_len == 126)
379
+	{
380
+		if (len < 4)
381
+		{
382
+			LM_WARN("message is too short\n");
383
+			if (close_connection(frame->wsc, LOCAL_CLOSE, 1002,
384
+						str_status_protocol_error) < 0)
385
+				LM_ERR("closing connection\n");
386
+			return -1;
387
+		}
388
+		mask_start = 4;
389
+
390
+		frame->payload_len = 	  ((buf[2] & 0xff) <<  8)
391
+					| ((buf[3] & 0xff) <<  0);
392
+	}
393
+	else if (frame->payload_len == 127)
394
+	{
395
+		if (len < 10)
396
+		{
397
+			LM_WARN("message is too short\n");
398
+			if (close_connection(frame->wsc, LOCAL_CLOSE, 1002,
399
+						str_status_protocol_error) < 0)
400
+				LM_ERR("closing connection\n");
401
+			return -1;
402
+		}
403
+		mask_start = 10;
404
+
405
+		if ((buf[2] & 0xff) != 0 || (buf[3] & 0xff) != 0
406
+			|| (buf[4] & 0xff) != 0 || (buf[5] & 0xff) != 0)
407
+		{
408
+			LM_WARN("message is too long\n");
409
+			if (close_connection(frame->wsc, LOCAL_CLOSE, 1009,
410
+						str_status_message_too_big) < 0)
411
+				LM_ERR("closing connection\n");
412
+			return -1;
413
+		}
414
+
415
+		/* Only decoding the last four bytes of the length...
416
+		   This limits the size of WebSocket messages that can be
417
+		   handled to 2^32 = which should be plenty for SIP! */
418
+	 	frame->payload_len =	  ((buf[6] & 0xff) << 24)
419
+					| ((buf[7] & 0xff) << 16)
420
+					| ((buf[8] & 0xff) <<  8)
421
+					| ((buf[9] & 0xff) <<  0);
422
+	}
423
+	else
424
+		mask_start = 2;
425
+
426
+	/* Decode mask */
427
+	frame->masking_key[0] = (buf[mask_start + 0] & 0xff);
428
+	frame->masking_key[1] = (buf[mask_start + 1] & 0xff);
429
+	frame->masking_key[2] = (buf[mask_start + 2] & 0xff);
430
+	frame->masking_key[3] = (buf[mask_start + 3] & 0xff);
431
+
432
+	/* Decode and unmask payload */
433
+	if (len != frame->payload_len + mask_start + 4)
434
+	{
435
+		LM_WARN("message not complete frame size %u but received %u\n",
436
+			frame->payload_len + mask_start + 4, len);
437
+		if (close_connection(frame->wsc, LOCAL_CLOSE, 1002,
438
+					str_status_protocol_error) < 0)
439
+			LM_ERR("closing connection\n");
440
+		return -1;
441
+	}
442
+	frame->payload_data = &buf[mask_start + 4];
443
+	for (i = 0; i < frame->payload_len; i++)
444
+	{
445
+		j = i % 4;
446
+		frame->payload_data[i]
447
+			= frame->payload_data[i] ^ frame->masking_key[j];
448
+	}
449
+
450
+	LM_INFO("Rx (decoded): %.*s\n",
451
+		(int) frame->payload_len, frame->payload_data);
452
+
453
+	return frame->opcode;
454
+}
455
+
453 456
 static int handle_sip_message(ws_frame_t *frame)
454 457
 {
455 458
 	LM_INFO("Received SIP message\n");
... ...
@@ -562,6 +565,26 @@ int ws_frame_received(void *data)
562 565
 	return 0;
563 566
 }
564 567
 
568
+static int ping_pong(ws_connection_t *wsc, int opcode)
569
+{
570
+	ws_frame_t frame;
571
+
572
+	memset(&frame, 0, sizeof(frame));
573
+	frame.fin = 1;
574
+	frame.opcode = opcode;
575
+	frame.payload_len = server_hdr.len;
576
+	frame.payload_data = server_hdr.s;
577
+	frame.wsc = wsc;
578
+
579
+	if (encode_and_send_ws_frame(&frame, CONN_CLOSE_DONT) < 0)
580
+	{	
581
+		LM_ERR("closing connection\n");
582
+		return -1;
583
+	}
584
+
585
+	return 0;
586
+}
587
+
565 588
 struct mi_root *ws_mi_close(struct mi_root *cmd, void *param)
566 589
 {
567 590
 	unsigned int id;
... ...
@@ -589,7 +612,7 @@ struct mi_root *ws_mi_close(struct mi_root *cmd, void *param)
589 612
 					str_status_too_many_params.len);
590 613
 	}
591 614
 
592
-	if ((wsc = wsconn_find(tcpconn_get(id, 0, 0, 0, 0))) == NULL)
615
+	if ((wsc = wsconn_get(id)) == NULL)
593 616
 	{
594 617
 		LM_ERR("bad connection ID parameter\n");
595 618
 		return init_mi_tree(400, str_status_bad_param.s,
... ...
@@ -607,26 +630,6 @@ struct mi_root *ws_mi_close(struct mi_root *cmd, void *param)
607 630
 	return init_mi_tree(200, MI_OK_S, MI_OK_LEN);
608 631
 }
609 632
 
610
-static int ping_pong(ws_connection_t *wsc, int opcode)
611
-{
612
-	ws_frame_t frame;
613
-
614
-	memset(&frame, 0, sizeof(frame));
615
-	frame.fin = 1;
616
-	frame.opcode = opcode;
617
-	frame.payload_len = server_hdr.len;
618
-	frame.payload_data = server_hdr.s;
619
-	frame.wsc = wsc;
620
-
621
-	if (encode_and_send_ws_frame(&frame, CONN_CLOSE_DONT) < 0)
622
-	{	
623
-		LM_ERR("closing connection\n");
624
-		return -1;
625
-	}
626
-
627
-	return 0;
628
-}
629
-
630 633
 static struct mi_root *mi_ping_pong(struct mi_root *cmd, void *param,
631 634
 					int opcode)
632 635
 {
... ...
@@ -655,7 +658,7 @@ static struct mi_root *mi_ping_pong(struct mi_root *cmd, void *param,
655 658
 					str_status_too_many_params.len);
656 659
 	}
657 660
 
658
-	if ((wsc = wsconn_find(tcpconn_get(id, 0, 0, 0, 0))) == NULL)
661
+	if ((wsc = wsconn_get(id)) == NULL)
659 662
 	{
660 663
 		LM_ERR("bad connection ID parameter\n");
661 664
 		return init_mi_tree(400, str_status_bad_param.s,
... ...
@@ -299,7 +299,11 @@ int ws_handle_handshake(struct sip_msg *msg)
299 299
 		return 0;
300 300
 
301 301
 	/* Add the connection to the WebSocket connection table */
302
-	wsconn_add(con);
302
+	wsconn_add(con->id);
303
+
304
+	/* Make sure Kamailio core sends future messages on this connection
305
+	   directly to this module */
306
+	con->flags |= F_CONN_WS;
303 307
 
304 308
 	return 0;
305 309
 }