Mercurial > prosody-hg

comparison plugins/mod_websocket.lua @ 6397:6f75f8043936

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
mod_websocket: Initial commit (based on the prosody-modules version)
author Florian Zeitz <florob@babelmonkeys.de>
date Fri, 05 Sep 2014 03:47:39 +0200
parents
children 2bf1b7e2149a
comparison
equal deleted inserted replaced
6396:17b54f523796 6397:6f75f8043936
1 -- Prosody IM
2 -- Copyright (C) 2012-2014 Florian Zeitz
3 --
4 -- This project is MIT/X11 licensed. Please see the
5 -- COPYING file in the source package for more information.
6 --
7
8 module:set_global();
9
10 local add_filter = require "util.filters".add_filter;
11 local sha1 = require "util.hashes".sha1;
12 local base64 = require "util.encodings".base64.encode;
13 local st = require "util.stanza";
14 local parse_xml = require "util.xml".parse;
15 local portmanager = require "core.portmanager";
16 local sm_destroy_session = sessionmanager.destroy_session;
17 local log = module._log;
18
19 local websocket_frames = require"net.websocket.frames";
20 local parse_frame = websocket_frames.parse;
21 local build_frame = websocket_frames.build;
22 local build_close = websocket_frames.build_close;
23 local parse_close = websocket_frames.parse_close;
24
25 local t_concat = table.concat;
26
27 local consider_websocket_secure = module:get_option_boolean("consider_websocket_secure");
28 local cross_domain = module:get_option("cross_domain_websocket");
29 if cross_domain then
30 if cross_domain == true then
31 cross_domain = "*";
32 elseif type(cross_domain) == "table" then
33 cross_domain = t_concat(cross_domain, ", ");
34 end
35 if type(cross_domain) ~= "string" then
36 cross_domain = nil;
37 end
38 end
39
40 local xmlns_framing = "urn:ietf:params:xml:ns:xmpp-framing";
41 local xmlns_streams = "http://etherx.jabber.org/streams";
42 local xmlns_client = "jabber:client";
43 local stream_xmlns_attr = {xmlns='urn:ietf:params:xml:ns:xmpp-streams'};
44
45 module:depends("c2s")
46 local sessions = module:shared("c2s/sessions");
47 local c2s_listener = portmanager.get_service("c2s").listener;
48
49 --- Session methods
50 local function session_open_stream(session)
51 local attr = {
52 xmlns = xmlns_framing,
53 version = "1.0",
54 id = session.streamid or "",
55 from = session.host
56 };
57 session.send(st.stanza("open", attr));
58 end
59
60 local function session_close(session, reason)
61 local log = session.log or log;
62 if session.conn then
63 if session.notopen then
64 session:open_stream();
65 end
66 if reason then -- nil == no err, initiated by us, false == initiated by client
67 local stream_error = st.stanza("stream:error");
68 if type(reason) == "string" then -- assume stream error
69 stream_error:tag(reason, {xmlns = 'urn:ietf:params:xml:ns:xmpp-streams' });
70 elseif type(reason) == "table" then
71 if reason.condition then
72 stream_error:tag(reason.condition, stream_xmlns_attr):up();
73 if reason.text then
74 stream_error:tag("text", stream_xmlns_attr):text(reason.text):up();
75 end
76 if reason.extra then
77 stream_error:add_child(reason.extra);
78 end
79 elseif reason.name then -- a stanza
80 stream_error = reason;
81 end
82 end
83 log("debug", "Disconnecting client, <stream:error> is: %s", tostring(stream_error));
84 session.send(stream_error);
85 end
86
87 session.send(st.stanza("close", { xmlns = xmlns_framing }));
88 function session.send() return false; end
89
90 local reason = (reason and (reason.name or reason.text or reason.condition)) or reason;
91 session.log("debug", "c2s stream for %s closed: %s", session.full_jid or ("<"..session.ip..">"), reason or "session closed");
92
93 -- Authenticated incoming stream may still be sending us stanzas, so wait for </stream:stream> from remote
94 local conn = session.conn;
95 if reason == nil and not session.notopen and session.type == "c2s" then
96 -- Grace time to process data from authenticated cleanly-closed stream
97 add_task(stream_close_timeout, function ()
98 if not session.destroyed then
99 session.log("warn", "Failed to receive a stream close response, closing connection anyway...");
100 sm_destroy_session(session, reason);
101 conn:write(build_close(1000, "Stream closed"));
102 conn:close();
103 end
104 end);
105 else
106 sm_destroy_session(session, reason);
107 conn:write(build_close(1000, "Stream closed"));
108 conn:close();
109 end
110 end
111 end
112
113
114 --- Filters
115 local function filter_open_close(data)
116 if not data:find(xmlns_framing, 1, true) then return data; end
117
118 local oc = parse_xml(data);
119 if not oc then return data; end
120 if oc.attr.xmlns ~= xmlns_framing then return data; end
121 if oc.name == "close" then return "</stream:stream>"; end
122 if oc.name == "open" then
123 oc.name = "stream:stream";
124 oc.attr.xmlns = nil;
125 oc.attr["xmlns:stream"] = xmlns_streams;
126 return oc:top_tag();
127 end
128
129 return data;
130 end
131 function handle_request(event, path)
132 local request, response = event.request, event.response;
133 local conn = response.conn;
134
135 if not request.headers.sec_websocket_key then
136 response.headers.content_type = "text/html";
137 return [[<!DOCTYPE html><html><head><title>Websocket</title></head><body>
138 <p>It works! Now point your WebSocket client to this URL to connect to Prosody.</p>
139 </body></html>]];
140 end
141
142 local wants_xmpp = false;
143 (request.headers.sec_websocket_protocol or ""):gsub("([^,]*),?", function (proto)
144 if proto == "xmpp" then wants_xmpp = true; end
145 end);
146
147 if not wants_xmpp then
148 return 501;
149 end
150
151 local function websocket_close(code, message)
152 conn:write(build_close(code, message));
153 conn:close();
154 end
155
156 local dataBuffer;
157 local function handle_frame(frame)
158 local opcode = frame.opcode;
159 local length = frame.length;
160 module:log("debug", "Websocket received frame: opcode=%0x, %i bytes", frame.opcode, #frame.data);
161
162 -- Error cases
163 if frame.RSV1 or frame.RSV2 or frame.RSV3 then -- Reserved bits non zero
164 websocket_close(1002, "Reserved bits not zero");
165 return false;
166 end
167
168 if opcode == 0x8 then -- close frame
169 if length == 1 then
170 websocket_close(1002, "Close frame with payload, but too short for status code");
171 return false;
172 elseif length >= 2 then
173 local status_code = parse_close(frame.data)
174 if status_code < 1000 then
175 websocket_close(1002, "Closed with invalid status code");
176 return false;
177 elseif ((status_code > 1003 and status_code < 1007) or status_code > 1011) and status_code < 3000 then
178 websocket_close(1002, "Closed with reserved status code");
179 return false;
180 end
181 end
182 end
183
184 if opcode >= 0x8 then
185 if length > 125 then -- Control frame with too much payload
186 websocket_close(1002, "Payload too large");
187 return false;
188 end
189
190 if not frame.FIN then -- Fragmented control frame
191 websocket_close(1002, "Fragmented control frame");
192 return false;
193 end
194 end
195
196 if (opcode > 0x2 and opcode < 0x8) or (opcode > 0xA) then
197 websocket_close(1002, "Reserved opcode");
198 return false;
199 end
200
201 if opcode == 0x0 and not dataBuffer then
202 websocket_close(1002, "Unexpected continuation frame");
203 return false;
204 end
205
206 if (opcode == 0x1 or opcode == 0x2) and dataBuffer then
207 websocket_close(1002, "Continuation frame expected");
208 return false;
209 end
210
211 -- Valid cases
212 if opcode == 0x0 then -- Continuation frame
213 dataBuffer[#dataBuffer+1] = frame.data;
214 elseif opcode == 0x1 then -- Text frame
215 dataBuffer = {frame.data};
216 elseif opcode == 0x2 then -- Binary frame
217 websocket_close(1003, "Only text frames are supported");
218 return;
219 elseif opcode == 0x8 then -- Close request
220 websocket_close(1000, "Goodbye");
221 return;
222 elseif opcode == 0x9 then -- Ping frame
223 frame.opcode = 0xA;
224 conn:write(build_frame(frame));
225 return "";
226 elseif opcode == 0xA then -- Pong frame
227 module:log("warn", "Received unexpected pong frame: " .. tostring(frame.data));
228 return "";
229 else
230 log("warn", "Received frame with unsupported opcode %i", opcode);
231 return "";
232 end
233
234 if frame.FIN then
235 local data = t_concat(dataBuffer, "");
236 dataBuffer = nil;
237 return data;
238 end
239 return "";
240 end
241
242 conn:setlistener(c2s_listener);
243 c2s_listener.onconnect(conn);
244
245 local session = sessions[conn];
246
247 session.secure = consider_websocket_secure or session.secure;
248
249 session.open_stream = session_open_stream;
250 session.close = session_close;
251
252 local frameBuffer = "";
253 add_filter(session, "bytes/in", function(data)
254 local cache = {};
255 frameBuffer = frameBuffer .. data;
256 local frame, length = parse_frame(frameBuffer);
257
258 while frame do
259 frameBuffer = frameBuffer:sub(length + 1);
260 local result = handle_frame(frame);
261 if not result then return; end
262 cache[#cache+1] = filter_open_close(result);
263 frame, length = parse_frame(frameBuffer);
264 end
265 return t_concat(cache, "");
266 end);
267
268 add_filter(session, "stanzas/out", function(stanza)
269 local attr = stanza.attr;
270 attr.xmlns = attr.xmlns or xmlns_client;
271 if stanza.name:find("^stream:") then
272 attr["xmlns:stream"] = attr["xmlns:stream"] or xmlns_streams;
273 end
274 return stanza;
275 end);
276
277 add_filter(session, "bytes/out", function(data)
278 return build_frame({ FIN = true, opcode = 0x01, data = tostring(data)});
279 end);
280
281 response.status_code = 101;
282 response.headers.upgrade = "websocket";
283 response.headers.connection = "Upgrade";
284 response.headers.sec_webSocket_accept = base64(sha1(request.headers.sec_websocket_key .. "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"));
285 response.headers.sec_webSocket_protocol = "xmpp";
286 response.headers.access_control_allow_origin = cross_domain;
287
288 return "";
289 end
290
291 function module.add_host(module)
292 module:depends("http");
293 module:provides("http", {
294 name = "websocket";
295 default_path = "xmpp-websocket";
296 route = {
297 ["GET"] = handle_request;
298 ["GET /"] = handle_request;
299 };
300 });
301 end