24 Sep
2015
24 Sep
'15
4:43 p.m.
Module: kamailio
Branch: master
Commit: 38a79cdc42b952ac0291d249241ca6499fc269aa
URL: https://github.com/kamailio/kamailio/commit/38a79cdc42b952ac0291d249241ca64…
Author: Hugh Waite <hugh.waite(a)xura.com>
Committer: Hugh Waite <hugh.waite(a)xura.com>
Date: 2015-09-23T21:59:09+01:00
websocket: Add support for SIP message fragmentation
- websocket connections are created with a BUF_SIZE (64K) buffer used for concatenating
frames
- continuation frames (fragments) are supported for the SIP sub-protocol
---
Modified: modules/websocket/ws_conn.c
Modified: modules/websocket/ws_conn.h
Modified: modules/websocket/ws_frame.c
---
Diff:
https://github.com/kamailio/kamailio/commit/38a79cdc42b952ac0291d249241ca64…
Patch:
https://github.com/kamailio/kamailio/commit/38a79cdc42b952ac0291d249241ca64…
---
diff --git a/modules/websocket/ws_conn.c b/modules/websocket/ws_conn.c
index 87d593d..b8ede8b 100644
--- a/modules/websocket/ws_conn.c
+++ b/modules/websocket/ws_conn.c
@@ -201,19 +201,20 @@ int wsconn_add(struct receive_info rcv, unsigned int sub_protocol)
LM_DBG("wsconn_add id [%d]\n", id);
/* Allocate and fill in new WebSocket connection */
- wsc = shm_malloc(sizeof(ws_connection_t));
+ wsc = shm_malloc(sizeof(ws_connection_t) + BUF_SIZE);
if (wsc == NULL)
{
LM_ERR("allocating shared memory\n");
return -1;
}
- memset(wsc, 0, sizeof(ws_connection_t));
+ memset(wsc, 0, sizeof(ws_connection_t) + BUF_SIZE);
wsc->id = id;
wsc->id_hash = id_hash;
wsc->state = WS_S_OPEN;
wsc->rcv = rcv;
wsc->sub_protocol = sub_protocol;
wsc->run_event = 0;
+ wsc->frag_buf.s = ((char*)wsc) + sizeof(ws_connection_t);
atomic_set(&wsc->refcnt, 0);
LM_DBG("wsconn_add new wsc => [%p], ref => [%d]\n", wsc,
atomic_get(&wsc->refcnt));
diff --git a/modules/websocket/ws_conn.h b/modules/websocket/ws_conn.h
index 029d570..df6400a 100644
--- a/modules/websocket/ws_conn.h
+++ b/modules/websocket/ws_conn.h
@@ -60,6 +60,8 @@ typedef struct ws_connection
atomic_t refcnt;
int run_event;
+
+ str frag_buf;
} ws_connection_t;
typedef struct
diff --git a/modules/websocket/ws_frame.c b/modules/websocket/ws_frame.c
index e890770..16f7901 100644
--- a/modules/websocket/ws_frame.c
+++ b/modules/websocket/ws_frame.c
@@ -431,15 +431,6 @@ static int decode_and_validate_ws_frame(ws_frame_t *frame,
frame->opcode = (buf[0] & 0xff) & BYTE0_MASK_OPCODE;
frame->mask = (buf[1] & 0xff) & BYTE1_MASK_MASK;
- if (!frame->fin)
- {
- LM_WARN("WebSocket fragmentation not supported in the sip "
- "sub-protocol\n");
- *err_code = 1002;
- *err_text = str_status_protocol_error;
- return -1;
- }
-
if (frame->rsv1 || frame->rsv2 || frame->rsv3)
{
LM_WARN("WebSocket reserved fields with non-zero values\n");
@@ -450,6 +441,10 @@ static int decode_and_validate_ws_frame(ws_frame_t *frame,
switch(frame->opcode)
{
+ case OPCODE_CONTINUATION:
+ LM_DBG("supported continuation frame: 0x%x\n",
+ (unsigned char) frame->opcode);
+ break;
case OPCODE_TEXT_FRAME:
case OPCODE_BINARY_FRAME:
LM_DBG("supported non-control frame: 0x%x\n",
@@ -651,6 +646,36 @@ int ws_frame_receive(void *data)
switch(opcode)
{
+ case OPCODE_CONTINUATION:
+ if (likely(frame.wsc->sub_protocol == SUB_PROTOCOL_SIP))
+ {
+ if (frame.wsc->frag_buf.len + frame.payload_len >= BUF_SIZE)
+ {
+ LM_ERR("Buffer overflow assembling websocket fragments %d + %d = %d\n",
frame.wsc->frag_buf.len, frame.payload_len, frame.wsc->frag_buf.len +
frame.payload_len);
+ wsconn_put(frame.wsc);
+ return -1;
+ }
+ memcpy(frame.wsc->frag_buf.s + frame.wsc->frag_buf.len, frame.payload_data,
frame.payload_len);
+ frame.wsc->frag_buf.len += frame.payload_len;
+ frame.wsc->frag_buf.s[frame.wsc->frag_buf.len] = '\0';
+
+ if (frame.fin)
+ {
+ ret = receive_msg(frame.wsc->frag_buf.s,
+ frame.wsc->frag_buf.len,
+ tcpinfo->rcv);
+ wsconn_put(frame.wsc);
+ return ret;
+ }
+ wsconn_put(frame.wsc);
+ return 0;
+ }
+ else
+ {
+ LM_ERR("Unsupported fragmented sub-protocol");
+ wsconn_put(frame.wsc);
+ return -1;
+ }
case OPCODE_TEXT_FRAME:
case OPCODE_BINARY_FRAME:
if (likely(frame.wsc->sub_protocol == SUB_PROTOCOL_SIP))
@@ -659,11 +684,23 @@ int ws_frame_receive(void *data)
frame.payload_data);
update_stat(ws_sip_received_frames, 1);
- wsconn_put(frame.wsc);
+ if (frame.fin)
+ {
- return receive_msg(frame.payload_data,
+ wsconn_put(frame.wsc);
+
+ return receive_msg(frame.payload_data,
frame.payload_len,
tcpinfo->rcv);
+ }
+ else
+ {
+ memcpy(frame.wsc->frag_buf.s, frame.payload_data, frame.payload_len);
+ frame.wsc->frag_buf.len = frame.payload_len;
+ frame.wsc->frag_buf.s[frame.wsc->frag_buf.len] = '\0';
+ wsconn_put(frame.wsc);
+ return 0;
+ }
}
else if (frame.wsc->sub_protocol == SUB_PROTOCOL_MSRP)
{