From c4bc8fbd5f67e2d5747ed8b412ca319c40afcc99 Mon Sep 17 00:00:00 2001 From: Zeping Bu Date: Sun, 13 Sep 2026 20:03:12 +0800 Subject: [PATCH 1/3] feat: port resty.websocket.proxy from Kong/lua-resty-websocket-proxy Import lib/resty/websocket/proxy.lua and its test suite from Kong/lua-resty-websocket-proxy@6564d0d (https://github.com/Kong/lua-resty-websocket-proxy), a reverse-proxying helper built on resty.websocket.client/server. It forwards WebSocket frames between a client and an upstream server, with support for frame inspection/rewriting via an on_frame callback, fragmented frame aggregation, and per-direction frame size/count limits. This is a separate repository with no shared history with lua-resty-websocket, so the code is brought in as a single import commit rather than cherry-picked commit by commit; the Co-authored-by trailers below credit everyone who authored a commit in the source repository. This file and its tests remain licensed under the Apache License, Version 2.0 (see LICENSE-APACHE), unlike the rest of this BSD-2-Clause-licensed repository; this exception is noted in README.markdown. The only functional change from the upstream source is in t/Tests.pm, which no longer references the misc/lua-resty-websocket git submodule path, since this repository already provides resty.websocket.client/server directly. Co-authored-by: Michael Martin <3277009+flrgh@users.noreply.github.com> Co-authored-by: Michael Martin Co-authored-by: Thibault Charbonnier Co-authored-by: Wietse Wind --- LICENSE-APACHE | 201 +++++++++ README.markdown | 2 + lib/resty/websocket/proxy.lua | 538 ++++++++++++++++++++++++ t/01-basic_proxy.t | 410 ++++++++++++++++++ t/02-timeouts.t | 147 +++++++ t/03-on_frame_callback.t | 453 ++++++++++++++++++++ t/04-fragmented_frames.t | 762 ++++++++++++++++++++++++++++++++++ t/05-wss.t | 92 ++++ t/06-error_handling.t | 86 ++++ t/07-invalid_usage.t | 202 +++++++++ t/08-limits.t | 565 +++++++++++++++++++++++++ t/Tests.pm | 117 ++++++ t/certs/cert.pem | 19 + t/certs/key.pem | 28 ++ 14 files changed, 3622 insertions(+) create mode 100644 LICENSE-APACHE create mode 100644 lib/resty/websocket/proxy.lua create mode 100644 t/01-basic_proxy.t create mode 100644 t/02-timeouts.t create mode 100644 t/03-on_frame_callback.t create mode 100644 t/04-fragmented_frames.t create mode 100644 t/05-wss.t create mode 100644 t/06-error_handling.t create mode 100644 t/07-invalid_usage.t create mode 100644 t/08-limits.t create mode 100644 t/Tests.pm create mode 100644 t/certs/cert.pem create mode 100644 t/certs/key.pem diff --git a/LICENSE-APACHE b/LICENSE-APACHE new file mode 100644 index 0000000..8dada3e --- /dev/null +++ b/LICENSE-APACHE @@ -0,0 +1,201 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "{}" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright {yyyy} {name of copyright owner} + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/README.markdown b/README.markdown index af3a1dc..5fc6ee8 100644 --- a/README.markdown +++ b/README.markdown @@ -715,6 +715,8 @@ Redistribution and use in source and binary forms, with or without modification, THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +`lib/resty/websocket/proxy.lua` and its accompanying tests (`t/01-basic_proxy.t` through `t/08-limits.t`, `t/Tests.pm`, `t/certs/`) are ported from [Kong/lua-resty-websocket-proxy](https://github.com/Kong/lua-resty-websocket-proxy) and remain licensed under the Apache License, Version 2.0, Copyright 2022 Kong Inc. The full license text is available at [LICENSE-APACHE](LICENSE-APACHE). No other part of this repository is affected by this exception. + [Back to TOC](#table-of-contents) See Also diff --git a/lib/resty/websocket/proxy.lua b/lib/resty/websocket/proxy.lua new file mode 100644 index 0000000..fc212cb --- /dev/null +++ b/lib/resty/websocket/proxy.lua @@ -0,0 +1,538 @@ +-- Ported from Kong/lua-resty-websocket-proxy (https://github.com/Kong/lua-resty-websocket-proxy). +-- +-- Copyright 2022 Kong Inc. +-- +-- Licensed under the Apache License, Version 2.0 (the "License"); you may +-- not use this file except in compliance with the License. You may obtain +-- a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, software +-- distributed under the License is distributed on an "AS IS" BASIS, +-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +-- See the License for the specific language governing permissions and +-- limitations under the License. +-- +-- The full text of the License is also available at LICENSE-APACHE in the +-- root of this repository. Unlike the rest of this repository (licensed +-- under the 2-clause BSD license, see README.markdown), this file and its +-- accompanying tests remain under the Apache License, Version 2.0. + +local new_tab = require "table.new" +local clear_tab = require "table.clear" +local ws_client = require "resty.websocket.client" +local ws_server = require "resty.websocket.server" + + +local type = type +local setmetatable = setmetatable +local insert = table.insert +local concat = table.concat +local yield = coroutine.yield +local fmt = string.format +local sub = string.sub +local gsub = string.gsub +local find = string.find +local log = ngx.log + + +local _DEBUG_PAYLOAD_MAX_LEN = 24 +local _STATES = { + INIT = 1, + ESTABLISHED = 2, + CLOSING = 3, +} + +local _TYP2OPCODE = { + ["continuation"] = 0x0, + ["text"] = 0x1, + ["binary"] = 0x2, + ["close"] = 0x8, + ["ping"] = 0x9, + ["pong"] = 0xa, +} + + +local _M = { + _VERSION = "0.0.1", +} + +local _mt = { __index = _M } + + +function _M.new(opts) + if opts == nil then + opts = new_tab(0, 0) + end + + if type(opts) ~= "table" then + error("opts must be a table", 2) + end + + if opts.on_frame ~= nil and type(opts.on_frame) ~= "function" then + error("opts.on_frame must be a function", 2) + end + + if opts.recv_timeout ~= nil and type(opts.recv_timeout) ~= "number" then + error("opts.recv_timeout must be a number", 2) + end + + if opts.client_max_frame_size ~= nil + and (type(opts.client_max_frame_size) ~= "number" + or opts.client_max_frame_size < 1) + then + error("opts.client_max_frame_size must be a number >= 1", 2) + end + + if opts.client_max_fragments ~= nil + and (type(opts.client_max_fragments) ~= "number" + or opts.client_max_fragments < 1) + then + error("opts.client_max_fragments must be a number >= 1", 2) + end + + if opts.upstream_max_frame_size ~= nil + and (type(opts.upstream_max_frame_size) ~= "number" + or opts.upstream_max_frame_size < 1) + then + error("opts.upstream_max_frame_size must be a number >= 1", 2) + end + + if opts.upstream_max_fragments ~= nil + and (type(opts.upstream_max_fragments) ~= "number" + or opts.upstream_max_fragments < 1) + then + error("opts.upstream_max_fragments must be a number >= 1", 2) + end + + + -- TODO: provide a means of passing options through to the + -- resty.websocket.client constructor (like `max_payload_len`) + local client, err = ws_client:new() + if not client then + return nil, "failed to create client: " .. err + end + + local self = { + client = client, + server = nil, + upstream_uri = nil, + on_frame = opts.on_frame, + recv_timeout = opts.recv_timeout, + client_max_frame_size = opts.client_max_frame_size, + client_max_fragments = opts.client_max_fragments, + upstream_max_frame_size = opts.upstream_max_frame_size, + upstream_max_fragments = opts.upstream_max_fragments, + aggregate_fragments = opts.aggregate_fragments, + debug = opts.debug, + client_state = _STATES.INIT, + upstream_state = _STATES.INIT, + co_client = nil, + co_server = nil, + } + + return setmetatable(self, _mt) +end + + +function _M:dd(...) + if self.debug then + return log(ngx.DEBUG, ...) + end +end + + +local function send_close_frame(self, ws, role, code, data) + self:dd(role, fmt(" closing:\ncode: %s\nreason: %q", code, data)) + + local ok, err = ws:send_close(code, data) + if not ok then + log(ngx.ERR, "failed sending close frame to ", role, ": ", err) + end +end + + +local function close(self, role, code, data, peer_code, peer_data) + local self_ws, peer_ws + local self_state = role .. "_state" + local peer + + if role == "client" then + self_ws = self.server + peer_ws = self.client + peer = "upstream" + else + -- role == "upstream" + self_ws = self.client + peer_ws = self.server + peer = "client" + end + + send_close_frame(self, self_ws, role, code, data) + self[self_state] = _STATES.CLOSING + send_close_frame(self, peer_ws, peer, peer_code, peer_data) +end + + +local function forwarder(self, ctx) + local role = ctx.role + local buf = ctx.buf + local self_ws, peer_ws + local self_state, peer_state + local frame_typ + local frame_size, frame_count = 0, 0 + local on_frame = self.on_frame + local max_frame_size = ctx.max_frame_size + local max_fragments = ctx.max_fragments + + self_state = role .. "_state" + + --assert(self[self_state] == _STATES.ESTABLISHED) + + if role == "client" then + self_ws = self.server + peer_ws = self.client + peer_state = "upstream_state" + + else + -- role == "upstream" + self_ws = self.client + peer_ws = self.server + peer_state = "client_state" + end + + while true do + if self[peer_state] == _STATES.CLOSING then + return + end + + if self.recv_timeout then + self_ws:set_timeout(self.recv_timeout) + end + + self:dd(role, " receiving frame...") + + local data, typ, err = self_ws:recv_frame() + if not data then + if find(err, "timeout", 1, true) then + log(ngx.INFO, fmt("timeout receiving frame from %s, reopening", + role)) + -- continue + + elseif find(err, "closed", 1, true) then + self[self_state] = _STATES.CLOSING + return role + + else + log(ngx.ERR, fmt("failed receiving frame from %s: %s", + role, err)) + self[self_state] = _STATES.CLOSING + return role, err + end + end + + -- special flags + + local code + local opcode = _TYP2OPCODE[typ] + local fin = true + if err == "again" then + fin = false + err = nil + end + + if typ then + if not opcode then + log(ngx.EMERG, "NYI - unknown frame type: ", typ, + " (dropping connection)") + return + end + + if typ == "close" then + code = err + end + + -- debug + + if self.debug and (not err or typ == "close") then + local extra = "" + local arrow + + if typ == "close" then + arrow = role == "client" and "--x" or "x--" + + else + arrow = role == "client" and "-->" or "<--" + end + + local payload = data and gsub(data, "\n", "\\n") or "" + if #payload > _DEBUG_PAYLOAD_MAX_LEN then + payload = sub(payload, 1, _DEBUG_PAYLOAD_MAX_LEN) .. "[...]" + end + + if code then + extra = fmt("\n code: %d", code) + end + + if frame_typ then + extra = fmt("\n initial type: \"%s\"", frame_typ) + end + + self:dd(fmt("\n[frame] downstream %s resty.proxy %s upstream\n" .. + " aggregating: %s\n" .. + " type: \"%s\"%s\n" .. + " payload: %s (len: %d)\n" .. + " fin: %s", + arrow, arrow, + self.aggregate_fragments, + typ, extra, + fmt("%q", payload), data and #data or 0, + fin)) + end + + local bytes + local forward = true + local data_frame = typ == "text" + or typ == "binary" + or typ == "continuation" + + -- limits + + if data_frame then + frame_size = frame_size + #data + + if max_frame_size and frame_size > max_frame_size then + log(ngx.INFO, fmt("%s frame size (%s) exceeds limit, closing", + role, frame_size)) + close(self, role, 1009, "Payload Too Large", 1001, "") + + return role + end + + frame_count = frame_count + 1 + + if max_fragments and frame_count > max_fragments then + log(ngx.INFO, fmt("%s frame count (%s) exceeds limit, closing", + role, frame_count)) + close(self, role, 1009, "Payload Too Large", 1001, "") + + return role + end + end + + + -- fragmentation + + if self.aggregate_fragments and data_frame then + if not fin then + self:dd(role, " received fragmented frame, buffering") + insert(buf, data) + forward = false + + -- stash data frame type of initial fragment + frame_typ = frame_typ or typ + + -- continue + + elseif #buf > 0 then + self:dd(role, " received last fragmented frame, forwarding") + insert(buf, data) + data = concat(buf, "") + clear_tab(buf) + + -- restore initial fragment type and opcode + typ = frame_typ + frame_typ = nil + opcode = _TYP2OPCODE[typ] + end + end + + -- forward + + if forward then + + -- callback + + if on_frame then + local updated, updated_code = on_frame(self, role, typ, + data, fin, code) + if updated ~= nil then + if type(updated) ~= "string" then + error("opts.on_frame return value must be " .. + "nil or a string") + end + end + + data = updated + + if typ == "close" and updated_code ~= nil then + if type(updated_code) ~= "number" then + error("opts.on_frame status code return value " .. + "must be nil or a number") + end + + code = updated_code + end + end + + if on_frame and data == nil then + self:dd(role, " dropping ", typ, " frame after on_frame handler requested it") + + -- continue: while true + + else + if typ == "close" then + log(ngx.INFO, "forwarding close with code: ", code, ", payload: ", + data) + + bytes, err = peer_ws:send_close(code, data) + + else + bytes, err = peer_ws:send_frame(fin, opcode, data) + end + + if not bytes then + log(ngx.ERR, fmt("failed forwarding a frame from %s: %s", + role, err)) + -- continue + end + end + + + if data_frame then + frame_size = 0 + frame_count = 0 + end + end + + -- continue: while true + end + + self:dd(role, " yielding") + + yield(self) + end +end + + +function _M:connect_upstream(uri, opts) + if self.upstream_state == _STATES.ESTABLISHED then + log(ngx.WARN, fmt("connection with upstream at %q already established", + self.upstream_uri)) + return true + end + + self:dd("connecting to \"", uri, "\" upstream") + + local ok, err, res = self.client:connect(uri, opts) + if not ok then + return nil, err + end + + self:dd("connected to \"", uri, "\" upstream") + + self.upstream_uri = uri + self.upstream_state = _STATES.ESTABLISHED + + return true, nil, res +end + + +function _M:connect_client() + if self.client_state == _STATES.ESTABLISHED then + log(ngx.WARN, "client handshake already completed") + return true + end + + self:dd("completing client handshake") + + local server, err = ws_server:new() + if not server then + return nil, err + end + + self:dd("completed client handshake") + + self.server = server + self.client_state = _STATES.ESTABLISHED + + return true +end + + +function _M:connect(uri, upstream_opts) + local ok, err = self:connect_upstream(uri, upstream_opts) + if not ok then + return nil, "failed connecting to upstream: " .. err + end + + ok, err = self:connect_client() + if not ok then + return nil, "failed client handshake: " .. err + end + + return true +end + + +function _M:execute() + if self.client_state ~= _STATES.ESTABLISHED then + return nil, "client handshake not complete" + end + + if self.upstream_state ~= _STATES.ESTABLISHED then + return nil, "upstream connection not established" + end + + self.co_client = ngx.thread.spawn(forwarder, self, { + role = "client", + buf = new_tab(0, 0), + max_frame_size = self.client_max_frame_size, + max_fragments = self.client_max_fragments, + }) + + self.co_server = ngx.thread.spawn(forwarder, self, { + role = "upstream", + buf = new_tab(0, 0), + max_frame_size = self.upstream_max_frame_size, + max_fragments = self.upstream_max_fragments, + }) + + local ok, res, err = ngx.thread.wait(self.co_client, self.co_server) + if not ok then + log(ngx.ERR, "failed to wait for websocket proxy threads: ", err) + + elseif res == "client" then + --assert(self.client_state == _STATES.CLOSING) + + self:dd(res, " thread terminated, killing server thread") + + ngx.thread.kill(self.co_server) + + self:dd("closing \"", self.upstream_uri, "\" upstream websocket") + + self.client:close() + + elseif res == "upstream" then + --assert(self.upstream_state == _STATES.CLOSING) + + self:dd(res, " thread terminated, killing client thread") + + ngx.thread.kill(self.co_client) + end + + self.co_client = nil + self.co_server = nil + self.client_state = _STATES.INIT + self.upstream_state = _STATES.INIT + + if err then + return nil, err + end + + return true +end + + +return _M diff --git a/t/01-basic_proxy.t b/t/01-basic_proxy.t new file mode 100644 index 0000000..3059531 --- /dev/null +++ b/t/01-basic_proxy.t @@ -0,0 +1,410 @@ +# vim:set ts=4 sts=4 sw=4 et ft=: + +use lib '.'; +use t::Tests; + +plan tests => repeat_each() * (blocks() * 4); + +run_tests(); + +__DATA__ + +=== TEST 1: forward a text frame back and forth +--- http_config eval: $t::Tests::HttpConfig +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new() + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wp:connect(proxy._tests.echo) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_text("hello world!")) + local data = assert(wb:recv_frame()) + ngx.say(data) + } + } +--- response_body +hello world! +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/frame type: text, payload: "hello world!"/ +--- no_error_log +[error] + + + +=== TEST 2: forward a ping/pong exchange +--- http_config eval: $t::Tests::HttpConfig +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new() + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wp:connect(proxy._tests.pong) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_ping("heartbeat client")) + local data, opcode = assert(wb:recv_frame()) + ngx.say(opcode, ": ", data) + } + } +--- response_body +pong: heartbeat server +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/frame type: ping, payload: "heartbeat client"/ +--- no_error_log +[error] + + + +=== TEST 3: forward a binary frame back and forth +--- http_config eval: $t::Tests::HttpConfig +--- config + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + local data, typ, err = wb:recv_frame() + if not data then + ngx.log(ngx.ERR, "failed receiving frame: ", err) + return ngx.exit(444) + end + + ngx.log(ngx.INFO, "frame type: ", typ, ", payload: \"", data, "\"") + + local bytes, err = wb:send_binary(data) + if not bytes then + ngx.log(ngx.ERR, "failed sending frame: ", err) + return ngx.exit(444) + end + } + } + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new() + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect(uri) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_binary("你好, WebSocket!")) + local data, opcode = assert(wb:recv_frame()) + ngx.say(opcode, ": ", data) + } + } +--- response_body +binary: 你好, WebSocket! +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/frame type: binary, payload: "你好, WebSocket!"/ +--- no_error_log +[error] + + + +=== TEST 4: forward close frame exchange from the client +--- http_config eval: $t::Tests::HttpConfig +--- config + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + local data, typ, err = wb:recv_frame() + if not data then + ngx.log(ngx.ERR, "failed receiving frame: ", err) + return ngx.exit(444) + end + + ngx.log(ngx.INFO, "frame type: ", typ, + ", code: ", err, + ", payload: \"", data, "\"") + + local bytes, err = wb:send_close() + if not bytes then + ngx.log(ngx.ERR, "failed sending close frame: ", err) + return ngx.exit(444) + end + } + } + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new() + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect(uri) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_close(1000, "goodbye")) + local data, opcode = assert(wb:recv_frame()) + ngx.say(opcode, ": ", data) + wb:close() + } + } +--- ignore_response_body +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/frame type: close, code: 1000, payload: "goodbye"/ +--- no_error_log +[crit] +[error] + + + +=== TEST 5: forward close frame exchange from upstream +--- http_config eval: $t::Tests::HttpConfig +--- config + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + local bytes, err = wb:send_close() + if not bytes then + ngx.log(ngx.ERR, "failed sending close frame: ", err) + return ngx.exit(444) + end + } + } + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new() + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect(uri) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + local data, opcode = assert(wb:recv_frame()) + ngx.say(opcode) + } + } +--- response_body +close +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/forwarding close with code: nil/ +--- no_error_log +[error] + + + +=== TEST 6: handshake with client before upstream +--- http_config eval: $t::Tests::HttpConfig +--- config + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + local data, typ, err = wb:recv_frame() + if not data then + ngx.log(ngx.ERR, "failed receiving frame: ", err) + return ngx.exit(444) + end + + ngx.log(ngx.INFO, "frame type: ", typ, ", payload: \"", data, "\"") + + local bytes, err = wb:send_text(data) + if not bytes then + ngx.log(ngx.ERR, "failed sending frame: ", err) + return ngx.exit(444) + end + } + } + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new() + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wp:connect_client() + if not ok then + ngx.log(ngx.ERR, "failed client handshake: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect_upstream(uri) + if not ok then + ngx.log(ngx.ERR, "failed connecting to upstream: ", err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_text("hello world!")) + local data = assert(wb:recv_frame()) + ngx.say(data) + } + } +--- response_body +hello world! +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/frame type: text, payload: "hello world!"/ +--- no_error_log +[error] diff --git a/t/02-timeouts.t b/t/02-timeouts.t new file mode 100644 index 0000000..60dd1ca --- /dev/null +++ b/t/02-timeouts.t @@ -0,0 +1,147 @@ +# vim:set ts=4 sts=4 sw=4 et ft=: + +use lib '.'; +use t::Tests; + +plan tests => repeat_each() * (blocks() * 4); + +run_tests(); + +__DATA__ + +=== TEST 1: on client recv_frame timeout +--- http_config eval: $t::Tests::HttpConfig +--- config + lua_socket_log_errors off; + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wb, err = proxy.new({ + recv_timeout = 80, + }) + if not wb then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wb:connect(proxy._tests.echo) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wb:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + ngx.sleep(0.1) + assert(wb:send_text("hello world!")) + local data = assert(wb:recv_frame()) + ngx.say(data) + } + } +--- response_body +hello world! +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/.*?timeout receiving frame from client, reopening.* +.*?timeout receiving frame from upstream, reopening.* +.*?frame type: text, payload: "hello world!".*/ +--- no_error_log +[crit] + + + +=== TEST 2: on upstream recv_frame timeout +--- http_config eval: $t::Tests::HttpConfig +--- config + lua_socket_log_errors off; + + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + local data, typ, err = wb:recv_frame() + if not data then + ngx.log(ngx.ERR, "failed receiving frame: ", err) + return ngx.exit(444) + end + + ngx.log(ngx.INFO, "frame type: ", typ, ", payload: \"", data, "\"") + + ngx.sleep(0.1) + + local bytes, err = wb:send_text(data) + if not bytes then + ngx.log(ngx.ERR, "failed sending frame: ", err) + return ngx.exit(444) + end + } + } + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wb, err = proxy.new({ + recv_timeout = 80, + }) + if not wb then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wb:connect(uri) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wb:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_text("hello world!")) + local data = assert(wb:recv_frame()) + ngx.say(data) + } + } +--- response_body +hello world! +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/.*?frame type: text, payload: "hello world!".* +.*?timeout receiving frame from upstream, reopening.*/ +--- no_error_log +[crit] diff --git a/t/03-on_frame_callback.t b/t/03-on_frame_callback.t new file mode 100644 index 0000000..4d92df0 --- /dev/null +++ b/t/03-on_frame_callback.t @@ -0,0 +1,453 @@ +# vim:set ts=4 sts=4 sw=4 et ft=: + +use lib '.'; +use t::Tests; + +plan tests => repeat_each() * (blocks() * 4); + +run_tests(); + +__DATA__ + +=== TEST 1: invokes opts.on_frame function on each client/upstream frame +--- http_config eval: $t::Tests::HttpConfig +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local function on_frame(_, role, typ, data, fin, code) + ngx.log(ngx.INFO, "from: ", role, ", type: ", typ, + ", payload: ", data, + ", fin: ", fin, ", code: ", code, + ", context: ", ngx.get_phase()) + -- test: only return data (code == nil) + return data + end + + local wp, err = proxy.new({ on_frame = on_frame }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wp:connect(proxy._tests.echo) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_text("hello world!")) + local data = assert(wb:recv_frame()) + ngx.say(data) + } + } +--- response_body +hello world! +--- grep_error_log eval: qr/\[lua\].*?from:.*/ +--- grep_error_log_out eval +qr/.*?from: client, type: text, payload: hello world!, fin: true, code: nil, context: content.* +.*?from: upstream, type: text, payload: hello world!, fin: true, code: nil, context: content.*/ +--- no_error_log +[error] + + + +=== TEST 2: opts.on_frame can update an upstream text frame payload +--- http_config eval: $t::Tests::HttpConfig +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local function on_frame(_, role, typ, data, fin, code) + ngx.log(ngx.INFO, "from: ", role, ", type: ", typ, + ", payload: ", data, ", fin: ", fin) + + return "updated " .. role .. " frame", code + end + + local wp, err = proxy.new({ on_frame = on_frame }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wp:connect(proxy._tests.echo) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_text("hello world!")) + local data = assert(wb:recv_frame()) + ngx.say(data) + } + } +--- response_body +updated upstream frame +--- grep_error_log eval: qr/\[lua\].*from:.*/ +--- grep_error_log_out eval +qr/.*?from: client, type: text, payload: hello world!, fin: true.* +.*?from: upstream, type: text, payload: updated client frame, fin: true.*/ +--- no_error_log +[error] + + + +=== TEST 3: opts.on_frame can update an upstream binary frame payload +--- http_config eval: $t::Tests::HttpConfig +--- config + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + local data, typ, err = wb:recv_frame() + if not data then + ngx.log(ngx.ERR, "failed receiving frame: ", err) + return ngx.exit(444) + end + + local bytes, err = wb:send_binary(data) + if not bytes then + ngx.log(ngx.ERR, "failed sending frame: ", err) + return ngx.exit(444) + end + } + } + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local function on_frame(_, role, typ, data, fin, code) + return "updated " .. role .. " frame (" .. typ .. ")" + end + + local wp, err = proxy.new({ on_frame = on_frame }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect(uri) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_binary("你好, WebSocket!")) + local data, opcode = assert(wb:recv_frame()) + ngx.say(opcode, ": ", data) + } + } +--- response_body +binary: updated upstream frame (binary) +--- no_error_log +[error] +[crit] + + + +=== TEST 4: opts.on_frame can update an upstream close frame payload +--- log_level: debug +--- http_config eval: $t::Tests::HttpConfig +--- config + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + local bytes, err = wb:send_close(1000, "server close") + if not bytes then + ngx.log(ngx.ERR, "failed sending close frame: ", err) + return ngx.exit(444) + end + } + } + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + local fmt = string.format + + local function on_frame(_, role, typ, data, fin, code) + local msg = fmt("updated %s frame (typ: %s, code: %d)", role, typ, code) + + ngx.log(ngx.DEBUG, fmt("updated %s frame payload from %s to %s", + role, fmt("%q", data), fmt("%q", msg))) + + return msg, code + end + + local wp, err = proxy.new({ on_frame = on_frame }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect(uri) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + local data, typ, err = assert(wb:recv_frame()) + ngx.say(typ) + ngx.say(data) + ngx.say(err) + } + } +--- response_body +close +updated upstream frame (typ: close, code: 1000) +1000 +--- no_error_log +[error] +[crit] + + + +=== TEST 5: opts.on_frame can update an upstream close frame status code +--- log_level: debug +--- http_config eval: $::HttpConfig +--- config + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + local bytes, err = wb:send_close(1000, "server close") + if not bytes then + ngx.log(ngx.ERR, "failed sending close frame: ", err) + return ngx.exit(444) + end + } + } + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + local fmt = string.format + + local function on_frame(_, role, typ, data, fin, code) + local updated = 1001 + + ngx.log(ngx.DEBUG, fmt("updated %s [%s] status from %s to %s", + role, typ, code, updated)) + + return data, updated + end + + local wp, err = proxy.new({ on_frame = on_frame }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect(uri) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + local data, typ, err = assert(wb:recv_frame()) + ngx.say(typ) + ngx.say(data) + ngx.say(err) + } + } +--- response_body +close +server close +1001 +--- no_error_log +[error] +[crit] + + + +=== TEST 6: opts.on_frame can drop an upstream frame +--- log_level: debug +--- http_config eval: $::HttpConfig +--- config + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + local payloads = { "a", "b", "drop me", "c"} + + for _, data in ipairs(payloads) do + local ok, err = wb:send_text(data) + if not ok then + ngx.log(ngx.ERR, "failed sending payload: ", err) + return ngx.exit(444) + end + end + + local bytes, err = wb:send_close(1000, "server close") + if not bytes then + ngx.log(ngx.ERR, "failed sending close frame: ", err) + return ngx.exit(444) + end + } + } + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local function on_frame(_, role, typ, data, fin, code) + if typ == "text" and data == "drop me" then + ngx.log(ngx.DEBUG, "dropping 'drop me' frame") + data = nil + end + + return data, code + end + + local wp, err = proxy.new({ on_frame = on_frame }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect(uri) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local fmt = string.format + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + + repeat + local data, typ, err = assert(wb:recv_frame()) + ngx.say(fmt("typ: %s, data: %q, err/code: %s", typ, data, err)) + until typ == "close" + } + } +--- response_body +typ: text, data: "a", err/code: nil +typ: text, data: "b", err/code: nil +typ: text, data: "c", err/code: nil +typ: close, data: "server close", err/code: 1000 +--- error_log +dropping 'drop me' frame +--- no_error_log +[error] diff --git a/t/04-fragmented_frames.t b/t/04-fragmented_frames.t new file mode 100644 index 0000000..92026f6 --- /dev/null +++ b/t/04-fragmented_frames.t @@ -0,0 +1,762 @@ +# vim:set ts=4 sts=4 sw=4 et ft=: + +use lib '.'; +use t::Tests; + +plan tests => repeat_each() * (blocks() * 4); + +run_tests(); + +__DATA__ + +=== TEST 1: forwards fragmented frames by default +--- http_config eval: $::HttpConfig +--- config + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + for i = 1, 2 do + local data, typ, err = wb:recv_frame() + if not data then + ngx.log(ngx.ERR, "failed receiving frame: ", err) + return ngx.exit(444) + end + + ngx.log(ngx.INFO, "frame type: ", typ, ", payload: \"", data, "\"") + + local bytes, err = wb:send_text(data) + if not bytes then + ngx.log(ngx.ERR, "failed sending frame: ", err) + return ngx.exit(444) + end + end + } + } + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new() + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect(uri) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + + for i = 1, 2 do + if i == 1 then + assert(wb:send_frame(false, 0x1, "hello")) + else + assert(wb:send_frame(true, 0x0, "world")) + end + + local data = assert(wb:recv_frame()) + ngx.say(data) + end + + wb:close() + } + } +--- response_body +hello +world +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/.*?frame type: text, payload: "hello".* +.*?frame type: continuation, payload: "world".*/ +--- no_error_log +[error] + + + +=== TEST 2: opts.aggregate_fragments assembles fragmented client frames +--- http_config eval: $::HttpConfig +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new({ aggregate_fragments = true }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wp:connect(proxy._tests.echo) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_frame(false, 0x1, "hello")) + assert(wb:send_frame(true, 0x0, " world")) + local data = assert(wb:recv_frame()) + ngx.say(data) + wb:close() + } + } +--- response_body +hello world +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/.*?frame type: text, payload: "hello world".*/ +--- no_error_log +[error] + + + +=== TEST 3: opts.aggregate_fragments assembles fragmented server frames +--- http_config eval: $::HttpConfig +--- config + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + local data, typ, err = wb:recv_frame() + if not data then + ngx.log(ngx.ERR, "failed receiving frame: ", err) + return ngx.exit(444) + end + + ngx.log(ngx.INFO, "frame type: ", typ, ", payload: \"", data, "\"") + + local bytes, err = wb:send_frame(false, 0x1, "") + if not bytes then + ngx.log(ngx.ERR, "failed sending initial fragment: ", err) + return ngx.exit(444) + end + + for word in string.gmatch(data, "[^%s]+") do + local bytes, err = wb:send_frame(false, 0x0, word) + if not bytes then + ngx.log(ngx.ERR, "failed sending fragment: ", err) + return ngx.exit(444) + end + end + + local bytes, err = wb:send_frame(true, 0x0, "") + if not bytes then + ngx.log(ngx.ERR, "failed sending last fragment: ", err) + return ngx.exit(444) + end + } + } + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new({ aggregate_fragments = true }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect(uri) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_text("hello world")) + local data = assert(wb:recv_frame()) + ngx.say(data) + wb:close() + } + } +--- response_body +helloworld +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/.*?frame type: text, payload: "hello world".*/ +--- no_error_log +[error] + + + +=== TEST 4: opts.aggregate_fragments assembles fragmented frames consecutively +--- http_config eval: $::HttpConfig +--- config + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + for i = 1, 2 do + local data, typ, err = wb:recv_frame() + if not data then + ngx.log(ngx.ERR, "failed receiving frame: ", err) + return ngx.exit(444) + end + + ngx.log(ngx.INFO, "frame type: ", typ, ", payload: \"", data, "\"") + + local bytes, err = wb:send_frame(false, 0x1, "") + if not bytes then + ngx.log(ngx.ERR, "failed sending initial fragment: ", err) + return ngx.exit(444) + end + + for word in string.gmatch(data, "[^%s]+") do + local bytes, err = wb:send_frame(false, 0x0, word) + if not bytes then + ngx.log(ngx.ERR, "failed sending fragment: ", err) + return ngx.exit(444) + end + end + + local bytes, err = wb:send_frame(true, 0x0, "") + if not bytes then + ngx.log(ngx.ERR, "failed sending last fragment: ", err) + return ngx.exit(444) + end + end + } + } + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new({ aggregate_fragments = true }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect(uri) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_frame(false, 0x1, "hello")) + assert(wb:send_frame(true, 0x0, " world")) + local data = assert(wb:recv_frame()) + ngx.say(data) + + assert(wb:send_frame(false, 0x1, "goodbye")) + assert(wb:send_frame(true, 0x0, " world")) + local data = assert(wb:recv_frame()) + ngx.say(data) + + wb:close() + } + } +--- response_body +helloworld +goodbyeworld +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/.*?frame type: text, payload: "hello world".* +.*?frame type: text, payload: "goodbye world".*/ +--- no_error_log +[error] + + + +=== TEST 5: opts.on_frame with opts.aggregate_fragments +--- http_config eval: $::HttpConfig +--- config + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + for i = 1, 2 do + local data, typ, err = wb:recv_frame() + if not data then + ngx.log(ngx.ERR, "failed receiving frame: ", err) + return ngx.exit(444) + end + + local bytes, err = wb:send_frame(false, 0x1, "") + if not bytes then + ngx.log(ngx.ERR, "failed sending initial fragment: ", err) + return ngx.exit(444) + end + + for word in string.gmatch(data, "[^%s]+") do + local bytes, err = wb:send_frame(false, 0x0, word) + if not bytes then + ngx.log(ngx.ERR, "failed sending fragment: ", err) + return ngx.exit(444) + end + end + + local bytes, err = wb:send_frame(true, 0x0, "") + if not bytes then + ngx.log(ngx.ERR, "failed sending last fragment: ", err) + return ngx.exit(444) + end + end + } + } + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local function on_frame(_, role, typ, data, fin, code) + ngx.log(ngx.INFO, "from: ", role, ", type: ", typ, + ", payload: ", data, ", fin: ", fin) + + return "updated " .. role .. " frame", code + end + + local wp, err = proxy.new({ + aggregate_fragments = true, + on_frame = on_frame, + }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect(uri) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_frame(false, 0x1, "hello")) + assert(wb:send_frame(true, 0x0, " world")) + local data = assert(wb:recv_frame()) + ngx.say(data) + + assert(wb:send_frame(false, 0x1, "goodbye")) + assert(wb:send_frame(true, 0x0, " world")) + local data = assert(wb:recv_frame()) + ngx.say(data) + + wb:close() + } + } +--- response_body +updated upstream frame +updated upstream frame +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/.*?from: client, type: text, payload: hello world, fin: true.* +.*?from: upstream, type: text, payload: updatedclientframe, fin: true.* +.*?from: client, type: text, payload: goodbye world, fin: true.* +.*?from: upstream, type: text, payload: updatedclientframe, fin: true.*/ +--- no_error_log +[error] + + + +=== TEST 6: opts.on_frame without opts.aggregate_fragments +--- http_config eval: $::HttpConfig +--- config + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + for i = 1, 2 do + local data, typ, err = wb:recv_frame() + if not data then + ngx.log(ngx.ERR, "failed receiving frame: ", err) + return ngx.exit(444) + end + + local bytes, err = wb:send_frame(false, 0x1, "") + if not bytes then + ngx.log(ngx.ERR, "failed sending initial fragment: ", err) + return ngx.exit(444) + end + + for word in string.gmatch(data, "[^%s]+") do + local bytes, err = wb:send_frame(false, 0x0, word) + if not bytes then + ngx.log(ngx.ERR, "failed sending fragment: ", err) + return ngx.exit(444) + end + end + + local bytes, err = wb:send_frame(true, 0x0, "") + if not bytes then + ngx.log(ngx.ERR, "failed sending last fragment: ", err) + return ngx.exit(444) + end + end + } + } + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local function on_frame(_, role, typ, data, fin, code) + ngx.log(ngx.INFO, "from: ", role, ", type: ", typ, + ", payload: ", data, ", fin: ", fin) + + return "updated " .. role .. " frame", code + end + + local wp, err = proxy.new({ + aggregate_fragments = false, + on_frame = on_frame, + }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect(uri) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_frame(false, 0x1, "hello")) + assert(wb:send_frame(true, 0x0, " world")) + local data = assert(wb:recv_frame()) + ngx.say(data) + + assert(wb:send_frame(false, 0x1, "goodbye")) + assert(wb:send_frame(true, 0x0, " world")) + local data = assert(wb:recv_frame()) + ngx.say(data) + + wb:close() + } + } +--- response_body +updated upstream frame +updated upstream frame +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/.*?from: client, type: text, payload: hello, fin: false.* +.*?from: client, type: continuation, payload: world, fin: true.* +.*?from: upstream, type: text, payload: , fin: false.* +.*?from: upstream, type: continuation, payload: updated, fin: false.* +.*?from: upstream, type: continuation, payload: client, fin: false.* +.*?from: upstream, type: continuation, payload: frame, fin: false.* +.*?from: upstream, type: continuation, payload: , fin: true.* +.*?from: upstream, type: text, payload: , fin: false.* +.*?from: upstream, type: continuation, payload: updated, fin: false.* +.*?from: upstream, type: continuation, payload: client, fin: false.* +.*?from: upstream, type: continuation, payload: frame, fin: false.* +.*?from: upstream, type: continuation, payload: , fin: true.*/ +--- no_error_log +[error] + + + +=== TEST 7: control frames interleaved with fragmented data frames (opts.aggregate_fragments off) +--- http_config eval: $::HttpConfig +--- config + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local function check(ok, err, msg) + if not ok then + ngx.log(ngx.ERR, msg, ": ", err) + return ngx.exit(444) + end + end + + local wb, err = server:new() + check(wb, err, "failed creating server") + + local ok, err = wb:send_frame(false, 0x1, "") + check(ok, err, "failed sending initial fragment") + + local payloads = { "a", "b", "c" } + + for i, data in ipairs(payloads) do + ok, err = wb:send_frame(false, 0x0, data) + check(ok, err, "failed sending partial data frame") + + ok, err = wb:send_ping(i) + check(ok, err, "failed sending ping") + end + + ok, err = wb:send_frame(true, 0x0, "") + check(ok, err, "failed sending last fragment") + + local bytes, err = wb:send_close(1000, "server close") + check(bytes, err, "failed sending close frame") + } + } + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local function check(ok, err, msg) + if not ok then + ngx.log(ngx.ERR, msg, ": ", err) + return ngx.exit(444) + end + end + + local wp, err = proxy.new({ aggregate_fragments = false }) + check(wp, err, "failed creating proxy") + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect_upstream(uri) + check(ok, err, "failed connecting to upstream") + + ok, err = wp:connect_client() + check(ok, err, "failed client handshake") + + local done, err = wp:execute() + check(done, err, "failed proxying") + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local fmt = string.format + + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + + repeat + local data, typ, err = wb:recv_frame() + ngx.say(fmt("typ: %s, data: %q, err/code: %s", typ, data, err)) + until typ == "close" + } + } +--- response_body +typ: text, data: "", err/code: again +typ: continuation, data: "a", err/code: again +typ: ping, data: "1", err/code: nil +typ: continuation, data: "b", err/code: again +typ: ping, data: "2", err/code: nil +typ: continuation, data: "c", err/code: again +typ: ping, data: "3", err/code: nil +typ: continuation, data: "", err/code: nil +typ: close, data: "server close", err/code: 1000 +--- no_error_log +[error] +[crit] + + + +=== TEST 8: control frames interleaved with fragmented data frames (opts.aggregate_fragments on) +--- http_config eval: $::HttpConfig +--- config + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local function check(ok, err, msg) + if not ok then + ngx.log(ngx.ERR, msg, ": ", err) + return ngx.exit(444) + end + end + + local wb, err = server:new() + check(wb, err, "failed creating server") + + local ok, err = wb:send_frame(false, 0x1, "") + check(ok, err, "failed sending initial fragment") + + local payloads = { "a", "b", "c" } + + for i, data in ipairs(payloads) do + ok, err = wb:send_frame(false, 0x0, data) + check(ok, err, "failed sending partial data frame") + + ok, err = wb:send_ping(i) + check(ok, err, "failed sending ping") + end + + ok, err = wb:send_frame(true, 0x0, "") + check(ok, err, "failed sending final fragment") + + local bytes, err = wb:send_close(1000, "server close") + check(bytes, err, "failed sending close frame") + } + } + + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local function check(ok, err, msg) + if not ok then + ngx.log(ngx.ERR, msg, ": ", err) + return ngx.exit(444) + end + end + + local wp, err = proxy.new({ aggregate_fragments = true }) + check(wp, err, "failed creating proxy") + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect_upstream(uri) + check(ok, err, "failed connecting to upstream") + + ok, err = wp:connect_client() + check(ok, err, "failed client handshake") + + local done, err = wp:execute() + check(done, err, "failed proxying") + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local fmt = string.format + + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + + repeat + local data, typ, err = wb:recv_frame() + ngx.say(fmt("typ: %s, data: %q, err/code: %s", typ, data, err)) + until typ == "close" + } + } +--- response_body +typ: ping, data: "1", err/code: nil +typ: ping, data: "2", err/code: nil +typ: ping, data: "3", err/code: nil +typ: text, data: "abc", err/code: nil +typ: close, data: "server close", err/code: 1000 +--- no_error_log +[error] +[crit] diff --git a/t/05-wss.t b/t/05-wss.t new file mode 100644 index 0000000..126ede6 --- /dev/null +++ b/t/05-wss.t @@ -0,0 +1,92 @@ +# vim:set ts=4 sts=4 sw=4 et ft=: + +use lib '.'; +use t::Tests; + +plan tests => repeat_each() * (blocks() * 4); + +run_tests(); + +__DATA__ + +=== TEST 1: forward a text frame back and forth over wss +--- http_config eval +qq{ + $t::Tests::HttpConfig + + server { + listen $ENV{TEST_NGINX_PORT2} ssl; + ssl_certificate $ENV{TEST_NGINX_CERT_DIR}/cert.pem; + ssl_certificate_key $ENV{TEST_NGINX_CERT_DIR}/key.pem; + + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + local data, typ, err = wb:recv_frame() + if not data then + ngx.log(ngx.ERR, "failed receiving frame: ", err) + return ngx.exit(444) + end + + ngx.log(ngx.INFO, "frame type: ", typ, ", payload: ", data) + + local bytes, err = wb:send_text(data) + if not bytes then + ngx.log(ngx.ERR, "failed sending frame: ", err) + return ngx.exit(444) + end + } + } + } +} +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new() + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wp:connect("wss://127.0.0.1:9001/upstream") + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_text("hello world!")) + local data = assert(wb:recv_frame()) + ngx.say(data) + } + } +--- response_body +hello world! +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/frame type: text, payload: hello world!/ +--- no_error_log +[error] diff --git a/t/06-error_handling.t b/t/06-error_handling.t new file mode 100644 index 0000000..47a4990 --- /dev/null +++ b/t/06-error_handling.t @@ -0,0 +1,86 @@ +# vim:set ts=4 sts=4 sw=4 et ft=: + +use lib '.'; +use t::Tests; + +plan tests => repeat_each() * (blocks() * 4); + +run_tests(); + +__DATA__ + +=== TEST 1: wss:// proxy over ws:// upstream +--- http_config eval +qq{ + $t::Tests::HttpConfig + + server { + listen $ENV{TEST_NGINX_PORT2}; + + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + local data, typ, err = wb:recv_frame() + if not data then + ngx.log(ngx.ERR, "failed receiving frame: ", err) + return ngx.exit(444) + end + + ngx.log(ngx.INFO, "frame type: ", typ, ", payload: ", data) + + local bytes, err = wb:send_text(data) + if not bytes then + ngx.log(ngx.ERR, "failed sending frame: ", err) + return ngx.exit(444) + end + } + } + } +} +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new() + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wp:connect("wss://127.0.0.1:9001/upstream") + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + wb:connect(uri) + } + } +--- ignore_response_body +--- error_log +SSL_do_handshake() failed +failed connecting to upstream: ssl handshake failed: handshake failed +--- no_error_log +runtime error diff --git a/t/07-invalid_usage.t b/t/07-invalid_usage.t new file mode 100644 index 0000000..db09d1c --- /dev/null +++ b/t/07-invalid_usage.t @@ -0,0 +1,202 @@ +# vim:set ts=4 sts=4 sw=4 et ft=: + +use lib '.'; +use t::Tests; + +plan tests => repeat_each() * (blocks() * 4); + +run_tests(); + +__DATA__ + +=== TEST 1: calling connect_upstream() while already established logs a warning +--- http_config eval: $t::Tests::HttpConfig +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new() + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wp:connect(proxy._tests.echo) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + assert(wp:connect_upstream(uri)) + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_text("hello world!")) + local data = assert(wb:recv_frame()) + ngx.say(data) + } + } +--- response_body +hello world! +--- grep_error_log eval: qr/\[(info|warn)\].*/ +--- grep_error_log_out eval +qr/\A\[warn\] .*? connection with upstream at "ws:.*?" already established.* +\[info\] .*? frame type: text, payload: "hello world!"/ +--- no_error_log +[error] + + + +=== TEST 2: calling connect_client() while client handshake already completed logs a warning +--- http_config eval: $t::Tests::HttpConfig +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new() + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wp:connect(proxy._tests.echo) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + assert(wp:connect_client(uri)) + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_text("hello world!")) + local data = assert(wb:recv_frame()) + ngx.say(data) + } + } +--- response_body +hello world! +--- grep_error_log eval: qr/\[(info|warn)\].*/ +--- grep_error_log_out eval +qr/\A\[warn\] .*? client handshake already completed.* +\[info\] .*? frame type: text, payload: "hello world!"/ +--- no_error_log +[error] + + + +=== TEST 3: calling execute() without having completed the client handshake +--- http_config eval: $t::Tests::HttpConfig +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new() + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wp:connect_upstream(proxy._tests.echo) + if not ok then + ngx.log(ngx.ERR, "failed connecting to upstream: ", err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + } + } +--- error_code: 500 +--- ignore_response_body +--- grep_error_log eval: qr/\[error\].*/ +--- grep_error_log_out eval +qr/\A\[error\] .*? failed proxying: client handshake not complete.*/ +--- no_error_log +[crit] +[emerg] + + + +=== TEST 4: calling execute() without having established the upstream connection +--- http_config eval: $t::Tests::HttpConfig +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new({debug = true}) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wp:connect_client() + if not ok then + ngx.log(ngx.ERR, "failed client handshake: ", err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + } + } +--- response_body +--- grep_error_log eval: qr/\[error\].*/ +--- grep_error_log_out eval +qr/\A\[error\] .*? failed proxying: upstream connection not established.*/ +--- no_error_log +[crit] diff --git a/t/08-limits.t b/t/08-limits.t new file mode 100644 index 0000000..4497087 --- /dev/null +++ b/t/08-limits.t @@ -0,0 +1,565 @@ +# vim:set ts=4 sts=4 sw=4 et ft=: + +use lib '.'; +use t::Tests; + +plan tests => repeat_each() * (blocks() * 4); + +run_tests(); + +__DATA__ + +=== TEST 1: limiting individual frame size (client) +--- http_config eval: $t::Tests::HttpConfig +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new({ client_max_frame_size = 10 }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wp:connect(proxy._tests.echo .. "?repeat=1") + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_text("this is way too long")) + local data, typ, err = wb:recv_frame() + ngx.say(string.format("data: %q, typ: %s, err: %s", data, typ, err)) + } + } +--- response_body +data: "Payload Too Large", typ: close, err: 1009 +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/frame type: close, payload: ""/ +--- no_error_log +[error] + + + +=== TEST 2: limiting individual frame size (upstream) +--- http_config eval: $t::Tests::HttpConfig +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new({ upstream_max_frame_size = 10 }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect(uri) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + local bytes, err = wb:send_text("this is way too long") + if not bytes then + ngx.log(ngx.ERR, "failed sending frame: ", err) + return ngx.exit(444) + end + + local data, typ, err = wb:recv_frame() + if not data then + ngx.log(ngx.ERR, "failed receiving frame: ", err) + return ngx.exit(444) + end + + ngx.log(ngx.INFO, "frame type: ", typ, ", payload: \"", data, "\"") + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + assert(wb:send_text("hello")) + local data, typ, err = wb:recv_frame() + ngx.say(string.format("data: %q, typ: %s, err: %s", data, typ, err)) + } + } +--- response_body +data: "", typ: close, err: 1001 +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/frame type: close, payload: "Payload Too Large"/ +--- no_error_log +[error] + + + +=== TEST 3: limiting aggregated frame size (client) +--- http_config eval: $t::Tests::HttpConfig +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new({ + client_max_frame_size = 10, + aggregate_fragments = true, + }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wp:connect(proxy._tests.pong .. "?repeat=1") + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + + local fmt = string.format + + local function ping_pong(i) + local bytes, err = wb:send_ping(i) + if not bytes then + return nil, fmt("send failed: %s", err) + end + + local data, typ, err = wb:recv_frame() + if not data then + return nil, fmt("recv failed: %s", err) + elseif typ ~= "pong" then + return nil, fmt("unexpected frame %s => %q", typ, data) + end + + return true + end + + for i = 1, 5 do + local opcode = (i == 1 and 0x1) or 0x0 + local bytes, err = wb:send_frame(false, opcode, "11") + if not bytes then + ngx.log(ngx.ERR, "failed sending fragment ", i, ": ", err) + return ngx.exit(500) + end + + local ok, err = ping_pong(i) + if not ok then + ngx.log(ngx.ERR, "failed ping-pong: ", err) + return ngx.exit(500) + end + end + + local bytes, err = wb:send_frame(false, 0x0, "1") + if not bytes then + ngx.log(ngx.ERR, "failed sending final fragment: ", err) + return ngx.exit(500) + end + + local data, typ, err = wb:recv_frame() + ngx.say(string.format("data: %q, typ: %s, err: %s", data, typ, err)) + } + } +--- response_body +data: "Payload Too Large", typ: close, err: 1009 +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/frame type: close, payload: ""/ +--- no_error_log +[error] + + + +=== TEST 4: limiting aggregated frame size (upstream) +--- http_config eval: $t::Tests::HttpConfig +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new({ + aggregate_fragments = true, + upstream_max_frame_size = 10 + }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect(uri) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + for i = 1, 5 do + local opcode = (i == 1 and 0x1) or 0x0 + local bytes, err = wb:send_frame(false, opcode, "11") + if not bytes then + ngx.log(ngx.ERR, "failed sending fragment ", i, ": ", err) + return ngx.exit(444) + end + end + + local bytes, err = wb:send_frame(false, 0x0, "1") + if not bytes then + ngx.log(ngx.ERR, "failed sending final fragment: ", err) + return ngx.exit(444) + end + + local data, typ, err = wb:recv_frame() + ngx.log(ngx.INFO, "frame type: ", typ, ", payload: \"", data, "\", code: ", err) + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + repeat + local data, typ, err = wb:recv_frame() + ngx.say(string.format("data: %q, typ: %s, err: %s", + data, typ, err)) + until typ == "close" or not data + } + } +--- response_body +data: "", typ: close, err: 1001 +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/frame type: close, payload: "Payload Too Large", code: 1009/ +--- no_error_log +[error] + + + +=== TEST 5: control frames are not subject to max_frame_size +--- http_config eval: $t::Tests::HttpConfig +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new({ + client_max_frame_size = 2, + upstream_max_frame_size = 2, + }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wp:connect(proxy._tests.pong .. "?repeat=1") + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + + for i = 1, 2 do + local sent, err = wb:send_ping("test-" .. i) + if not sent then + ngx.log(ngx.ERR, "failed sending ping: ", err) + return ngx.exit(500) + end + + local data, typ, err = wb:recv_frame() + if not data then + ngx.log(ngx.ERR, "failed to receive pong: ", err) + return ngx.exit(500) + + elseif typ ~= "pong" then + ngx.log(ngx.ERR, "unexpecting response to ping: ", typ) + return ngx.exit(500) + + elseif #data <= 2 then + ngx.log(ngx.ERR, "broken test--pong frame is too short") + return ngx.exit(500) + end + + ngx.say(string.format("data: %q, typ: %s", data, typ)) + end + + assert(wb:send_close(1002, "goodbye")) + } + } +--- response_body +data: "heartbeat server", typ: pong +data: "heartbeat server", typ: pong +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/.*?frame type: ping, payload: "test-1".* +.*?frame type: ping, payload: "test-2".* +.*?forwarding close with code: 1002.* +.*?frame type: close, payload: "goodbye".*/ +--- no_error_log +[error] + + + +=== TEST 6: limiting the number of fragments (client) +--- http_config eval: $t::Tests::HttpConfig +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new({ + client_max_fragments = 5, + aggregate_fragments = true, + debug = true, + }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local ok, err = wp:connect(proxy._tests.pong .. "?repeat=1") + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + + local fmt = string.format + + local function ping_pong(i) + local bytes, err = wb:send_ping(i) + if not bytes then + return nil, fmt("send failed: %s", err) + end + + local data, typ, err = wb:recv_frame() + if not data then + return nil, fmt("recv failed: %s", err) + elseif typ ~= "pong" then + return nil, fmt("unexpected frame %s => %q", typ, data) + end + + return true + end + + for i = 1, 5 do + local opcode = (i == 1 and 0x1) or 0x0 + local bytes, err = wb:send_frame(false, opcode, "11") + if not bytes then + ngx.log(ngx.ERR, "failed sending fragment ", i, ": ", err) + return ngx.exit(500) + end + + local ok, err = ping_pong(i) + if not ok then + ngx.log(ngx.ERR, "failed ping-pong: ", err) + return ngx.exit(500) + end + end + + local bytes, err = wb:send_frame(false, 0x0, "1") + if not bytes then + ngx.log(ngx.ERR, "failed sending final fragment: ", err) + return ngx.exit(500) + end + + local data, typ, err = wb:recv_frame() + ngx.say(string.format("data: %q, typ: %s, err: %s", data, typ, err)) + } + } +--- response_body +data: "Payload Too Large", typ: close, err: 1009 +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/frame type: close, payload: ""/ +--- no_error_log +[error] + + + +=== TEST 7: limiting the number of fragments (upstream) +--- http_config eval: $t::Tests::HttpConfig +--- config + location /proxy { + content_by_lua_block { + local proxy = require "resty.websocket.proxy" + + local wp, err = proxy.new({ + aggregate_fragments = true, + upstream_max_fragments = 5, + }) + if not wp then + ngx.log(ngx.ERR, "failed creating proxy: ", err) + return ngx.exit(444) + end + + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/upstream" + local ok, err = wp:connect(uri) + if not ok then + ngx.log(ngx.ERR, err) + return ngx.exit(444) + end + + local done, err = wp:execute() + if not done then + ngx.log(ngx.ERR, "failed proxying: ", err) + return ngx.exit(444) + end + } + } + + location /upstream { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + for i = 1, 5 do + local opcode = (i == 1 and 0x1) or 0x0 + local bytes, err = wb:send_frame(false, opcode, "11") + if not bytes then + ngx.log(ngx.ERR, "failed sending fragment ", i, ": ", err) + return ngx.exit(444) + end + end + + local bytes, err = wb:send_frame(false, 0x0, "1") + if not bytes then + ngx.log(ngx.ERR, "failed sending final fragment: ", err) + return ngx.exit(444) + end + + local data, typ, err = wb:recv_frame() + ngx.log(ngx.INFO, "frame type: ", typ, ", payload: \"", data, "\", code: ", err) + } + } + + location /t { + content_by_lua_block { + local client = require "resty.websocket.client" + local wb = assert(client:new()) + local uri = "ws://127.0.0.1:" .. ngx.var.server_port .. "/proxy" + + assert(wb:connect(uri)) + repeat + local data, typ, err = wb:recv_frame() + ngx.say(string.format("data: %q, typ: %s, err: %s", + data, typ, err)) + until typ == "close" or not data + } + } +--- response_body +data: "", typ: close, err: 1001 +--- grep_error_log eval: qr/\[lua\].*/ +--- grep_error_log_out eval +qr/frame type: close, payload: "Payload Too Large", code: 1009/ +--- no_error_log +[error] \ No newline at end of file diff --git a/t/Tests.pm b/t/Tests.pm new file mode 100644 index 0000000..9e36557 --- /dev/null +++ b/t/Tests.pm @@ -0,0 +1,117 @@ +package t::Tests; + +use strict; +use Test::Nginx::Socket::Lua -Base; +use Cwd qw(cwd); + +our $pwd = cwd(); + +# TODO: switch to unix sockets once supported by lua-resty-websocket +# -> will conflict when using TEST_NGINX_RANDOMIZE +$ENV{TEST_NGINX_PORT_UPSTREAM} ||= 1985; +$ENV{TEST_NGINX_PORT2} ||= 9001; +$ENV{TEST_NGINX_CERT_DIR} ||= File::Spec->catdir(server_root(), '..', 'certs'); + +our $HttpConfig = qq{ + lua_package_path "$pwd/lib/?.lua;;"; + + init_worker_by_lua_block { + local proxy = require "resty.websocket.proxy" + + proxy._tests = { + echo = "ws://127.0.0.1:$ENV{TEST_NGINX_PORT_UPSTREAM}/echo", + pong = "ws://127.0.0.1:$ENV{TEST_NGINX_PORT_UPSTREAM}/pong", + } + } + + server { + listen $ENV{TEST_NGINX_PORT_UPSTREAM}; + + location /echo { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + local once = not ngx.var.arg_repeat + + repeat + local data, typ, err = wb:recv_frame() + if not data then + ngx.log(ngx.ERR, "failed receiving frame: ", err) + return ngx.exit(444) + end + + ngx.log(ngx.INFO, "frame type: ", typ, + ", payload: \\"", data, + "\\"") + + local bytes, err + if typ == "close" then + bytes, err = wb:send_close(err, data) + else + bytes, err = wb:send_text(data) + end + + if not bytes then + ngx.log(ngx.ERR, "failed sending frame: ", err) + return ngx.exit(444) + end + until typ == "close" or once + } + } + + location /pong { + content_by_lua_block { + local server = require "resty.websocket.server" + + local wb, err = server:new() + if not wb then + ngx.log(ngx.ERR, "failed creating server: ", err) + return ngx.exit(444) + end + + local once = not ngx.var.arg_repeat + + repeat + local data, typ, err = wb:recv_frame() + if not data then + ngx.log(ngx.ERR, "failed receiving frame: ", err) + return ngx.exit(444) + end + + ngx.log(ngx.INFO, "frame type: ", typ, + ", payload: \\"", data, "\\"") + + local bytes, err = wb:send_pong("heartbeat server") + if not bytes then + ngx.log(ngx.ERR, "failed sending frame: ", err) + return ngx.exit(444) + end + until typ == "close" or once + } + } + } +}; + +our @EXPORT = qw( + $pwd + $HttpConfig +); + +add_block_preprocessor(sub { + my $block = shift; + + if (!defined $block->request) { + $block->set_value("request", "GET /t"); + } +}); + +log_level('info'); +no_long_string(); + +1; diff --git a/t/certs/cert.pem b/t/certs/cert.pem new file mode 100644 index 0000000..9a82834 --- /dev/null +++ b/t/certs/cert.pem @@ -0,0 +1,19 @@ +-----BEGIN CERTIFICATE----- +MIIDCTCCAfGgAwIBAgIUG/V46duZqnXkisF9s+Vs0iMhHzAwDQYJKoZIhvcNAQEL +BQAwFDESMBAGA1UEAwwJbG9jYWxob3N0MB4XDTIxMTAyOTAxMjcwMVoXDTMxMTAy +NzAxMjcwMVowFDESMBAGA1UEAwwJbG9jYWxob3N0MIIBIjANBgkqhkiG9w0BAQEF +AAOCAQ8AMIIBCgKCAQEAtM8uxNfetZHmupQ2dlIb81gFm78jeRMJmjDaX0B7HADo +X7XER7SQMgH5We2XGXxcFFg/q2dQyMO85DiWizij4Zh37NeFQXAHhhVXKNvS2RPT ++x2SY5iFKzwdxPT7d6fBJwOpAuHRVYAl3c/tWEqHGwoQBlVF7SksSJTIoP8+Ar0s +lFr6/SnoHMUjXxqRTVG+3RGASV2wo25YdZbwyGhmoe9rYSWIdt5gKh8F84JyRGdu +BSaSOfpV9P56Et08d26Bgy6xK3G5Q50+aqqHNbt36DS866aKlnxgRqbq10NPjcTb +asHnE5YebodJU6clnreSb1xMuqHcSzyJJAkQYIOttwIDAQABo1MwUTAdBgNVHQ4E +FgQUI+SBnc0cXEAG0h/iQWBWNhf52K0wHwYDVR0jBBgwFoAUI+SBnc0cXEAG0h/i +QWBWNhf52K0wDwYDVR0TAQH/BAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEAL/h8 +eD4iLRY+GLOBAnB14UYTCD7kEzgsXALPI0TVh66Gt89wMtzMC/4//xhpRtysMctn +510pbJrvpLmX1hp+cXL3wKpKgHT2NO7F/5bHt6mAttm9LWgKRsH2pqngIjDEROFt +9IWNsAVEAOAm+p/45rkR4Ca3DGenMxSTw9nDr8rODhnN62smC9T1QU/7QPPJTRYA +XyFnPxj5uOJ5/9CHVOtftG862Q85P/p8Jm9OEBWQLJ911rGnCfKPoN1aCm5OnZjG +rhk00FOe6GnWWwO0EBE+B9TkQ614o9q0a5z9xwnt0zoPrI87O6jVVcbB7X2yuIY5 +gEPKlPyPwYaW9eGi9g== +-----END CERTIFICATE----- diff --git a/t/certs/key.pem b/t/certs/key.pem new file mode 100644 index 0000000..4a2ab1c --- /dev/null +++ b/t/certs/key.pem @@ -0,0 +1,28 @@ +-----BEGIN PRIVATE KEY----- +MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC0zy7E1961kea6 +lDZ2UhvzWAWbvyN5EwmaMNpfQHscAOhftcRHtJAyAflZ7ZcZfFwUWD+rZ1DIw7zk +OJaLOKPhmHfs14VBcAeGFVco29LZE9P7HZJjmIUrPB3E9Pt3p8EnA6kC4dFVgCXd +z+1YSocbChAGVUXtKSxIlMig/z4CvSyUWvr9KegcxSNfGpFNUb7dEYBJXbCjblh1 +lvDIaGah72thJYh23mAqHwXzgnJEZ24FJpI5+lX0/noS3Tx3boGDLrErcblDnT5q +qoc1u3foNLzrpoqWfGBGpurXQ0+NxNtqwecTlh5uh0lTpyWet5JvXEy6odxLPIkk +CRBgg623AgMBAAECggEABxtdqz8Q8fIaL5tDyMLRdm8vST/qFQg6iGUDMFtIe4FN +oPV0EZ13TX/mXTKJjebld6dAWWxmMH4BHrdas342ctebXyPZiQjuQsbz9hct4np1 +CpnO9zs3gMuMIAHBIKLnZLEwI3zczY+L7XwOyDmltdpfZuBTL08+/ltax3lA5rqX +0vti0HdgTtNWY8ur4sDpRWNP86gvoLn8wljI/bnnQ8Qjh+XNMaJ4dMl0jm8pR6aE +JaGn95xrJfuoPfxAjYoskxC+JTDZ56gUS+QFrG2Ys2mvI8s/PDGpeT7JCNTHo5kZ +M7xHB1TKyRWNQZOabRqQb9dgwGQTufxBPCzWsxYYAQKBgQDvp1wRe3x0ka/7Zauu +BZjEUdPud8GcJ5g7E/6aCong10z3B2DqJnTZUKFfDSxqxRyH6zVQfL/2mc6Ae1hA +Qanvz8hRzG3N9/fqMQZwmvl/+dyUcPUIA9qPSocQPOq23D06ml0YM2WY3a4/DrGN +KdMeYcbwBqJbPlESQ+155pRF9wKBgQDBJFQzD8U5r0qzMOFnaW6kDfcTguNObikM +QT4rKbSvXWCJalTO3fYKGn5QQJmipXlto7N7pogC5Niyq75faSyCExH6m4mS/yAq +fS/bujvkv/7caCuk0I4sH+AGXiQT4r+ZLamzfG5S5RZsJRKdKBLPPhlHQp+w8eP9 +MWRx6Y7mQQKBgB1XiFY7oElFbR6CnDp6RPIEcsZHs1TDJEhXcly53ZfBxJKaPKtu +efABh0B0BHQuHggmLCjmnTo8KqRgdbWoVH4gGo6pUOhe6+OojPlMgC/DD9a83cNV +dXUF0vOSAcrRvE1oiO1lnZLR0Xu2+NYnwMl/fzP0G/y/7H0oA/Ng39aRAoGACBY1 +UE4AvAGS5x3M0j+f2k6bYb3BNq92GIVCzRvZ7N3EBPzmwCKbAqFJWKLby+uRwf1w +rmSynSxnxlNajTQiIPAbeJq6j+UOu7iyIEkT6OgBC8lwyl3RFhffkWSvrgV4cDlx +OYqkM+RTpAOJW/spjnPTKyCm/wmhuiAvTHWy/gECgYEA0owJNvNNNK/RCPCS8N1H +KoSeMRZIQHd7jBy9a1ZhZGSjkIQLC5GOpJacnIThiTSjy1wcKrJW2m2j0HjruVdL +gzE4GsDant2ZX4+kXCrL0qvCI7IZ+STNCC/Fdya3rnbUOuVBoiHu5XHjltH2drCX +SJqZyEN5i/QzQ6lWhsnhYOo= +-----END PRIVATE KEY----- From 6d764fe41629ef896cbfd107054e8ac5b506eadd Mon Sep 17 00:00:00 2001 From: Zeping Bu Date: Sun, 13 Sep 2026 20:27:43 +0800 Subject: [PATCH 2/3] fix: address CodeRabbit findings on the ported websocket proxy - Wrap the opts.on_frame callback invocation in pcall(); a callback that errors or returns an invalid type now closes the connection gracefully and reports an error instead of crashing the forwarder coroutine, which ngx.thread.wait's failure branch (see below) was not handling correctly. - Fix the ngx.thread.wait() failure branch in execute(): on ok == false, the error message is in the second return value (bound to 'res' here), not 'err', so the failure was logged as nil and, since 'err' stayed nil, execute() went on to report success. Also kill both threads in this branch instead of leaving the other one running. - Only reset frame_count when a data frame's fin flag is true. Resetting it after every forwarded frame (the previous behavior) meant client_max_fragments/upstream_max_fragments could never trigger when aggregate_fragments is false, since the count never accumulated past 1. - Stop logging the full close-frame payload and the full upstream URI unconditionally (INFO/WARN levels, not gated by opts.debug): log the close payload's length instead of its content, and drop the URI (which may carry sensitive query-string data) from the duplicate-connection warning. - Default to ssl_verify = true for the upstream connection unless the caller explicitly sets opts.ssl_verify, so a wss:// upstream is verified by default. t/05-wss.t now passes ssl_verify = false explicitly, since it connects to a self-signed test certificate. - t/05-wss.t and t/06-error_handling.t: the upstream server is bound to $ENV{TEST_NGINX_PORT2} via '--http_config eval', but 'wp:connect()' had the port 9001 hardcoded in a plain (non-eval) --config block. Switch --config to --config eval and interpolate $ENV{TEST_NGINX_PORT2} so the test doesn't break when that variable is overridden (e.g. under TEST_NGINX_RANDOMIZE). --- lib/resty/websocket/proxy.lua | 56 +++++++++++++++++++++++++++-------- t/05-wss.t | 10 +++++-- t/06-error_handling.t | 6 ++-- 3 files changed, 56 insertions(+), 16 deletions(-) diff --git a/lib/resty/websocket/proxy.lua b/lib/resty/websocket/proxy.lua index fc212cb..bc9b211 100644 --- a/lib/resty/websocket/proxy.lua +++ b/lib/resty/websocket/proxy.lua @@ -355,12 +355,22 @@ local function forwarder(self, ctx) -- callback if on_frame then - local updated, updated_code = on_frame(self, role, typ, - data, fin, code) + local call_ok, updated, updated_code = pcall(on_frame, + self, role, typ, data, fin, code) + if not call_ok then + log(ngx.ERR, fmt("opts.on_frame callback errored: %s", + updated)) + self[self_state] = _STATES.CLOSING + return role, updated + end + if updated ~= nil then if type(updated) ~= "string" then - error("opts.on_frame return value must be " .. - "nil or a string") + log(ngx.ERR, "opts.on_frame return value must " .. + "be nil or a string") + self[self_state] = _STATES.CLOSING + return role, "opts.on_frame return value must " .. + "be nil or a string" end end @@ -368,8 +378,11 @@ local function forwarder(self, ctx) if typ == "close" and updated_code ~= nil then if type(updated_code) ~= "number" then - error("opts.on_frame status code return value " .. - "must be nil or a number") + log(ngx.ERR, "opts.on_frame status code return " .. + "value must be nil or a number") + self[self_state] = _STATES.CLOSING + return role, "opts.on_frame status code return " .. + "value must be nil or a number" end code = updated_code @@ -383,8 +396,8 @@ local function forwarder(self, ctx) else if typ == "close" then - log(ngx.INFO, "forwarding close with code: ", code, ", payload: ", - data) + log(ngx.INFO, "forwarding close with code: ", code, + ", payload length: ", data and #data or 0) bytes, err = peer_ws:send_close(code, data) @@ -402,7 +415,9 @@ local function forwarder(self, ctx) if data_frame then frame_size = 0 - frame_count = 0 + if fin then + frame_count = 0 + end end end @@ -418,14 +433,27 @@ end function _M:connect_upstream(uri, opts) if self.upstream_state == _STATES.ESTABLISHED then - log(ngx.WARN, fmt("connection with upstream at %q already established", - self.upstream_uri)) + -- don't log the URI: it may carry sensitive data in its query string + log(ngx.WARN, "connection with upstream already established") return true end self:dd("connecting to \"", uri, "\" upstream") - local ok, err, res = self.client:connect(uri, opts) + -- default to verifying the upstream's TLS certificate unless the caller + -- explicitly asked not to; copy opts so the caller's table isn't mutated + local connect_opts = new_tab(0, 8) + if opts then + for k, v in pairs(opts) do + connect_opts[k] = v + end + end + + if connect_opts.ssl_verify == nil then + connect_opts.ssl_verify = true + end + + local ok, err, res = self.client:connect(uri, connect_opts) if not ok then return nil, err end @@ -501,7 +529,11 @@ function _M:execute() local ok, res, err = ngx.thread.wait(self.co_client, self.co_server) if not ok then + -- res carries the error message here, not err + err = res log(ngx.ERR, "failed to wait for websocket proxy threads: ", err) + ngx.thread.kill(self.co_client) + ngx.thread.kill(self.co_server) elseif res == "client" then --assert(self.client_state == _STATES.CLOSING) diff --git a/t/05-wss.t b/t/05-wss.t index 126ede6..512ceeb 100644 --- a/t/05-wss.t +++ b/t/05-wss.t @@ -46,7 +46,8 @@ qq{ } } } ---- config +--- config eval +qq{ location /proxy { content_by_lua_block { local proxy = require "resty.websocket.proxy" @@ -57,7 +58,11 @@ qq{ return ngx.exit(444) end - local ok, err = wp:connect("wss://127.0.0.1:9001/upstream") + -- proxy.lua now defaults to ssl_verify = true; the test upstream + -- uses a self-signed cert with no trusted CA configured, so + -- verification must be explicitly disabled here + local ok, err = wp:connect("wss://127.0.0.1:$ENV{TEST_NGINX_PORT2}/upstream", + {ssl_verify = false}) if not ok then ngx.log(ngx.ERR, err) return ngx.exit(444) @@ -83,6 +88,7 @@ qq{ ngx.say(data) } } +} --- response_body hello world! --- grep_error_log eval: qr/\[lua\].*/ diff --git a/t/06-error_handling.t b/t/06-error_handling.t index 47a4990..12e4245 100644 --- a/t/06-error_handling.t +++ b/t/06-error_handling.t @@ -44,7 +44,8 @@ qq{ } } } ---- config +--- config eval +qq{ location /proxy { content_by_lua_block { local proxy = require "resty.websocket.proxy" @@ -55,7 +56,7 @@ qq{ return ngx.exit(444) end - local ok, err = wp:connect("wss://127.0.0.1:9001/upstream") + local ok, err = wp:connect("wss://127.0.0.1:$ENV{TEST_NGINX_PORT2}/upstream") if not ok then ngx.log(ngx.ERR, err) return ngx.exit(444) @@ -78,6 +79,7 @@ qq{ wb:connect(uri) } } +} --- ignore_response_body --- error_log SSL_do_handshake() failed From a69554a18b51a779613faa42c377f3a2da67c9d9 Mon Sep 17 00:00:00 2001 From: Zeping Bu Date: Sun, 13 Sep 2026 20:32:10 +0800 Subject: [PATCH 3/3] test: update t/07-invalid_usage.t for the trimmed duplicate-connect warning The previous commit dropped the upstream URI from the 'connection with upstream already established' warning log (it may carry sensitive query-string data). Update this test's expected log pattern to match. --- t/07-invalid_usage.t | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/t/07-invalid_usage.t b/t/07-invalid_usage.t index db09d1c..47a7a19 100644 --- a/t/07-invalid_usage.t +++ b/t/07-invalid_usage.t @@ -53,7 +53,7 @@ __DATA__ hello world! --- grep_error_log eval: qr/\[(info|warn)\].*/ --- grep_error_log_out eval -qr/\A\[warn\] .*? connection with upstream at "ws:.*?" already established.* +qr/\A\[warn\] .*? connection with upstream already established.* \[info\] .*? frame type: text, payload: "hello world!"/ --- no_error_log [error]