From 81a7c44a05f6a31a044f4415e26e45699e8a2e55 Mon Sep 17 00:00:00 2001 From: zixun Date: Sat, 20 Jul 2019 06:02:36 -0700 Subject: [PATCH] add ws/wss server and client support --- examples/simplewebsocket.lua | 90 +++++++ lualib/http/httpc.lua | 77 +----- lualib/http/internal.lua | 81 ++++++ lualib/http/websocket.lua | 465 +++++++++++++++++++++++++++++++++++ 4 files changed, 637 insertions(+), 76 deletions(-) create mode 100755 examples/simplewebsocket.lua create mode 100755 lualib/http/websocket.lua diff --git a/examples/simplewebsocket.lua b/examples/simplewebsocket.lua new file mode 100755 index 00000000..1d798d95 --- /dev/null +++ b/examples/simplewebsocket.lua @@ -0,0 +1,90 @@ +local skynet = require "skynet" +local socket = require "skynet.socket" +local service = require "skynet.service" +local websocket = require "http.websocket" + +local handle = {} +local MODE = ... + +if MODE == "agent" then + function handle.connect(id) + print("ws connect from: " .. tostring(id)) + end + + function handle.handshake(id, header) + print("ws handshake from: " .. tostring(id)) + print("----header-----") + for k,v in pairs(header) do + print(k,v) + end + print("--------------") + end + + function handle.message(id, msg) + websocket.write(id, msg) + end + + function handle.ping(id) + print("ws ping from: " .. tostring(id) .. "\n") + end + + function handle.pong(id) + print("ws pong from: " .. tostring(id)) + end + + function handle.close(id, code, reason) + print("ws close from: " .. tostring(id), code, reason) + end + + function handle.error(id) + print("ws error from: " .. tostring(id)) + end + + skynet.start(function () + skynet.dispatch("lua", function (_,_, id, protocol) + websocket.accept(id, handle, protocol) + end) + end) + +else + local function simple_echo_client_service(protocol) + local skynet = require "skynet" + local websocket = require "http.websocket" + local url = string.format("%s://127.0.0.1:9948/", protocol) + local ws_id = websocket.connect(url) + while true do + local msg = "hello world!" + websocket.write(ws_id, msg) + print(">: " .. msg) + local resp, close_reason = websocket.read(ws_id) + print("<: " .. (resp and resp or "[Close] " .. close_reason)) + if not resp then + print("echo server close.") + break + end + websocket.ping(ws_id) + skynet.sleep(100) + end + end + + skynet.start(function () + local agent = {} + for i= 1, 20 do + agent[i] = skynet.newservice(SERVICE_NAME, "agent") + end + local balance = 1 + local protocol = "ws" + local id = socket.listen("0.0.0.0", 9948) + skynet.error(string.format("Listen websocket port 9948 protocol:%s", protocol)) + socket.start(id, function(id, addr) + print(string.format("accept client socket_id: %s addr:%s", id, addr)) + skynet.send(agent[balance], "lua", id, protocol) + balance = balance + 1 + if balance > #agent then + balance = 1 + end + end) + -- test echo client + service.new("websocket_echo_client", simple_echo_client_service, protocol) + end) +end \ No newline at end of file diff --git a/lualib/http/httpc.lua b/lualib/http/httpc.lua index a26a1505..da4e8b02 100644 --- a/lualib/http/httpc.lua +++ b/lualib/http/httpc.lua @@ -8,81 +8,6 @@ local table = table local httpc = {} -local function request(interface, method, host, url, recvheader, header, content) - local read = interface.read - local write = interface.write - local header_content = "" - if header then - if not header.host then - header.host = host - end - for k,v in pairs(header) do - header_content = string.format("%s%s:%s\r\n", header_content, k, v) - end - else - header_content = string.format("host:%s\r\n",host) - end - - if content then - local data = string.format("%s %s HTTP/1.1\r\n%scontent-length:%d\r\n\r\n", method, url, header_content, #content) - write(data) - write(content) - else - local request_header = string.format("%s %s HTTP/1.1\r\n%scontent-length:0\r\n\r\n", method, url, header_content) - write(request_header) - end - - local tmpline = {} - local body = internal.recvheader(read, tmpline, "") - if not body then - error(socket.socket_error) - end - - local statusline = tmpline[1] - local code, info = statusline:match "HTTP/[%d%.]+%s+([%d]+)%s+(.*)$" - code = assert(tonumber(code)) - - local header = internal.parseheader(tmpline,2,recvheader or {}) - if not header then - error("Invalid HTTP response header") - end - - local length = header["content-length"] - if length then - length = tonumber(length) - end - local mode = header["transfer-encoding"] - if mode then - if mode ~= "identity" and mode ~= "chunked" then - error ("Unsupport transfer-encoding") - end - end - - if mode == "chunked" then - body, header = internal.recvchunkedbody(read, nil, header, body) - if not body then - error("Invalid response body") - end - else - -- identity mode - if length then - if #body >= length then - body = body:sub(1,length) - else - local padding = read(length - #body) - body = body .. padding - end - elseif code == 204 or code == 304 or code < 200 then - body = "" - -- See https://stackoverflow.com/questions/15991173/is-the-content-length-header-required-for-a-http-1-0-response - else - -- no content-length, read all - body = body .. interface.readall() - end - end - - return code, body -end local async_dns @@ -172,7 +97,7 @@ function httpc.request(method, host, url, recvheader, header, content) if interface.init then interface.init() end - local ok , statuscode, body = pcall(request, interface, method, host, url, recvheader, header, content) + local ok , statuscode, body = pcall(internal.request, interface, method, host, url, recvheader, header, content) finish = true socket.close(fd) if interface.close then diff --git a/lualib/http/internal.lua b/lualib/http/internal.lua index 6f76b86d..ec45ff50 100644 --- a/lualib/http/internal.lua +++ b/lualib/http/internal.lua @@ -141,4 +141,85 @@ function M.recvchunkedbody(readbytes, bodylimit, header, body) return result, header end + +function M.request(interface, method, host, url, recvheader, header, content) + local is_ws = interface.websocket + local read = interface.read + local write = interface.write + local header_content = "" + if header then + if not header.host then + header.host = host + end + for k,v in pairs(header) do + header_content = string.format("%s%s:%s\r\n", header_content, k, v) + end + else + header_content = string.format("host:%s\r\n",host) + end + + if content then + local data = string.format("%s %s HTTP/1.1\r\n%scontent-length:%d\r\n\r\n", method, url, header_content, #content) + write(data) + write(content) + else + local request_header = string.format("%s %s HTTP/1.1\r\n%scontent-length:0\r\n\r\n", method, url, header_content) + write(request_header) + end + + local tmpline = {} + local body = M.recvheader(read, tmpline, "") + if not body then + error(socket.socket_error) + end + + local statusline = tmpline[1] + local code, info = statusline:match "HTTP/[%d%.]+%s+([%d]+)%s+(.*)$" + code = assert(tonumber(code)) + + local header = M.parseheader(tmpline,2,recvheader or {}) + if not header then + error("Invalid HTTP response header") + end + + local length = header["content-length"] + if length then + length = tonumber(length) + end + local mode = header["transfer-encoding"] + if mode then + if mode ~= "identity" and mode ~= "chunked" then + error ("Unsupport transfer-encoding") + end + end + + if mode == "chunked" then + body, header = M.recvchunkedbody(read, nil, header, body) + if not body then + error("Invalid response body") + end + else + -- identity mode + if length then + if #body >= length then + body = body:sub(1,length) + else + local padding = read(length - #body) + body = body .. padding + end + elseif code == 204 or code == 304 or code < 200 then + body = "" + -- See https://stackoverflow.com/questions/15991173/is-the-content-length-header-required-for-a-http-1-0-response + elseif is_ws and code == 101 then + -- if websocket handshake success + return code, body + else + -- no content-length, read all + body = body .. interface.readall() + end + end + + return code, body +end + return M diff --git a/lualib/http/websocket.lua b/lualib/http/websocket.lua new file mode 100755 index 00000000..81f8d96a --- /dev/null +++ b/lualib/http/websocket.lua @@ -0,0 +1,465 @@ +local internal = require "http.internal" +local socket = require "skynet.socket" +local crypt = require "skynet.crypt" +local httpd = require "http.httpd" +local skynet = require "skynet" +local sockethelper = require "http.sockethelper" +local socket_error = sockethelper.socket_error + +local GLOBAL_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" + +local M = {} +local ws_pool = {} + + +local function write_handshake(self, host, url, header) + local key = crypt.base64encode(crypt.randomkey()..crypt.randomkey()) + local request_header = { + ["Upgrade"] = "websocket", + ["Connection"] = "Upgrade", + ["Sec-WebSocket-Version"] = "13", + ["Sec-WebSocket-Key"] = key + } + if header then + for k,v in pairs(header) do + assert(request_header[k] == nil, k) + request_header[k] = v + end + end + + local recvheader = {} + local code, body = internal.request(self, "GET", host, url, recvheader, request_header) + if code ~= 101 then + error(string.format("websocket handshake error: code[%s] info:%s", code, body)) + end + + if not recvheader["upgrade"] or recvheader["upgrade"]:lower() ~= "websocket" then + error("websocket handshake upgrade must websocket") + end + + if not recvheader["connection"] or recvheader["connection"]:lower() ~= "upgrade" then + error("websocket handshake connection must upgrade") + end + + local sw_key = recvheader["sec-websocket-accept"] + if not sw_key then + error("websocket handshake need Sec-WebSocket-Accept") + end + + local guid = self.guid + sw_key = crypt.base64decode(sw_key) + if sw_key ~= crypt.sha1(key .. guid) then + error("websocket handshake invalid Sec-WebSocket-Accept") + end +end + + +local function read_handshake(self) + local tmpline = {} + local header_body = internal.recvheader(self.read, tmpline, "") + if not header_body then + return 413 + end + + local request = assert(tmpline[1]) + local method, url, httpver = request:match "^(%a+)%s+(.-)%s+HTTP/([%d%.]+)$" + assert(method and url and httpver) + if method:lower() ~= "get" then + return 400, "need GET method" + end + + httpver = assert(tonumber(httpver)) + if httpver < 1.0 or httpver > 1.1 then + return 505 -- HTTP Version not supported + end + + local header = internal.parseheader(tmpline, 2, {}) + if not header then + return 400 -- Bad request + end + if not header["upgrade"] or header["upgrade"]:lower() ~= "websocket" then + return 426, "Upgrade Required" + end + + if not header["host"] then + return 400, "host Required" + end + + if not header["connection"] or header["connection"]:lower() ~= "upgrade" then + return 400, "Connection must Upgrade" + end + + local sw_key = header["sec-websocket-key"] + if not sw_key then + return 400, "Sec-WebSocket-Key Required" + else + local raw_key = crypt.base64decode(sw_key) + if #raw_key ~= 16 then + return 400, "Sec-WebSocket-Key invalid" + end + end + + if not header["sec-websocket-version"] or header["sec-websocket-version"] ~= "13" then + return 400, "Sec-WebSocket-Version must 13" + end + + local sw_protocol = header["sec-websocket-protocol"] + local sub_pro = "" + if sw_protocol then + for sub_protocol in string.gmatch(sw_protocol, "[^%s,]+") do + if sub_protocol == "chat" then + sub_pro = "Sec-WebSocket-Protocol: chat\r\n" + has_chat = true + break + end + end + if not has_chat then + return 400, "Sec-WebSocket-Protocol need include chat" + end + end + + -- response handshake + local accept = crypt.base64encode(crypt.sha1(sw_key .. self.guid)) + local resp = "HTTP/1.1 101 Switching Protocols\r\n".. + "Upgrade: websocket\r\n".. + "Connection: Upgrade\r\n".. + string.format("Sec-WebSocket-Accept: %s\r\n", accept).. + sub_pro .. + "\r\n" + self.write(resp) + return nil, header +end + +local function try_handle(self, method, ...) + local handle = self.handle + local f = handle and handle[method] + if f then + f(self.id, ...) + end +end + +local op_code = { + ["frame"] = 0x00, + ["text"] = 0x01, + ["binary"] = 0x02, + ["close"] = 0x08, + ["ping"] = 0x09, + ["pong"] = 0x0A, + [0x00] = "frame", + [0x01] = "text", + [0x02] = "binary", + [0x08] = "close", + [0x09] = "ping", + [0x0A] = "pong", +} + +local function write_frame(self, op, payload_data) + payload_data = payload_data or "" + local payload_len = #payload_data + local send_buf = {} + local op_v = assert(op_code[op]) + local v1 = 0x80 | op_v -- fin is 1 with opcode + local s + -- mask set to 0 + if payload_len < 126 then + s = string.pack("I1I1", v1, payload_len) + elseif payload_len < 0xffff then + s = string.pack("I1I1>I2", v1, 126, payload_len) + else + s = string.pack("I1I1>I8", v1, 127, payload_len) + end + + self.write(s) + if payload_len > 0 then + self.write(payload_data) + end +end + + +local function read_frame(self) + local s = self.read(2) + local v1, v2 = string.unpack("I1I1", s) + local fin = (v1 & 0x80) ~= 0 + local rsv1 = (v1 & 0x40) ~= 0 + local rsv2 = (v1 & 0x20) ~= 0 + local rsv3 = (v1 & 0x10) ~= 0 + local op = v1 & 0x0f + local mask = (v2 & 0x80) ~= 0 + local payload_len = (v2 & 0x7f) + if payload_len == 126 then + s = self.read(2) + payload_len = string.unpack(">I2", s) + elseif payload_len == 127 then + s = self.read(8) + payload_len = string.unpack(">I8", s) + end + + -- print("fin, rsv1, rsv2, rsv3, op, mask, payload_len", + -- fin, rsv1, rsv2, rsv3, op, mask, payload_len) + local masking_key + if mask then + s = self.read(4) + local k1, k2, k3, k4 = string.unpack("I1I1I1I1", s) + masking_key = {k1, k2, k3, k4} + end + + local payload_data = payload_len>0 and self.read(payload_len) or "" + if masking_key then + local t = {} + local len = #payload_data + for i=1, len do + local c = string.byte(payload_data, i) + local m = masking_key[(i-1) % 4 + 1] + local v = c ~ m + t[i] = string.char(v) + end + payload_data = table.concat(t) + end + return fin, assert(op_code[op]), payload_data +end + + +local function resolve_accept(self) + try_handle(self, "connect") + local code, err = read_handshake(self) + if code then + local ok, s = httpd.write_response(self.write, code, err) + if not ok then + error(s) + end + end + + local header = err + try_handle(self, "handshake", header) + local recv_buf = {} + while true do + local fin, op, payload_data = read_frame(self) + if op == "close" then + local code, reason + local payload_len = #payload_data + if payload_len > 2 then + local fmt = string.format(">I2s[%d]", payload_len - 2) + code, reason = string.unpack(fmt, payload_data) + end + try_handle(self, "close", code, reason) + break + elseif op == "ping" then + write_frame(self, "pong") + try_handle(self, "ping") + elseif op == "pong" then + try_handle(self, "pong") + else + if fin and #recv_buf == 0 then + try_handle(self, "message", payload_data) + else + recv_buf[#recv_buf+1] = payload_data + if fin then + local s = table.concat(recv_buf) + try_handle(self, "message", s) + recv_buf = {} -- clear recv_buf + end + end + end + end +end + + +local SSLCTX_CLIENT = nil +local function _new_client_ws(socket_id, protocol) + local obj + if protocol == "ws" then + obj = { + websocket = true, + close = function () + socket.close(socket_id) + end, + read = sockethelper.readfunc(socket_id), + write = sockethelper.writefunc(socket_id), + readall = function () + return socket.readall(socket_id) + end, + } + elseif protocol == "wss" then + local tls = require "http.tlshelper" + SSLCTX_CLIENT = SSLCTX_CLIENT or tls.newctx() + local tls_ctx = tls.newtls("client", SSLCTX_CLIENT) + local init = tls.init_requestfunc(socket_id, tls_ctx) + init() + obj = { + websocket = true, + close = function () + socket.close(socket_id) + tls.closefunc(tls_ctx) + end, + read = tls.readfunc(socket_id, tls_ctx), + write = tls.writefunc(socket_id, tls_ctx), + readall = tls.readallfunc(socket_id, tls_ctx), + } + else + error(string.format("invalid websocket protocol:%s", tostring(protocol))) + end + obj.id = assert(socket_id) + obj.guid = GLOBAL_GUID + ws_pool[socket_id] = obj + return obj +end + + +local SSLCTX_SERVER = nil +local function _new_server_ws(socket_id, handle, protocol) + local obj + if protocol == "ws" then + obj = { + close = function () + socket.close(socket_id) + end, + read = sockethelper.readfunc(socket_id), + write = sockethelper.writefunc(socket_id), + } + + elseif protocol == "wss" then + local tls = require "http.tlshelper" + if not SSLCTX_SERVER then + SSLCTX_SERVER = tls.newctx() + -- gen cert and key + -- openssl req -x509 -newkey rsa:2048 -days 3650 -nodes -keyout server-key.pem -out server-cert.pem + local certfile = skynet.getenv("certfile") or "./server-cert.pem" + local keyfile = skynet.getenv("keyfile") or "./server-key.pem" + SSLCTX_SERVER:set_cert(certfile, keyfile) + end + local tls_ctx = tls.newtls("server", SSLCTX_SERVER) + local init = tls.init_responsefunc(socket_id, tls_ctx) + init() + obj = { + close = function () + socket.close(socket_id) + tls.closefunc(tls_ctx) + end, + read = tls.readfunc(socket_id, tls_ctx), + write = tls.writefunc(socket_id, tls_ctx), + } + + else + error(string.format("invalid websocket protocol:%s", tostring(protocol))) + end + + obj.id = assert(socket_id) + obj.handle = handle + obj.guid = GLOBAL_GUID + ws_pool[socket_id] = obj + return obj +end + + +local function _close_websocket(ws_obj) + local id = ws_obj.id + assert(ws_pool[id] == ws_obj) + ws_pool[id] = nil + ws_obj.close() +end + + +-- handle interface +-- connect / handshake / message / ping / pong / close / error +function M.accept(socket_id, handle, protocol) + socket.start(socket_id) + protocol = protocol or "ws" + local ws_obj = _new_server_ws(socket_id, handle, protocol) + local on_warning = handle and handle["warning"] + if on_warning then + socket.warning(socket_id, function (id, sz) + on_warning(ws_obj, sz) + end) + end + + local ok, err = xpcall(resolve_accept, debug.traceback, ws_obj) + _close_websocket(ws_obj) + if not ok then + if err == socket_error then + try_handle(ws_obj, "error", ws_obj) + else + error(err) + end + end +end + + +function M.connect(url, header) + local protocol, host, uri = string.match(url, "^(wss?)://([^/]+)(.*)$") + if protocol ~= "wss" and protocol ~= "ws" then + error(string.format("invalid protocol: %s", protocol)) + end + + assert(host) + local host_name, host_port = string.match(host, "^([^:]+):?(%d*)$") + assert(host_name and host_port) + if host_port == "" then + host_port = protocol == "ws" and 80 or 443 + end + + uri = uri == "" and "/" or uri + local socket_id = socket.open(host_name, host_port) + assert(socket_id) + local ws_obj = _new_client_ws(socket_id, protocol) + write_handshake(ws_obj, host_name, uri, header) + return socket_id +end + + +function M.read(id) + local ws_obj = assert(ws_pool[id]) + local recv_buf + while true do + local fin, op, payload_data = read_frame(ws_obj) + if op == "close" then + _close_websocket(ws_obj) + return false, payload_data + elseif op == "ping" then + write_frame(ws_obj, "pong") + elseif op ~= "pong" then -- op is frame, text binary + if fin and not recv_buf then + return payload_data + else + recv_buf = recv_buf or {} + recv_buf[#recv_buf+1] = payload_data + if fin then + local s = table.concat(recv_buf) + return s + end + end + end + end + assert(false) +end + + +function M.write(id, data, fmt) + local ws_obj = assert(ws_pool[id]) + fmt = fmt or "text" + assert(fmt == "text" or fmt == "binary") + write_frame(ws_obj, fmt, data) +end + + +function M.ping(id) + local ws_obj = assert(ws_pool[id]) + write_frame(ws_obj, "ping") +end + + +function M.close(id, code ,reason) + local ws_obj = ws_pool[id] + if not ws_obj then + return + end + + pcall(function () + reason = reason or "" + local payload_data = code and string.pack(">I2s", code, reason) or nil + write_frame(ws_obj, "close", payload_data) + end) + _close_websocket(ws_obj) +end + + +return M \ No newline at end of file