This is an automated email from the ASF dual-hosted git repository.
bzp2010 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/apisix.git
The following commit(s) were added to refs/heads/master by this push:
new 271e799696 fix(websocket): address review findings on the ws/wss proxy
path (#13977)
271e799696 is described below
commit 271e799696370769ef3ab876f89f21951414dc3b
Author: Zeping Bai <[email protected]>
AuthorDate: Tue Sep 22 09:22:35 2026 +0800
fix(websocket): address review findings on the ws/wss proxy path (#13977)
---
apisix/init.lua | 97 ++++++++++----
apisix/plugins/traffic-split.lua | 4 +-
apisix/upstream.lua | 8 ++
docs/en/latest/admin-api.md | 2 +-
docs/zh/latest/admin-api.md | 2 +-
t/lib/server.lua | 82 +++++++++---
t/node/websocket-proxy.spec.mts | 273 ++++++++++++++++++++++++++++++++++++---
t/node/websocket-proxy.t | 4 +
8 files changed, 412 insertions(+), 60 deletions(-)
diff --git a/apisix/init.lua b/apisix/init.lua
index 9440ea9e13..b25069c744 100644
--- a/apisix/init.lua
+++ b/apisix/init.lua
@@ -310,8 +310,7 @@ end
-- host per upstream.pass_host: pass = client's Host, rewrite = configured
--- upstream_host, node = picked node's host[:port]. Also used directly by the
--- websocket phase, which has no nginx variable to fall back on for "pass".
+-- upstream_host, node = picked node's host[:port].
local function compute_upstream_host(api_ctx, picked_server)
local pass_host = api_ctx.pass_host or "pass"
if pass_host == "rewrite" then
@@ -347,6 +346,16 @@ local function set_upstream_headers(api_ctx, picked_server)
end
+-- "example.com:443" -> "example.com", "[::1]:443" -> "::1": the name a TLS
+-- handshake sends as SNI and verifies the certificate against, which must not
+-- carry the port an upstream Host header may have. Same as nginx does for
+-- proxy_ssl_name.
+local function host_without_port(host)
+ local m = ngx_re_match(host, [=[^(?:\[([^\]]+)\]|([^:]+))]=], "jo")
+ return m and (m[1] or m[2]) or host
+end
+
+
-- hop-by-hop headers, plus handshake headers connect() already sets itself
-- (host/protocols/origin opts, or generated Sec-WebSocket-Key/-Version).
local ws_skip_forward_headers = {
@@ -1068,6 +1077,7 @@ function _M.websocket_content_phase()
ngx.ctx = fetch_ctx()
local api_ctx = ngx.ctx.api_ctx
local up_conf = api_ctx.upstream_conf
+ local up_scheme = api_ctx.upstream_scheme
-- a Route's own `timeout` overrides upstream.timeout, same as
-- set_balancer_opts() does for the plain proxy_pass path
local route = api_ctx.matched_route
@@ -1082,7 +1092,7 @@ function _M.websocket_content_phase()
-- resolve upstream.tls once, same as https/grpcs in apisix/upstream.lua
local ssl_verify, client_cert, client_priv_key
- if api_ctx.matched_upstream.scheme == "wss" and up_conf.tls then
+ if up_scheme == "wss" and up_conf.tls then
ssl_verify = up_conf.tls.verify
if up_conf.tls.client_cert or up_conf.tls.client_cert_id then
@@ -1126,7 +1136,7 @@ function _M.websocket_content_phase()
upstream_new_opts = {max_recv_len = upstream_max_len, max_send_len =
client_max_len}
end
- local ok, proxy, err = pcall(ws_proxy.new, {
+ local proxy_opts = {
aggregate_fragments = true,
recv_timeout = recv_timeout_ms,
client_new_opts = client_new_opts,
@@ -1162,19 +1172,23 @@ function _M.websocket_content_phase()
local new_frame = role_handler.get_frame()
return new_frame.payload, new_frame.code
end
- })
- if not ok then
- ngx.log(ngx.ERR, "failed to create proxy: ", proxy)
- return core.response.exit(500)
- end
- if not proxy then
- ngx.log(ngx.ERR, "failed to create proxy: ", err)
- return core.response.exit(500)
+ }
+
+ -- A client that got a non-101 answer is marked fatal and its socket
+ -- closed (see resty.websocket.client), so it cannot serve a retry against
+ -- another node: every connection attempt gets a fresh proxy.
+ local function new_proxy()
+ local ok, proxy, err = pcall(ws_proxy.new, proxy_opts)
+ if not ok then
+ return nil, proxy
+ end
+
+ return proxy, err
end
- -- proxy:connect() only sends the 101 response to the downstream client
- -- after it has successfully connected upstream, so it's safe to retry
- -- against another node here without having committed to the client yet.
+ -- the 101 response goes to the downstream client only in connect_client(),
+ -- after an upstream connection has succeeded, so it's safe to retry
against
+ -- another node here without having committed to the client yet.
local retries = up_conf.retries
if not retries or retries < 0 then
retries = #up_conf.nodes - 1
@@ -1196,8 +1210,8 @@ function _M.websocket_content_phase()
request_uri = api_ctx.var.uri .. (api_ctx.var.is_args or "") ..
(api_ctx.var.args or "")
end
+ local proxy, ok, connect_err
local server = api_ctx.picked_server
- local ok, connect_err
for attempt = 0, retries do
if attempt > 0 and retry_deadline and retry_deadline < ngx_now() then
ngx.log(ngx.ERR, "websocket proxy retry timeout, retry count: ",
attempt,
@@ -1205,15 +1219,32 @@ function _M.websocket_content_phase()
return core.response.exit(502)
end
+ local err
+ proxy, err = new_proxy()
+ if not proxy then
+ ngx.log(ngx.ERR, "failed to create proxy: ", err)
+ return core.response.exit(500)
+ end
+
if connect_timeout_ms then
proxy.client:set_timeout(connect_timeout_ms)
end
- local endpoint = str_format("%s://%s:%d%s",
api_ctx.matched_upstream.scheme,
- server.host, server.port, request_uri)
- ok, connect_err = proxy:connect(endpoint, {
- host = compute_upstream_host(api_ctx, server),
- server_name = server.domain,
+ -- what proxy_pass would send as Host and use as SNI: honors a host
+ -- set by plugins such as proxy-rewrite, and follows a retried node
+ -- for pass_host = node
+ set_upstream_host(api_ctx, server)
+ local host = api_ctx.var.upstream_host
+ if not host or host == "" then
+ host = api_ctx.var.http_host
+ end
+
+ -- the request URI is left out of anything logged: its query string
+ -- may carry credentials
+ local node_addr = str_format("%s://%s:%d", up_scheme, server.host,
server.port)
+ ok, connect_err = proxy:connect_upstream(node_addr .. request_uri, {
+ host = host,
+ server_name = host_without_port(host),
headers = ws_headers,
protocols = ws_protocols,
origin = ws_origin,
@@ -1225,7 +1256,7 @@ function _M.websocket_content_phase()
break
end
- ngx.log(ngx.ERR, "failed to connect to websocket upstream ", endpoint,
+ ngx.log(ngx.ERR, "failed to connect to websocket upstream ", node_addr,
": ", connect_err)
-- no balancer_by_lua* here, so report the outcome ourselves; a parsed
@@ -1261,7 +1292,27 @@ function _M.websocket_content_phase()
return core.response.exit(502)
end
- local done, err = proxy:execute()
+ -- The server side of the proxy answers the client's Sec-WebSocket-Protocol
+ -- offer by echoing it back as it stands, which would announce a
subprotocol
+ -- the upstream never selected. Leave it exactly what the upstream picked,
+ -- or nothing, before completing the client handshake.
+ local resp_headers = proxy.client:get_resp_headers()
+ local selected = resp_headers and resp_headers.sec_websocket_protocol
+ if type(selected) == "table" then
+ selected = selected[1]
+ end
+ core.request.set_header(api_ctx, "Sec-WebSocket-Protocol", selected)
+
+ local done, err = proxy:connect_client()
+ if not done then
+ ngx.log(ngx.ERR, "failed to complete the client websocket handshake:
", err)
+ return core.response.exit(400)
+ end
+
+ -- there is no header filter phase on this path to do this on the 101
+ api_ctx.var.request_type = "websocket"
+
+ done, err = proxy:execute()
if not done then
ngx.log(ngx.ERR, "failed proxying: ", err)
return core.response.exit(502)
diff --git a/apisix/plugins/traffic-split.lua b/apisix/plugins/traffic-split.lua
index 35243f502c..f01f719c78 100644
--- a/apisix/plugins/traffic-split.lua
+++ b/apisix/plugins/traffic-split.lua
@@ -201,7 +201,9 @@ local function set_upstream(upstream_info, ctx)
end
core.log.info("upstream_key: ", upstream_key)
upstream.set(ctx, upstream_key, ctx.conf_version, up_conf)
- if upstream_info.scheme == "https" then
+ -- the schemes handle_upstream() dispatches on ctx.upstream_scheme for
+ local scheme = upstream_info.scheme
+ if scheme == "https" or scheme == "ws" or scheme == "wss" then
upstream.set_scheme(ctx, up_conf)
end
return
diff --git a/apisix/upstream.lua b/apisix/upstream.lua
index 9f061d8543..76865fc064 100644
--- a/apisix/upstream.lua
+++ b/apisix/upstream.lua
@@ -673,6 +673,14 @@ local function check_upstream_conf(in_dp, conf)
then
return false, "`upstream_host` can't be empty when `pass_host` is
`rewrite`"
end
+
+ -- the ws/wss client connects through a plain cosocket, which can only
+ -- trust the global lua_ssl_trusted_certificate, not a per-upstream
store
+ if (conf.scheme == "ws" or conf.scheme == "wss")
+ and conf.tls and conf.tls.ca_certs
+ then
+ return false, "`tls.ca_certs` is not supported by the `ws`/`wss`
scheme"
+ end
end
if conf.tls and conf.tls.client_cert then
diff --git a/docs/en/latest/admin-api.md b/docs/en/latest/admin-api.md
index 7b0ac85340..f3e48ca6d6 100644
--- a/docs/en/latest/admin-api.md
+++ b/docs/en/latest/admin-api.md
@@ -1020,7 +1020,7 @@ In addition to the equalization algorithm selections,
Upstream also supports pas
| tls.client_key | False, can't be used with `tls.client_cert_id`
| HTTPS certificate private key | Sets the client private key
while connecting to a TLS Upstream.
[...]
| tls.client_cert_id | False, can't be used with `tls.client_cert`
and `tls.client_key` | SSL | Set the referenced
[SSL](#ssl) id.
[...]
| tls.verify | False
| Boolean | Enables or disables
verification of the Upstream certificate. Falls back to the nginx configuration
when unset. Also used by the `kafka` scheme.
[...]
-| tls.ca_certs | False
| Array of HTTPS certificates | CA certificates used to
verify the Upstream certificate, replacing the ones loaded from
`ssl_trusted_certificate`.
[...]
+| tls.ca_certs | False
| Array of HTTPS certificates | CA certificates used to
verify the Upstream certificate, replacing the ones loaded from
`ssl_trusted_certificate`. Not supported when `scheme` is `ws` or `wss`, which
only trust `ssl_trusted_certificate`.
[...]
| keepalive_pool.size | False
| Auxiliary | Sets `keepalive` directive
dynamically.
[...]
| keepalive_pool.idle_timeout | False
| Auxiliary | Sets `keepalive_timeout`
directive dynamically.
[...]
| keepalive_pool.requests | False
| Auxiliary | Sets `keepalive_requests`
directive dynamically.
[...]
diff --git a/docs/zh/latest/admin-api.md b/docs/zh/latest/admin-api.md
index b08318c76b..b8f3bd82b7 100644
--- a/docs/zh/latest/admin-api.md
+++ b/docs/zh/latest/admin-api.md
@@ -1028,7 +1028,7 @@ APISIX 的 Upstream 除了基本的负载均衡算法选择外，还支持对上
| tls.client_key | 否，不能和 `tls.client_cert_id` 一起使用 |
https 证书私钥 | 设置跟上游通信时的客户端私钥，详细信息请参考下文。
| |
| tls.client_cert_id | 否，不能和 `tls.client_cert`、`tls.client_key` 一起使用 | SSL
| 设置引用的 SSL id，详见 [SSL](#ssl)。
| |
| tls.verify | 否 | Boolean
| 开启或关闭上游证书校验，不设置时沿用 nginx 的配置，详细信息请参考下文。Kafka 上游同样使用该字段。
| |
-| tls.ca_certs | 否 | https
证书数组 | 用于校验上游证书的 CA 证书，设置后将取代 `ssl_trusted_certificate` 中加载的证书，详细信息请参考下文。
| |
+| tls.ca_certs | 否 | https
证书数组 | 用于校验上游证书的 CA 证书，设置后将取代 `ssl_trusted_certificate`
中加载的证书，详细信息请参考下文。`scheme` 为 `ws` 或 `wss` 时不支持该字段，只会信任
`ssl_trusted_certificate`。
| |
|keepalive_pool.size | 否 | 辅助 |
动态设置 `keepalive` 指令，详细信息请参考下文。 |
|keepalive_pool.idle_timeout | 否
| 辅助 | 动态设置 `keepalive_timeout` 指令，详细信息请参考下文。 |
|keepalive_pool.requests | 否 | 辅助
| 动态设置 `keepalive_requests` 指令，详细信息请参考下文。 |
diff --git a/t/lib/server.lua b/t/lib/server.lua
index ab0782d7d1..b2f057dc4f 100644
--- a/t/lib/server.lua
+++ b/t/lib/server.lua
@@ -372,6 +372,26 @@ function _M.wolf_rbac_custom_headers()
end
+-- send_close/send_pong return bytes, err; log a failure instead of silently
+-- dropping it, so a broken close/pong shows up in the fixture's error log.
+local function ws_send_close(wb, code, msg)
+ local bytes, err = wb:send_close(code, msg)
+ if not bytes then
+ ngx.log(ngx.ERR, "failed to send close frame: ", err)
+ end
+ return bytes, err
+end
+
+
+local function ws_send_pong(wb, data)
+ local bytes, err = wb:send_pong(data)
+ if not bytes then
+ ngx.log(ngx.ERR, "failed to send pong frame: ", err)
+ end
+ return bytes, err
+end
+
+
function _M.websocket_handshake()
local websocket = require "resty.websocket.server"
local wb, err = websocket:new()
@@ -412,10 +432,10 @@ function _M.websocket_echo()
end
if typ == "close" then
- wb:send_close(1000, "")
+ ws_send_close(wb, 1000, "")
return
elseif typ == "ping" then
- wb:send_pong(data)
+ ws_send_pong(wb, data)
elseif typ == "text" or typ == "binary" then
local send = typ == "text" and wb.send_text or wb.send_binary
local bytes, send_err = send(wb, data)
@@ -430,6 +450,35 @@ function _M.websocket_echo()
end
+-- Like websocket_echo, but the node listening on 1981 refuses the handshake
+-- with a plain 503 instead: that node stays reachable at the TCP level, so an
+-- active tcp health check never marks it unhealthy on its own, while a
+-- websocket client sees a non-101 response and has to retry another node.
+function _M.websocket_echo_or_reject()
+ if ngx.var.server_port == "1981" then
+ return ngx.exit(503)
+ end
+
+ return _M.websocket_echo()
+end
+
+
+-- Like websocket_echo, but answers the handshake with the one subprotocol
+-- named by ?select=<name>, or with none at all for ?select=none (or no
+-- select), regardless of what the client offered. Falls into the same echo
+-- loop afterwards.
+function _M.websocket_subprotocol()
+ local select = ngx.var.arg_select
+ if select and select ~= "" and select ~= "none" then
+ ngx.req.set_header("Sec-WebSocket-Protocol", select)
+ else
+ ngx.req.clear_header("Sec-WebSocket-Protocol")
+ end
+
+ return _M.websocket_echo()
+end
+
+
-- Like websocket_echo, but with a raised max_payload_len (and, through it,
-- max_recv_len/max_send_len) so this fixture itself is never the bottleneck
-- for a >64K single-frame test: whatever the test observes then comes from
@@ -453,10 +502,10 @@ function _M.websocket_echo_large()
end
if typ == "close" then
- wb:send_close(1000, "")
+ ws_send_close(wb, 1000, "")
return
elseif typ == "ping" then
- wb:send_pong(data)
+ ws_send_pong(wb, data)
elseif typ == "text" or typ == "binary" then
local send = typ == "text" and wb.send_text or wb.send_binary
local bytes, send_err = send(wb, data)
@@ -494,10 +543,10 @@ function _M.websocket_ack_large()
end
if typ == "close" then
- wb:send_close(1000, "")
+ ws_send_close(wb, 1000, "")
return
elseif typ == "ping" then
- wb:send_pong(data)
+ ws_send_pong(wb, data)
elseif typ == "text" or typ == "binary" then
local bytes, send_err = wb:send_text("received:" .. #data)
if not bytes then
@@ -540,10 +589,10 @@ function _M.websocket_send_large()
end
if typ == "close" then
- wb:send_close(1000, "")
+ ws_send_close(wb, 1000, "")
return
elseif typ == "ping" then
- wb:send_pong(data)
+ ws_send_pong(wb, data)
elseif typ == "text" or typ == "binary" then
local send = typ == "text" and wb.send_text or wb.send_binary
local ok, echo_err = send(wb, data)
@@ -587,10 +636,10 @@ function _M.websocket_echo_uri()
end
if typ == "close" then
- wb:send_close(1000, "")
+ ws_send_close(wb, 1000, "")
return
elseif typ == "ping" then
- wb:send_pong(data)
+ ws_send_pong(wb, data)
elseif typ == "text" or typ == "binary" then
local send = typ == "text" and wb.send_text or wb.send_binary
local ok, echo_err = send(wb, data)
@@ -606,8 +655,8 @@ end
-- Like websocket_echo, but the first thing it sends back is a text frame
--- carrying the X-Real-IP/X-Forwarded-For it actually received as JSON, so a
--- test can confirm what a fronting proxy set them to. Falls into the same
+-- carrying the Host/X-Real-IP/X-Forwarded-For it actually received as JSON, so
+-- a test can confirm what a fronting proxy set them to. Falls into the same
-- echo loop afterwards.
function _M.websocket_echo_headers()
local websocket = require "resty.websocket.server"
@@ -619,6 +668,7 @@ function _M.websocket_echo_headers()
local headers = ngx.req.get_headers()
local bytes, send_err = wb:send_text(json_encode({
+ host = headers["Host"],
x_real_ip = headers["X-Real-IP"],
x_forwarded_for = headers["X-Forwarded-For"],
}))
@@ -638,10 +688,10 @@ function _M.websocket_echo_headers()
end
if typ == "close" then
- wb:send_close(1000, "")
+ ws_send_close(wb, 1000, "")
return
elseif typ == "ping" then
- wb:send_pong(data)
+ ws_send_pong(wb, data)
elseif typ == "text" or typ == "binary" then
local send = typ == "text" and wb.send_text or wb.send_binary
local ok, echo_err = send(wb, data)
@@ -691,7 +741,7 @@ function _M.websocket_fragment()
return
end
if typ == "close" then
- wb:send_close(1000, "")
+ ws_send_close(wb, 1000, "")
return
end
::continue::
@@ -709,7 +759,7 @@ function _M.websocket_close_upstream_initiated()
return ngx.exit(400)
end
- wb:send_close(1000, "bye")
+ ws_send_close(wb, 1000, "bye")
end
diff --git a/t/node/websocket-proxy.spec.mts b/t/node/websocket-proxy.spec.mts
index a1043b447e..d5b3735662 100644
--- a/t/node/websocket-proxy.spec.mts
+++ b/t/node/websocket-proxy.spec.mts
@@ -16,6 +16,8 @@
*/
import { describe, expect, it, jest } from '@jest/globals';
import axios from 'axios';
+import { readFileSync } from 'node:fs';
+import { type IncomingHttpHeaders, request } from 'node:http';
import WS from 'ws';
import { request as requestAdminAPI } from '../ts/admin_api';
@@ -53,7 +55,7 @@ const createRoute = async (
upstream,
plugins,
});
- expect(res.status).toBe(res.status < 300 ? res.status : 200);
+ expect(res.status).toBeLessThan(300);
// give etcd -> apisix config sync a moment to land before the first request
await wait(300);
return id;
@@ -67,15 +69,76 @@ const createRoute = async (
// that instant. PUTting the same fixed route id instead is a plain
// overwrite, so there's no delete in flight to race with.
const ECHO_ROUTE_ID = 'ws-proxy-echo';
-const putEchoRoute = async (upstream: object, plugins?: object) => {
- const res = await requestAdminAPI(`/apisix/admin/routes/${ECHO_ROUTE_ID}`,
'PUT', {
- uri: '/websocket_echo',
+const putRoute = async (id: string, uri: string, upstream: object, plugins?:
object) => {
+ const res = await requestAdminAPI(`/apisix/admin/routes/${id}`, 'PUT', {
+ uri,
upstream,
plugins,
});
- expect(res.status).toBe(res.status < 300 ? res.status : 200);
+ expect(res.status).toBeLessThan(300);
await wait(300);
};
+const putEchoRoute = (upstream: object, plugins?: object) =>
+ putRoute(ECHO_ROUTE_ID, '/websocket_echo', upstream, plugins);
+// same idea for the fixture that reports the handshake headers it received
+const putHeadersRoute = (upstream: object, plugins?: object) =>
+ putRoute('ws-proxy-headers', '/websocket_echo_headers', upstream, plugins);
+// and for the fixture whose 127.0.0.1:1981 node refuses the handshake with a
503
+const REJECT_ROUTE_ID = 'ws-proxy-reject';
+const putRejectRoute = (upstream: object, plugins?: object) =>
+ putRoute(REJECT_ROUTE_ID, '/websocket_echo_or_reject', upstream, plugins);
+
+// Opens a websocket connection and resolves with the first frame received,
+// without sending anything: for fixtures that speak first.
+const receiveFirst = (url: string, protocols?: string[]) =>
+ new Promise<{ data: string; protocol: string }>((resolve, reject) => {
+ const ws = new WS(url, protocols);
+ ws.on('message', (data) => {
+ resolve({ data: data.toString(), protocol: ws.protocol });
+ ws.close();
+ });
+ ws.on('error', reject);
+ });
+
+// Resolves with the headers of the 101 answer to a handshake that offers the
+// given subprotocols. A raw request rather than a WebSocket client, since the
+// clients reject a server that selects none of the subprotocols they offered,
+// which is exactly the answer some of these cases are about.
+const handshakeHeaders = (path: string, protocols: string[]) =>
+ new Promise<IncomingHttpHeaders>((resolve, reject) => {
+ const req = request({
+ host: '127.0.0.1',
+ port: 1984,
+ path,
+ headers: {
+ Connection: 'Upgrade',
+ Upgrade: 'websocket',
+ 'Sec-WebSocket-Key': 'dGhlIHNhbXBsZSBub25jZQ==',
+ 'Sec-WebSocket-Version': '13',
+ 'Sec-WebSocket-Protocol': protocols.join(', '),
+ },
+ });
+ req.on('upgrade', (res, socket) => {
+ socket.destroy();
+ resolve(res.headers);
+ });
+ req.on('response', (res) => reject(new Error(`unexpected status
${res.statusCode}`)));
+ req.on('error', reject);
+ req.end();
+ });
+
+// A plain http upgrade request, for asserting on the status the proxy itself
+// answers with when it cannot complete the handshake.
+const rawUpgrade = (path: string) =>
+ axios.get(`http://127.0.0.1:1984${path}`, {
+ headers: {
+ Connection: 'Upgrade',
+ Upgrade: 'websocket',
+ 'Sec-WebSocket-Key': 'dGhlIHNhbXBsZSBub25jZQ==',
+ 'Sec-WebSocket-Version': '13',
+ },
+ validateStatus: () => true,
+ });
// Opens a websocket connection, sends one text frame, resolves with the
// first frame received in reply (or rejects on error/close-before-reply).
@@ -124,6 +187,14 @@ describe('websocket-proxy (ws/wss upstream scheme)', () =>
{
// example-plugin's ws_client_frame/ws_upstream_frame hooks append
// "-client"/"-upstream" to every text frame they see, in-flight.
'example-plugin': { i: 1 },
+ // the log phase runs once the session is over: report the request
+ // type there, which only a websocket session should have set
+ 'serverless-post-function': {
+ phase: 'log',
+ functions: [
+ 'return function(conf, ctx) ngx.log(ngx.WARN, "ws request_type:
", ctx.var.request_type) end',
+ ],
+ },
},
);
@@ -349,31 +420,43 @@ describe('websocket-proxy (ws/wss upstream scheme)', ()
=> {
});
describe('passive health check', () => {
- it('marks a node unhealthy after enough failed connection attempts', async
() => {
- await putEchoRoute({
+ it('marks a node unhealthy after it answers the handshake with a failing
status', async () => {
+ // 127.0.0.1:1981 accepts TCP connections but answers every handshake
with
+ // a 503 (see websocket_echo_or_reject), so the active tcp check below
can
+ // never flag it on its own: only the passive http status report the
proxy
+ // makes for the non-101 response can move it to unhealthy.
+ await putRejectRoute({
type: 'roundrobin',
scheme: 'ws',
retries: 1,
- nodes: { [DEAD_NODE]: 1, [ECHO_NODE]: 1 },
+ nodes: { '127.0.0.1:1981': 1, [ECHO_NODE]: 1 },
checks: {
- active: { type: 'tcp', http_path: '/', timeout: 1, healthy: {
interval: 1 } },
- passive: { unhealthy: { tcp_failures: 1 } },
+ // probes only once at startup and then stay out of the way, so they
can
+ // neither flag the node unhealthy nor flip it back to healthy again
+ active: {
+ type: 'tcp',
+ host: '127.0.0.1',
+ timeout: 1,
+ healthy: { interval: 3600 },
+ unhealthy: { interval: 3600 },
+ },
+ passive: { unhealthy: { http_statuses: [503], http_failures: 1 } },
},
});
- // one connect attempt is enough to report a tcp failure for DEAD_NODE
- await sendAndReceive('/websocket_echo', 'hello');
-
let unhealthyFound = false;
for (let i = 0; i < 10 && !unhealthyFound; i++) {
+ // each request may or may not pick the 503 node first; the retry makes
+ // it succeed either way, and a pick of that node reports the failure
+ expect(await sendAndReceive('/websocket_echo_or_reject',
'hello')).toBe('hello');
await wait(500);
- const res = await
requestAdminAPI(`/v1/healthcheck/routes/${ECHO_ROUTE_ID}`);
- const { nodes } = res.data as { nodes: { ip: string; port: number;
status: string }[] };
- unhealthyFound = nodes.some((n) => n.port === 1 && n.status !==
'healthy');
+ const res = await
requestAdminAPI(`/v1/healthcheck/routes/${REJECT_ROUTE_ID}`);
+ const { nodes } = res.data as { nodes: { port: number; status: string
}[] };
+ unhealthyFound = nodes.some((n) => n.port === 1981 && n.status !==
'healthy');
}
expect(unhealthyFound).toBe(true);
- }, 15000);
+ }, 30000);
});
describe('upstream URI forwarding', () => {
@@ -426,7 +509,7 @@ describe('websocket-proxy (ws/wss upstream scheme)', () => {
describe('client address headers', () => {
it('overrides X-Real-IP and appends this hop to X-Forwarded-For, not what
the client sent', async () => {
- await createRoute('/websocket_echo_headers', {
+ await putHeadersRoute({
type: 'roundrobin',
scheme: 'ws',
nodes: { [ECHO_NODE]: 1 },
@@ -451,6 +534,160 @@ describe('websocket-proxy (ws/wss upstream scheme)', ()
=> {
});
});
+ describe('upstream Host header', () => {
+ it('honors the host set by proxy-rewrite on the upstream handshake', async
() => {
+ await putHeadersRoute(
+ { type: 'roundrobin', scheme: 'ws', nodes: { [ECHO_NODE]: 1 } },
+ { 'proxy-rewrite': { host: 'rewritten.example.com' } },
+ );
+
+ const { data } = await
receiveFirst('ws://127.0.0.1:1984/websocket_echo_headers');
+ expect(JSON.parse(data).host).toBe('rewritten.example.com');
+ });
+
+ it("sends the retried node's own host with pass_host: node", async () => {
+ await putHeadersRoute({
+ type: 'roundrobin',
+ scheme: 'ws',
+ pass_host: 'node',
+ retries: 1,
+ nodes: { [DEAD_NODE]: 100, [ECHO_NODE]: 1 },
+ });
+
+ const { data } = await
receiveFirst('ws://127.0.0.1:1984/websocket_echo_headers');
+ expect(JSON.parse(data).host).toBe(ECHO_NODE);
+ });
+ });
+
+ describe('wss upstream', () => {
+ // the fake server's TLS listener; its certificate is issued for test.com
+ const TLS_NODE = '127.0.0.1:1983';
+
+ it('proxies over TLS with certificate verification off', async () => {
+ await putHeadersRoute({
+ type: 'roundrobin',
+ scheme: 'wss',
+ tls: { verify: false },
+ nodes: { [TLS_NODE]: 1 },
+ });
+
+ const { data } = await
receiveFirst('ws://127.0.0.1:1984/websocket_echo_headers');
+ expect(JSON.parse(data).host).toBe('127.0.0.1:1984');
+ });
+
+ it('verifies the certificate against the upstream host, port excluded',
async () => {
+ await putHeadersRoute({
+ type: 'roundrobin',
+ scheme: 'wss',
+ pass_host: 'rewrite',
+ upstream_host: 'test.com:1983',
+ tls: { verify: true },
+ nodes: { [TLS_NODE]: 1 },
+ });
+
+ const { data } = await
receiveFirst('ws://127.0.0.1:1984/websocket_echo_headers');
+ expect(JSON.parse(data).host).toBe('test.com:1983');
+ });
+
+ it('refuses an upstream whose certificate does not match the host', async
() => {
+ await putHeadersRoute({
+ type: 'roundrobin',
+ scheme: 'wss',
+ tls: { verify: true },
+ nodes: { [TLS_NODE]: 1 },
+ });
+
+ const res = await rawUpgrade('/websocket_echo_headers');
+ expect(res.status).toBe(502);
+ });
+
+ it('rejects tls.ca_certs, which the ws/wss client cannot apply', async ()
=> {
+ const cert = readFileSync(new URL('../certs/apisix.crt',
import.meta.url), 'utf8');
+ const res = await requestAdminAPI(
+ '/apisix/admin/upstreams/ws-proxy-ca-certs',
+ 'PUT',
+ {
+ type: 'roundrobin',
+ scheme: 'wss',
+ tls: { verify: true, ca_certs: [cert] },
+ nodes: { [TLS_NODE]: 1 },
+ },
+ undefined,
+ { validateStatus: () => true },
+ );
+ expect(res.status).toBe(400);
+ });
+ });
+
+ describe('subprotocol negotiation', () => {
+ it('answers the client with the subprotocol the upstream selected', async
() => {
+ await createRoute('/websocket_subprotocol', {
+ type: 'roundrobin',
+ scheme: 'ws',
+ nodes: { [ECHO_NODE]: 1 },
+ });
+
+ const headers = await
handshakeHeaders('/websocket_subprotocol?select=chat', [
+ 'other',
+ 'chat',
+ ]);
+ expect(headers['sec-websocket-protocol']).toBe('chat');
+ });
+
+ it('answers with no subprotocol when the upstream selected none', async ()
=> {
+ // echoing the client's whole offer back instead would announce
+ // subprotocols the upstream never agreed to
+ const headers = await
handshakeHeaders('/websocket_subprotocol?select=none', [
+ 'other',
+ 'chat',
+ ]);
+ expect(headers['sec-websocket-protocol']).toBeUndefined();
+ });
+ });
+
+ describe('traffic-split', () => {
+ it('proxies frames through a ws upstream chosen by traffic-split', async
() => {
+ // the route's own upstream is plain http: only the traffic-split pick
is ws
+ await putEchoRoute(
+ { type: 'roundrobin', scheme: 'http', nodes: { [ECHO_NODE]: 1 } },
+ {
+ 'traffic-split': {
+ rules: [
+ {
+ weighted_upstreams: [
+ {
+ upstream: {
+ type: 'roundrobin',
+ scheme: 'ws',
+ nodes: { [ECHO_NODE]: 1 },
+ },
+ weight: 1,
+ },
+ ],
+ },
+ ],
+ },
+ },
+ );
+
+ expect(await sendAndReceive('/websocket_echo', 'hello')).toBe('hello');
+ });
+ });
+
+ describe('upstream retry after a non-101 handshake', () => {
+ it('retries the next node and completes the session on it', async () => {
+ // 127.0.0.1:1981 answers the handshake with a 503
(websocket_echo_or_reject)
+ await putRejectRoute({
+ type: 'roundrobin',
+ scheme: 'ws',
+ retries: 1,
+ nodes: { '127.0.0.1:1981': 100, [ECHO_NODE]: 1 },
+ });
+
+ expect(await sendAndReceive('/websocket_echo_or_reject',
'hello')).toBe('hello');
+ });
+ });
+
describe('frame size (websocket-proxy plugin)', () => {
it('closes the connection on a single frame over the 65535-byte default',
async () => {
await putEchoRoute({
diff --git a/t/node/websocket-proxy.t b/t/node/websocket-proxy.t
index 73b8fe9251..dc62eac3e0 100644
--- a/t/node/websocket-proxy.t
+++ b/t/node/websocket-proxy.t
@@ -29,6 +29,10 @@ __DATA__
--- max_size: 2048000
--- exec
cd t && pnpm test node/websocket-proxy.spec.mts 2>&1
+--- error_log
+plugin ws_handshake phase
+plugin ws_close phase
+ws request_type: websocket
--- no_error_log
failed to execute the script with status
--- response_body eval
