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 52eb4e31c6 feat(websocket): add enhanced proxy and plugin hook (#13939)
52eb4e31c6 is described below
commit 52eb4e31c6fdcc45a4426dcc55473faf59a3888c
Author: Zeping Bai <[email protected]>
AuthorDate: Fri Sep 18 16:28:16 2026 +0800
feat(websocket): add enhanced proxy and plugin hook (#13939)
---
apisix-master-0.rockspec | 1 +
apisix/balancer.lua | 48 +++-
apisix/cli/ngx_tpl.lua | 10 +
apisix/core.lua | 1 +
apisix/core/websocket.lua | 86 +++++++
apisix/init.lua | 280 +++++++++++++++++++-
apisix/plugins/example-plugin.lua | 28 ++
apisix/schema_def.lua | 4 +-
apisix/upstream.lua | 2 +
docs/en/latest/admin-api.md | 11 +-
docs/en/latest/plugin-develop.md | 27 ++
docs/en/latest/terminology/plugin.md | 2 +
docs/zh/latest/admin-api.md | 11 +-
t/APISIX.pm | 27 ++
t/lib/server.lua | 219 ++++++++++++++++
t/node/websocket-proxy.spec.mts | 479 +++++++++++++++++++++++++++++++++++
t/node/websocket-proxy.t | 35 +++
t/package.json | 2 +
t/pnpm-lock.yaml | 27 ++
19 files changed, 1275 insertions(+), 25 deletions(-)
diff --git a/apisix-master-0.rockspec b/apisix-master-0.rockspec
index 40b41a0222..8c1fc783bd 100644
--- a/apisix-master-0.rockspec
+++ b/apisix-master-0.rockspec
@@ -32,6 +32,7 @@ description = {
dependencies = {
"lua-resty-ctxdump = 0.1-0",
+ "api7-lua-resty-websocket = 0.1.0-0",
"api7-lua-resty-redis-connector = 0.13.0",
"lyaml = 6.2.8-1",
"api7-lua-resty-dns-client = 7.1.2-0",
diff --git a/apisix/balancer.lua b/apisix/balancer.lua
index 0258d4bb54..132cf89378 100644
--- a/apisix/balancer.lua
+++ b/apisix/balancer.lua
@@ -249,10 +249,30 @@ local function
parse_server_for_upstream_host(picked_server, upstream_scheme)
end
+-- reports a connection outcome (get_last_failure()-shaped state/code) for the
+-- node ctx.balancer_ip/balancer_port currently point at
+local function report_failure(ctx, checker, up_conf, state, code)
+ local host = up_conf.checks and up_conf.checks.active and
up_conf.checks.active.host
+ local port = up_conf.checks and up_conf.checks.active and
up_conf.checks.active.port
+ if state == "failed" then
+ if code == 504 then
+ checker:report_timeout(ctx.balancer_ip, port or ctx.balancer_port,
host)
+ else
+ checker:report_tcp_failure(ctx.balancer_ip, port or
ctx.balancer_port, host)
+ end
+ else
+ checker:report_http_status(ctx.balancer_ip, port or ctx.balancer_port,
host, code)
+ end
+end
+
+
-- pick_server will be called:
-- 1. in the access phase so that we can set headers according to the picked
server
-- 2. each time we need to retry upstream
-local function pick_server(route, ctx)
+--
+-- prev_failure, when given, overrides get_last_failure() for callers outside
+-- balancer_by_lua* that already know their own connection's outcome.
+local function pick_server(route, ctx, prev_failure)
local up_conf = ctx.upstream_conf
local nodes_count = #up_conf.nodes
@@ -293,18 +313,13 @@ local function pick_server(route, ctx)
end
if checker then
- local state, code = get_last_failure()
- local host = up_conf.checks and up_conf.checks.active and
up_conf.checks.active.host
- local port = up_conf.checks and up_conf.checks.active and
up_conf.checks.active.port
- if state == "failed" then
- if code == 504 then
- checker:report_timeout(ctx.balancer_ip, port or
ctx.balancer_port, host)
- else
- checker:report_tcp_failure(ctx.balancer_ip, port or
ctx.balancer_port, host)
- end
+ local state, code
+ if prev_failure then
+ state, code = prev_failure.state, prev_failure.code
else
- checker:report_http_status(ctx.balancer_ip, port or
ctx.balancer_port, host, code)
+ state, code = get_last_failure()
end
+ report_failure(ctx, checker, up_conf, state, code)
end
end
@@ -385,6 +400,17 @@ end
_M.pick_server = pick_server
+-- reports a final failure with no next node to pick_server() for
+function _M.report_failure(ctx, prev_failure)
+ local checker = ctx.up_checker
+ if not checker then
+ return
+ end
+
+ report_failure(ctx, checker, ctx.upstream_conf, prev_failure.state,
prev_failure.code)
+end
+
+
-- Keyed by the `ca_certs` array itself: a config update always rebuilds that
-- table, so a stale digest can never outlive the certificates it was made
from.
local ca_certs_digest_cache = core.lrucache.new({
diff --git a/apisix/cli/ngx_tpl.lua b/apisix/cli/ngx_tpl.lua
index 567f288b15..8688fcfe20 100644
--- a/apisix/cli/ngx_tpl.lua
+++ b/apisix/cli/ngx_tpl.lua
@@ -1184,6 +1184,16 @@ http {
}
{% end %}
+ location @websocket_pass {
+ content_by_lua_block {
+ apisix.websocket_content_phase()
+ }
+
+ log_by_lua_block {
+ apisix.websocket_log_phase()
+ }
+ }
+
{% if enabled_plugins["proxy-mirror"] then %}
location = /proxy_mirror {
internal;
diff --git a/apisix/core.lua b/apisix/core.lua
index fceb7d6a0d..2c2bc2d23c 100644
--- a/apisix/core.lua
+++ b/apisix/core.lua
@@ -66,4 +66,5 @@ return {
event = require("apisix.core.event"),
env = require("apisix.core.env"),
data_encryption = require("apisix.core.data_encryption"),
+ websocket = require("apisix.core.websocket"),
}
diff --git a/apisix/core/websocket.lua b/apisix/core/websocket.lua
new file mode 100644
index 0000000000..64aebf5754
--- /dev/null
+++ b/apisix/core/websocket.lua
@@ -0,0 +1,86 @@
+--
+-- Licensed to the Apache Software Foundation (ASF) under one or more
+-- contributor license agreements. See the NOTICE file distributed with
+-- this work for additional information regarding copyright ownership.
+-- The ASF licenses this file to You 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.
+--
+local ngx = ngx
+local tostring = tostring
+
+local ROLE_CLIENT = "client"
+local ROLE_UPSTREAM = "upstream"
+local CTX_KEY_CLIENT = "websocket_client"
+local CTX_KEY_UPSTREAM = "websocket_upstream"
+
+-- ngx.ctx is per-request and can only be accessed from within a request
+-- context, so it must be fetched inside each wrapped function, not cached
+-- as a module-level upvalue at require() time (which also runs during
+-- init_by_lua, before any request exists).
+
+local function wrap_stash_frame(key)
+ return function(frame)
+ ngx.ctx[key] = frame
+ end
+end
+
+
+local function wrap_get_frame(key)
+ return function()
+ return ngx.ctx[key]
+ end
+end
+
+
+local function wrap_set_frame_data(key)
+ return function(data)
+ ngx.ctx[key].payload = data
+ end
+end
+
+
+local function wrap_set_status(key)
+ return function(status)
+ ngx.ctx[key].code = status
+ end
+end
+
+
+local _M = {
+ ROLE_CLIENT = ROLE_CLIENT,
+ ROLE_UPSTREAM = ROLE_UPSTREAM,
+ [ROLE_CLIENT] = {
+ stash_frame = wrap_stash_frame(CTX_KEY_CLIENT),
+ get_frame = wrap_get_frame(CTX_KEY_CLIENT),
+ set_frame_data = wrap_set_frame_data(CTX_KEY_CLIENT),
+ set_status = wrap_set_status(CTX_KEY_CLIENT),
+ --drop_frame = wrap_drop_frame
+ },
+ [ROLE_UPSTREAM] = {
+ stash_frame = wrap_stash_frame(CTX_KEY_UPSTREAM),
+ get_frame = wrap_get_frame(CTX_KEY_UPSTREAM),
+ set_frame_data = wrap_set_frame_data(CTX_KEY_UPSTREAM),
+ set_status = wrap_set_status(CTX_KEY_UPSTREAM),
+ },
+}
+
+function _M.get_role(role)
+ if role == ROLE_CLIENT or role == "client" then
+ return _M[ROLE_CLIENT]
+ elseif role == ROLE_UPSTREAM or role == "upstream" then
+ return _M[ROLE_UPSTREAM]
+ else
+ return nil, "invalid role: " .. tostring(role)
+ end
+end
+
+return _M
diff --git a/apisix/init.lua b/apisix/init.lua
index b4ef63195e..4e78d2335a 100644
--- a/apisix/init.lua
+++ b/apisix/init.lua
@@ -26,6 +26,7 @@ require("jit.opt").start("minstitch=2", "maxtrace=4000",
"maxmcode=4000", "maxirconst=1000")
require("apisix.patch").patch()
+local ws_proxy = require("resty.websocket.proxy")
local core = require("apisix.core")
local plugin = require("apisix.plugin")
local plugin_config = require("apisix.plugin_config")
@@ -62,6 +63,9 @@ local re_gsub = ngx.re.gsub
local str_byte = string.byte
local str_sub = string.sub
local str_char = string.char
+local str_format = string.format
+local str_find = string.find
+local str_lower = string.lower
local tonumber = tonumber
local type = type
local pairs = pairs
@@ -305,6 +309,23 @@ local function parse_domain_in_route(route)
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".
+local function compute_upstream_host(api_ctx, picked_server)
+ local pass_host = api_ctx.pass_host or "pass"
+ if pass_host == "rewrite" then
+ return api_ctx.upstream_host
+ end
+
+ if pass_host == "node" then
+ return picked_server.upstream_host
+ end
+
+ return api_ctx.var.http_host
+end
+
+
local function set_upstream_host(api_ctx, picked_server)
local up_conf = api_ctx.upstream_conf
if up_conf.pass_host then
@@ -317,12 +338,7 @@ local function set_upstream_host(api_ctx, picked_server)
return
end
- if pass_host == "rewrite" then
- api_ctx.var.upstream_host = api_ctx.upstream_host
- return
- end
-
- api_ctx.var.upstream_host = picked_server.upstream_host
+ api_ctx.var.upstream_host = compute_upstream_host(api_ctx, picked_server)
end
@@ -331,6 +347,46 @@ local function set_upstream_headers(api_ctx, picked_server)
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 = {
+ ["host"] = true,
+ ["connection"] = true,
+ ["upgrade"] = true,
+ ["keep-alive"] = true,
+ ["te"] = true,
+ ["trailers"] = true,
+ ["proxy-authenticate"] = true,
+ ["proxy-authorization"] = true,
+ ["content-length"] = true,
+ ["transfer-encoding"] = true,
+ ["sec-websocket-key"] = true,
+ ["sec-websocket-version"] = true,
+ ["sec-websocket-extensions"] = true,
+ ["sec-websocket-protocol"] = true,
+ ["origin"] = true,
+}
+
+
+-- forwards the client's other headers (Cookie, Authorization, ...) upstream.
+local function build_ws_forward_headers(api_ctx)
+ local headers = {}
+ for name, value in pairs(core.request.headers(api_ctx)) do
+ if not ws_skip_forward_headers[str_lower(name)] then
+ if type(value) == "table" then
+ for _, v in ipairs(value) do
+ headers[#headers + 1] = name .. ": " .. v
+ end
+ else
+ headers[#headers + 1] = name .. ": " .. value
+ end
+ end
+ end
+
+ return headers
+end
+
+
-- verify the TLS session resumption by checking if the SNI in the client hello
-- matches the hostname of the SSL session, this is to prevent the mTLS bypass
security issue.
local function verify_tls_session_resumption()
@@ -675,6 +731,24 @@ function _M.handle_upstream(api_ctx, route,
enable_websocket)
return ngx.exec("@grpc_pass")
end
+ if up_scheme == "wss" or up_scheme == "ws" then
+ -- @websocket_pass never runs proxy_pass, so it never gets the
+ -- `proxy_set_header X-Real-IP $remote_addr` / `... X-Forwarded-For
+ -- $proxy_add_x_forwarded_for` that ngx_tpl.lua's proxy_pass location
+ -- sends: set the same values here, once, so ws_handshake and later
+ -- phases see what proxy_pass routes would have seen, and
+ -- build_ws_forward_headers() (apisix/init.lua) can just forward them
+ -- like any other header instead of special-casing these two.
+ core.request.set_header(api_ctx, "X-Real-IP", api_ctx.var.remote_addr)
+ core.request.set_header(api_ctx, "X-Forwarded-For",
+ api_ctx.var.proxy_add_x_forwarded_for)
+
+ common_phase("ws_handshake")
+
+ stash_ngx_ctx()
+ return ngx.exec("@websocket_pass")
+ end
+
if api_ctx.dubbo_proxy_enabled then
stash_ngx_ctx()
return ngx.exec("@dubbo_pass")
@@ -989,6 +1063,200 @@ function _M.grpc_access_phase()
end
+-- call ws_x_frame hook
+function _M.websocket_content_phase()
+ ngx.ctx = fetch_ctx()
+ local api_ctx = ngx.ctx.api_ctx
+ local up_conf = api_ctx.upstream_conf
+ -- 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
+ local up_timeout = (route and route.value and route.value.timeout) or
up_conf.timeout
+ local connect_timeout_ms = up_timeout and up_timeout.connect and
up_timeout.connect * 1000
+ local recv_timeout_ms = up_timeout and up_timeout.read and up_timeout.read
* 1000
+ -- upstream.timeout.send is silently ignored for ws/wss
+
+ local ws_headers = build_ws_forward_headers(api_ctx)
+ local ws_protocols = core.request.header(api_ctx, "Sec-WebSocket-Protocol")
+ local ws_origin = core.request.header(api_ctx, "Origin")
+
+ -- 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
+ ssl_verify = up_conf.tls.verify
+
+ if up_conf.tls.client_cert or up_conf.tls.client_cert_id then
+ local cert_pem, key_pem
+ if up_conf.tls.client_cert_id then
+ cert_pem = api_ctx.upstream_ssl and api_ctx.upstream_ssl.cert
+ key_pem = api_ctx.upstream_ssl and api_ctx.upstream_ssl.key
+ else
+ cert_pem = up_conf.tls.client_cert
+ key_pem = up_conf.tls.client_key
+ end
+
+ local cert_err, key_err
+ client_cert, cert_err =
apisix_ssl.fetch_cert(api_ctx.var.upstream_host, cert_pem)
+ if not client_cert then
+ ngx.log(ngx.ERR, "failed to fetch websocket upstream client
cert: ", cert_err)
+ return core.response.exit(503)
+ end
+
+ client_priv_key, key_err =
apisix_ssl.fetch_pkey(api_ctx.var.upstream_host, key_pem)
+ if not client_priv_key then
+ ngx.log(ngx.ERR, "failed to fetch websocket upstream client
key: ", key_err)
+ return core.response.exit(503)
+ end
+ end
+ end
+
+ local ok, proxy, err = pcall(ws_proxy.new, {
+ aggregate_fragments = true,
+ recv_timeout = recv_timeout_ms,
+ on_frame = function(proxy, role, typ, payload, last, code)
+ -- proxy: [table] the proxy instance
+ -- role: [string] "client" or "upstream"
+ -- typ: [string] "text", "binary", "ping", "pong", "close"
+ -- payload: [string|nil] payload if any
+ -- last: [boolean] fin flag; true when aggregate_fragments
is on
+ -- code: [number|nil] code for "close" frames
+
+ local role_handler, err = core.websocket.get_role(role)
+ if not role_handler then
+ ngx.log(ngx.ERR, "invalid websocket role: ", err)
+ return
+ end
+
+ role_handler.stash_frame({
+ proxy = proxy,
+ type = typ,
+ payload = payload,
+ last = last,
+ code = code,
+ })
+
+ if role == "client" then
+ common_phase("ws_client_frame")
+ else
+ common_phase("ws_upstream_frame")
+ end
+
+ 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)
+ 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.
+ local retries = up_conf.retries
+ if not retries or retries < 0 then
+ retries = #up_conf.nodes - 1
+ end
+
+ local retry_deadline
+ if retries > 0 and up_conf.retry_timeout and up_conf.retry_timeout > 0 then
+ retry_deadline = ngx_now() + up_conf.retry_timeout
+ end
+
+ -- upstream_uri is only ever set by plugins like proxy-rewrite that
+ -- explicitly rewrite the forwarded path; the normal proxy_pass paths get
+ -- the client's original request URI for free from nginx's own passthrough
+ -- behavior, but we build the request line ourselves here, so we have to
+ -- fall back to the client's URI (plus query string) the same way
+ -- proxy-mirror.lua does.
+ local request_uri = api_ctx.var.upstream_uri
+ if not request_uri or request_uri == "" then
+ request_uri = api_ctx.var.uri .. (api_ctx.var.is_args or "") ..
(api_ctx.var.args or "")
+ end
+
+ 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,
+ ", deadline: ", retry_deadline, " now: ", ngx_now())
+ return core.response.exit(502)
+ 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,
+ headers = ws_headers,
+ protocols = ws_protocols,
+ origin = ws_origin,
+ ssl_verify = ssl_verify,
+ client_cert = client_cert,
+ client_priv_key = client_priv_key,
+ })
+ if ok then
+ break
+ end
+
+ ngx.log(ngx.ERR, "failed to connect to websocket upstream ", endpoint,
+ ": ", connect_err)
+
+ -- no balancer_by_lua* here, so report the outcome ourselves; a parsed
+ -- HTTP status (just not 101) is a passive HTTP status report, not
tcp_failure
+ local prev_failure
+ local resp_status_code = proxy.client.resp_status_code
+ if resp_status_code then
+ prev_failure = {state = "ok", code = tonumber(resp_status_code)}
+ elseif connect_err and str_find(connect_err, "timeout", 1, true) then
+ prev_failure = {state = "failed", code = 504}
+ else
+ prev_failure = {state = "failed", code = 599}
+ end
+
+ if attempt >= retries then
+ -- last attempt: report it, pick_server() won't be called again
+ load_balancer.report_failure(api_ctx, prev_failure)
+ break
+ end
+
+ local next_server, pick_err =
load_balancer.pick_server(api_ctx.matched_route,
+ api_ctx,
prev_failure)
+ if not next_server then
+ ngx.log(ngx.ERR, "failed to pick next websocket upstream server:
", pick_err)
+ break
+ end
+
+ server = next_server
+ api_ctx.picked_server = server
+ end
+
+ if not ok then
+ return core.response.exit(502)
+ end
+
+ local done, err = proxy:execute()
+ if not done then
+ ngx.log(ngx.ERR, "failed proxying: ", err)
+ return core.response.exit(502)
+ end
+end
+
+
+function _M.websocket_log_phase()
+ common_phase("ws_close")
+ _M.http_log_phase()
+end
+
+
local function set_resp_upstream_status(up_status)
local_conf = core.config.local_conf()
diff --git a/apisix/plugins/example-plugin.lua
b/apisix/plugins/example-plugin.lua
index 767ccfae72..e76ce72e9e 100644
--- a/apisix/plugins/example-plugin.lua
+++ b/apisix/plugins/example-plugin.lua
@@ -128,6 +128,34 @@ function _M.log(conf, ctx)
end
+function _M.ws_handshake(conf, ctx)
+ core.log.warn("plugin ws_handshake phase, conf: ", core.json.encode(conf))
+end
+
+
+function _M.ws_client_frame(conf, ctx)
+ local frame = core.websocket.client.get_frame()
+ core.log.warn("plugin ws_client_frame phase, type: ", frame.type)
+ if frame.type == "text" and frame.payload then
+ core.websocket.client.set_frame_data(frame.payload .. "-client")
+ end
+end
+
+
+function _M.ws_upstream_frame(conf, ctx)
+ local frame = core.websocket.upstream.get_frame()
+ core.log.warn("plugin ws_upstream_frame phase, type: ", frame.type)
+ if frame.type == "text" and frame.payload then
+ core.websocket.upstream.set_frame_data(frame.payload .. "-upstream")
+ end
+end
+
+
+function _M.ws_close(conf, ctx)
+ core.log.warn("plugin ws_close phase, conf: ", core.json.encode(conf))
+end
+
+
local function hello()
local args = ngx.req.get_uri_args()
if args["json"] then
diff --git a/apisix/schema_def.lua b/apisix/schema_def.lua
index 0ff40e5646..64f661f821 100644
--- a/apisix/schema_def.lua
+++ b/apisix/schema_def.lua
@@ -549,9 +549,9 @@ local upstream_schema = {
scheme = {
default = "http",
enum = {"grpc", "grpcs", "http", "https", "tcp", "tls", "udp",
- "kafka"},
+ "kafka", "ws", "wss"},
description = "The scheme of the upstream." ..
- " For L7 proxy, it can be one of grpc/grpcs/http/https." ..
+ " For L7 proxy, it can be one of
grpc/grpcs/http/https/ws/wss." ..
" For L4 proxy, it can be one of tcp/tls/udp." ..
" For specific protocols, it can be kafka."
},
diff --git a/apisix/upstream.lua b/apisix/upstream.lua
index acb6ab3663..9f061d8543 100644
--- a/apisix/upstream.lua
+++ b/apisix/upstream.lua
@@ -190,6 +190,8 @@ local scheme_to_port = {
https = 443,
grpc = 80,
grpcs = 443,
+ ws = 80,
+ wss = 443,
}
diff --git a/docs/en/latest/admin-api.md b/docs/en/latest/admin-api.md
index 77d8d76969..7b0ac85340 100644
--- a/docs/en/latest/admin-api.md
+++ b/docs/en/latest/admin-api.md
@@ -368,7 +368,7 @@ ID's as a text string must be of a length between 1 and 64
characters and they s
| plugin_config_id | False, can't be used with `script` | Plugin |
[Plugin config](terminology/plugin-config.md) bound to the Route.
|
|
| labels | False | Match Rules |
Attributes of the Route specified as key-value pairs.
|
{"version":"v2","build":"16","env":"production"} |
| timeout | False | Auxiliary |
Sets the timeout (in seconds) for connecting to, and sending and receiving
messages between the Upstream and the Route. This will overwrite the `timeout`
value configured in your [Upstream](#upstream).
| {"connect":
3, "send": 3, "read": 3} |
-| enable_websocket | False | Auxiliary |
Enables a websocket. Set to `false` by default.
|
|
+| enable_websocket | False | Auxiliary |
Enables a websocket. Set to `false` by default. This is a plain protocol
upgrade with no access to individual frames; see the note under Upstream
[`scheme`](#upstream) if a plugin needs to inspect or rewrite them.
|
|
| status | False | Auxiliary |
Enables the current Route. Set to `1` (enabled) by default.
| `1` to enable, `0` to disable
|
Example configuration:
@@ -671,7 +671,7 @@ Service resource request address:
/apisix/admin/services/{id}
| name | False | Auxiliary | Identifier for the Service.
| service-xxxx |
| desc | False | Auxiliary | Description of usage scenarios.
| service xxxx |
| labels | False | Match Rules | Attributes of the Service
specified as key-value pairs.
| {"version":"v2","build":"16","env":"production"} |
-| enable_websocket | False | Auxiliary | Enables a websocket. Set to
`false` by default.
| |
+| enable_websocket | False | Auxiliary | Enables a websocket. Set to
`false` by default. This is a plain protocol upgrade with no access to
individual frames; see the note under Upstream [`scheme`](#upstream) if a
plugin needs to inspect or rewrite them. |
|
| hosts | False | Match Rules | Matches with any one of the
multiple `host`s specified in the form of a non-empty list.
| ["foo.com", "*.bar.com"] |
Example configuration:
@@ -1014,7 +1014,7 @@ In addition to the equalization algorithm selections,
Upstream also supports pas
| desc | False
| Auxiliary | Description of usage
scenarios.
[...]
| pass_host | False
| Enumeration | Configures the `host` when
the request is forwarded to the upstream. Can be one of `pass`, `node` or
`rewrite`. Defaults to `pass` if not specified. `pass`- transparently passes
the client's host to the Upstream. `node`- uses the host configured in the node
of the Upstream. `rewrite`- Uses the value configured in `upstream_host`.
[...]
| upstream_host | False
| Auxiliary | Specifies the host of the
Upstream request. This is only valid if the `pass_host` is set to `rewrite`.
[...]
-| scheme | False
| Auxiliary | The scheme used when
communicating with the Upstream. For an L7 proxy, this value can be one of
`http`, `https`, `grpc`, `grpcs`. For an L4 proxy, this value could be one of
`tcp`, `udp`, `tls`. Defaults to `http`.
[...]
+| scheme | False
| Auxiliary | The scheme used when
communicating with the Upstream. For an L7 proxy, this value can be one of
`http`, `https`, `grpc`, `grpcs`, `ws`, `wss`. For an L4 proxy, this value
could be one of `tcp`, `udp`, `tls`. Defaults to `http`.
[...]
| labels | False
| Match Rules | Attributes of the Upstream
specified as `key-value` pairs.
[...]
| tls.client_cert | False, can't be used with `tls.client_cert_id`
| HTTPS certificate | Sets the client certificate
while connecting to a TLS Upstream.
[...]
| 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.
[...]
@@ -1046,6 +1046,11 @@ The following should be considered when setting the
`hash_on` value:
- When set to `consumer`, the `key` is optional and the key is set to the
`consumer_name` captured from the authentication Plugin.
- When set to `vars_combinations`, the `key` is required. The value of the key
can be a combination of any of the [Nginx
variables](http://nginx.org/en/docs/varindex.html) like
`$request_uri$remote_addr`.
+APISIX supports proxying WebSocket connections in two different ways, and they
don't combine:
+
+- Route or Service level [`enable_websocket`](#route) with an `http`/`https`
Upstream `scheme`. This is a plain protocol upgrade: nginx's own `proxy_pass`
forwards the raw TCP stream after the `101 Switching Protocols` handshake, and
no plugin phase sees the individual WebSocket frames.
+- Upstream `scheme: ws` or `scheme: wss`. APISIX parses and proxies the
WebSocket frames itself in both directions, which lets a plugin inspect or
rewrite frames in flight through the `ws_handshake`, `ws_client_frame`,
`ws_upstream_frame`, and `ws_close` phases. See the ["extra phase" section of
the plugin development guide](./plugin-develop.md#extra-phase) for how to hook
into them. `enable_websocket` is ignored on a Route or a Service whose Upstream
uses this scheme, since the connecti [...]
+
The features described below requires APISIX to be run on
[APISIX-Runtime](./FAQ.md#how-do-i-build-the-apisix-runtime-environment):
You can set the `scheme` to `tls`, which means "TLS over TCP".
diff --git a/docs/en/latest/plugin-develop.md b/docs/en/latest/plugin-develop.md
index cdf837cc75..e384ec18bd 100644
--- a/docs/en/latest/plugin-develop.md
+++ b/docs/en/latest/plugin-develop.md
@@ -217,6 +217,33 @@ function _M.delayed_body_filter(conf, ctx)
end
```
+When a route's `upstream.scheme` is `ws` or `wss`, APISIX proxies WebSocket
frames itself instead of letting nginx's `proxy_pass` transparently forward
them, so it can also run a plugin's logic against each frame. The normal
`rewrite`/`access`/`before_proxy` phases still run beforehand and `log` still
runs afterward; only `header_filter`/`body_filter`/`delayed_body_filter` are
skipped, since there's no separate response to filter. In their place, four
WebSocket-specific phases fire for s [...]
+
+* `ws_handshake` - runs once, after `before_proxy`, before APISIX attempts to
connect to the upstream.
+* `ws_client_frame` - runs once per frame received from the downstream client,
before it is forwarded to the upstream.
+* `ws_upstream_frame` - runs once per frame received from the upstream, before
it is forwarded to the downstream client.
+* `ws_close` - runs once, when the connection ends, before the normal `log`
phase.
+
+`ws_client_frame` and `ws_upstream_frame` can read and rewrite the frame in
flight through `core.websocket.client` and `core.websocket.upstream`
respectively (`core.websocket.get_role("client")` and
`core.websocket.get_role("upstream")` return the same two tables).
`get_frame()` returns the current frame (`type`, `payload`, `last`, `code`);
`set_frame_data(payload)` replaces the payload that actually gets forwarded:
+
+```lua
+function _M.ws_client_frame(conf, ctx)
+ local frame = core.websocket.client.get_frame()
+ if frame.type == "text" then
+ core.websocket.client.set_frame_data(frame.payload .. "-client")
+ end
+end
+
+function _M.ws_upstream_frame(conf, ctx)
+ local frame = core.websocket.upstream.get_frame()
+ if frame.type == "text" then
+ core.websocket.upstream.set_frame_data(frame.payload .. "-upstream")
+ end
+end
+```
+
+See
[`example-plugin`](https://github.com/apache/apisix/blob/master/apisix/plugins/example-plugin.lua)
for a complete reference implementation of all four phases.
+
### Implement the logic
Write the logic of the plugin in the corresponding phase. There are two
parameters `conf` and `ctx` in the phase method, take the `limit-conn` plugin
configuration as an example.
diff --git a/docs/en/latest/terminology/plugin.md
b/docs/en/latest/terminology/plugin.md
index 0aee6fc13b..3fe7cd6038 100644
--- a/docs/en/latest/terminology/plugin.md
+++ b/docs/en/latest/terminology/plugin.md
@@ -87,6 +87,8 @@ An installed plugin is first initialized. The configuration
of the plugin is the
When a request goes through APISIX, the plugin's corresponding methods are
executed in one or more of the following phases : `rewrite`, `access`,
`before_proxy`, `header_filter`, `body_filter`, and `log`. These phases are
largely influenced by the [OpenResty
directives](https://openresty-reference.readthedocs.io/en/latest/Directives/).
+A route whose `upstream.scheme` is `ws` or `wss` still runs
`rewrite`/`access`/`before_proxy`/`log` normally, but replaces
`header_filter`/`body_filter`/`delayed_body_filter` with four
WebSocket-specific phases: `ws_handshake`, `ws_client_frame`,
`ws_upstream_frame`, and `ws_close`. See the ["extra phase" section of the
plugin development guide](../plugin-develop.md#extra-phase) for details.
+
<br />
<div style={{textAlign: 'center'}}>
<img
src="https://static.apiseven.com/uploads/2023/03/09/ZsH5C8Og_plugins-phases.png";
alt="Routes Diagram" width="50%"/>
diff --git a/docs/zh/latest/admin-api.md b/docs/zh/latest/admin-api.md
index ec6aa9c8a7..b08318c76b 100644
--- a/docs/zh/latest/admin-api.md
+++ b/docs/zh/latest/admin-api.md
@@ -370,7 +370,7 @@ Route 也称之为路由，可以通过定义一些规则来匹配客户端的
| filter_func | 否 | 匹配规则 |
用户自定义的过滤函数。可以使用它来实现特殊场景的匹配要求实现。该函数默认接受一个名为 `vars` 的输入参数，可以用它来获取 NGINX 变量。
| function(vars) return
vars["arg_name"] == "json" end |
| labels | 否 | 匹配规则 | 标识附加属性的键值对。
| {"version":"v2","build":"16","env":"production"} |
| timeout | 否 | 辅助 | 为 Route 设置
Upstream 连接、发送消息和接收消息的超时时间（单位为秒）。该配置将会覆盖在 Upstream 中配置的 [timeout](#upstream)
选项。
| {"connect": 3, "send": 3, "read": 3}
|
-| enable_websocket | 否 | 辅助 | 当设置为 `true`
时，启用 `websocket`(boolean), 默认值为 `false`。
|
|
+| enable_websocket | 否 | 辅助 | 当设置为 `true`
时，启用 `websocket`(boolean), 默认值为
`false`。这只是纯粹的协议升级，插件无法访问单独的帧；如果插件需要读取或改写帧内容，请参考 Upstream [`scheme`](#upstream)
下的说明。
| |
| status | 否 | 辅助 | 当设置为 `1`
时，启用该路由，默认值为 `1`。
| `1` 表示启用，`0` 表示禁用。
|
:::note 注意
@@ -679,7 +679,7 @@ Service 是某类 API 的抽象（也可以理解为一组 Route 的抽象）。
| name | 否 | 辅助 | 服务名称。
|
|
| desc | 否 | 辅助 | 服务描述。
|
|
| labels | 否 | 匹配规则 | 标识附加属性的键值对。
|
{"version":"v2","build":"16","env":"production"} |
-| enable_websocket | 否 | 辅助 | `websocket`(boolean)
配置，默认值为 `false`。 |
|
+| enable_websocket | 否 | 辅助 | `websocket`(boolean)
配置，默认值为 `false`。这只是纯粹的协议升级，插件无法访问单独的帧；如果插件需要读取或改写帧内容，请参考 Upstream
[`scheme`](#upstream) 下的说明。 | |
| hosts | 否 | 匹配规则 | 非空列表形态的 `host`，表示允许有多个不同
`host`，匹配其中任意一个即可。| ["foo.com", "\*.bar.com"] |
Service 对象 JSON 配置示例：
@@ -1022,7 +1022,7 @@ APISIX 的 Upstream 除了基本的负载均衡算法选择外，还支持对上
| desc | 否 | 辅助
| 上游服务描述、使用场景等。
| |
| pass_host | 否 | 枚举
| 请求发给上游时的 `host` 设置选型。 [`pass`，`node`，`rewrite`] 之一，默认是 `pass`。`pass`: 将客户端的
host 透传给上游； `node`: 使用 `upstream` node 中配置的 `host`； `rewrite`: 使用配置项
`upstream_host` 的值。
| |
| upstream_host | 否 | 辅助
| 指定上游请求的 host，只在 `pass_host` 配置为 `rewrite` 时有效。
|
|
-| scheme | 否 | 辅助
| 跟上游通信时使用的 scheme。对于 7 层代理，可选值为 [`http`, `https`, `grpc`, `grpcs`]。对于 4
层代理，可选值为 [`tcp`, `udp`, `tls`]。默认值为 `http`，详细信息请参考下文。
|
+| scheme | 否 | 辅助
| 跟上游通信时使用的 scheme。对于 7 层代理，可选值为 [`http`, `https`, `grpc`, `grpcs`, `ws`,
`wss`]。对于 4 层代理，可选值为 [`tcp`, `udp`, `tls`]。默认值为 `http`，详细信息请参考下文。
|
| labels | 否 | 匹配规则
| 标识附加属性的键值对。
| {"version":"v2","build":"16","env":"production"} |
| tls.client_cert | 否，不能和 `tls.client_cert_id` 一起使用 | https
证书 | 设置跟上游通信时的客户端证书，详细信息请参考下文。
| |
| tls.client_key | 否，不能和 `tls.client_cert_id` 一起使用 |
https 证书私钥 | 设置跟上游通信时的客户端私钥，详细信息请参考下文。
| |
@@ -1053,6 +1053,11 @@ APISIX 的 Upstream 除了基本的负载均衡算法选择外，还支持对上
- 设为 `cookie` 时，`key` 为必传参数，其值为自定义的 cookie name，即 "cookie\_`key`"。请注意 cookie
name 是**区分大小写字母**的。例如：`cookie_x_foo` 与 `cookie_X_Foo` 表示不同的 `cookie`。
- 设为 `consumer` 时，`key` 不需要设置。此时哈希算法采用的 `key` 为认证通过的 `consumer_name`。
+APISIX 支持两种不同的方式来代理 WebSocket 连接，二者不能混用：
+
+- Route 或 Service 级别的 [`enable_websocket`](#route)，配合 `http`/`https` 的
Upstream `scheme`。这是纯粹的协议升级：`101 Switching Protocols` 握手完成后，由 nginx 自身的
`proxy_pass` 转发原始 TCP 流，没有任何插件 phase 能看到单独的 WebSocket 帧。
+- Upstream `scheme: ws` 或 `scheme: wss`。APISIX 会自己双向解析并代理 WebSocket 帧，插件可以通过
`ws_handshake`、`ws_client_frame`、`ws_upstream_frame`、`ws_close` 这几个 phase
在帧的转发过程中读取或改写它们，具体用法参考[插件开发指南的 "extra phase"
一节](./plugin-develop.md#extra-phase)。如果 Route 或 Service 所属的 Upstream 使用了这个
scheme，`enable_websocket` 会被忽略，因为连接根本不会走到它所配置的那条 `proxy_pass` 路径。
+
以下特性需要 APISIX 运行于 [APISIX-Runtime](./FAQ.md#如何构建-APISIX-Runtime-环境？)：
- `scheme` 可以设置成 `tls`，表示 `TLS over TCP`。
diff --git a/t/APISIX.pm b/t/APISIX.pm
index db1727424a..1ae608eeb9 100644
--- a/t/APISIX.pm
+++ b/t/APISIX.pm
@@ -259,6 +259,18 @@ my $disable_proxy_buffering_location = <<_EOC_;
}
_EOC_
+my $websocket_location = <<_EOC_;
+ location \@websocket_pass {
+ content_by_lua_block {
+ apisix.websocket_content_phase()
+ }
+
+ log_by_lua_block {
+ apisix.websocket_log_phase()
+ }
+ }
+_EOC_
+
my $a6_ngx_directives = "";
if ($version =~ m/\/apisix-nginx-module/) {
$a6_ngx_directives = <<_EOC_;
@@ -777,6 +789,20 @@ _EOC_
}
}
+ # accepts a connection on any path but never writes a response, so a
+ # client waiting on it reliably times out instead of being refused or
+ # having to depend on an unroutable address actually hanging
+ server {
+ listen 1986;
+ server_tokens off;
+
+ location / {
+ content_by_lua_block {
+ ngx.sleep(30)
+ }
+ }
+ }
+
$a6_ngx_directives
server {
@@ -1010,6 +1036,7 @@ _EOC_
$grpc_location
$dubbo_location
$disable_proxy_buffering_location
+ $websocket_location
location = /proxy_mirror {
internal;
diff --git a/t/lib/server.lua b/t/lib/server.lua
index c21975ff66..757c3c584a 100644
--- a/t/lib/server.lua
+++ b/t/lib/server.lua
@@ -389,6 +389,225 @@ end
_M.websocket_handshake_route = _M.websocket_handshake
+-- Echoes every text/binary frame it receives back to the sender unchanged,
+-- so a fronting proxy's frame-level plugin hooks can be observed by diffing
+-- what the client sent against what it gets back. Used by the
+-- websocket-enhanced (ws/wss upstream scheme) test suite.
+function _M.websocket_echo()
+ local websocket = require "resty.websocket.server"
+ local wb, err = websocket:new()
+ if not wb then
+ ngx.log(ngx.ERR, "failed to new websocket: ", err)
+ return ngx.exit(400)
+ end
+
+ while true do
+ local data, typ, err = wb:recv_frame()
+ if not data then
+ if err and err:find("timeout", 1, true) then
+ goto continue
+ end
+ ngx.log(ngx.ERR, "failed to receive frame: ", err)
+ return
+ end
+
+ if typ == "close" then
+ wb:send_close(1000, "")
+ return
+ elseif typ == "ping" then
+ wb:send_pong(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)
+ if not bytes then
+ ngx.log(ngx.ERR, "failed to echo frame: ", send_err)
+ return
+ end
+ end
+
+ ::continue::
+ end
+end
+
+
+-- Like websocket_echo, but the first thing it sends back is a text frame
+-- carrying the request URI (with query string) it was actually dispatched
+-- with, so a test can confirm what path/query a fronting proxy forwarded.
+-- Falls into the same echo loop afterwards.
+function _M.websocket_echo_uri()
+ local websocket = require "resty.websocket.server"
+ local wb, err = websocket:new()
+ if not wb then
+ ngx.log(ngx.ERR, "failed to new websocket: ", err)
+ return ngx.exit(400)
+ end
+
+ local bytes, send_err = wb:send_text(ngx.var.request_uri)
+ if not bytes then
+ ngx.log(ngx.ERR, "failed to send request_uri: ", send_err)
+ return
+ end
+
+ while true do
+ local data, typ, recv_err = wb:recv_frame()
+ if not data then
+ if recv_err and recv_err:find("timeout", 1, true) then
+ goto continue
+ end
+ ngx.log(ngx.ERR, "failed to receive frame: ", recv_err)
+ return
+ end
+
+ if typ == "close" then
+ wb:send_close(1000, "")
+ return
+ elseif typ == "ping" then
+ wb:send_pong(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)
+ if not ok then
+ ngx.log(ngx.ERR, "failed to echo frame: ", echo_err)
+ return
+ end
+ end
+
+ ::continue::
+ end
+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
+-- echo loop afterwards.
+function _M.websocket_echo_headers()
+ local websocket = require "resty.websocket.server"
+ local wb, err = websocket:new()
+ if not wb then
+ ngx.log(ngx.ERR, "failed to new websocket: ", err)
+ return ngx.exit(400)
+ end
+
+ local headers = ngx.req.get_headers()
+ local bytes, send_err = wb:send_text(json_encode({
+ x_real_ip = headers["X-Real-IP"],
+ x_forwarded_for = headers["X-Forwarded-For"],
+ }))
+ if not bytes then
+ ngx.log(ngx.ERR, "failed to send headers: ", send_err)
+ return
+ end
+
+ while true do
+ local data, typ, recv_err = wb:recv_frame()
+ if not data then
+ if recv_err and recv_err:find("timeout", 1, true) then
+ goto continue
+ end
+ ngx.log(ngx.ERR, "failed to receive frame: ", recv_err)
+ return
+ end
+
+ if typ == "close" then
+ wb:send_close(1000, "")
+ return
+ elseif typ == "ping" then
+ wb:send_pong(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)
+ if not ok then
+ ngx.log(ngx.ERR, "failed to echo frame: ", echo_err)
+ return
+ end
+ end
+
+ ::continue::
+ end
+end
+
+
+-- Sends one fragmented text message ("hello " + "world" as two continuation
+-- frames) right after the handshake, to verify a fronting proxy's
+-- aggregate_fragments option reassembles it into a single frame instead of
+-- forwarding (or invoking frame hooks on) two separate pieces.
+function _M.websocket_fragment()
+ local websocket = require "resty.websocket.server"
+ local wb, err = websocket:new()
+ if not wb then
+ ngx.log(ngx.ERR, "failed to new websocket: ", err)
+ return ngx.exit(400)
+ end
+
+ local ok, send_err = wb:send_frame(false, 0x1, "hello ")
+ if not ok then
+ ngx.log(ngx.ERR, "failed to send first fragment: ", send_err)
+ return
+ end
+
+ ok, send_err = wb:send_frame(true, 0x0, "world")
+ if not ok then
+ ngx.log(ngx.ERR, "failed to send final fragment: ", send_err)
+ return
+ end
+
+ -- drain until the client closes, so the connection doesn't just vanish
+ -- out from under the proxy mid-test
+ while true do
+ local data, typ, recv_err = wb:recv_frame()
+ if not data then
+ if recv_err and recv_err:find("timeout", 1, true) then
+ goto continue
+ end
+ return
+ end
+ if typ == "close" then
+ wb:send_close(1000, "")
+ return
+ end
+ ::continue::
+ end
+end
+
+
+-- Sends a close frame of its own right after the handshake, without waiting
+-- for the client to initiate one, to exercise an upstream-initiated close.
+function _M.websocket_close_upstream_initiated()
+ local websocket = require "resty.websocket.server"
+ local wb, err = websocket:new()
+ if not wb then
+ ngx.log(ngx.ERR, "failed to new websocket: ", err)
+ return ngx.exit(400)
+ end
+
+ wb:send_close(1000, "bye")
+end
+
+
+-- Completes the handshake, echoes exactly one frame, then vanishes without
+-- sending a close frame, to simulate an upstream that dies mid-session
+-- instead of closing cleanly.
+function _M.websocket_abrupt_close()
+ local websocket = require "resty.websocket.server"
+ local wb, err = websocket:new()
+ if not wb then
+ ngx.log(ngx.ERR, "failed to new websocket: ", err)
+ return ngx.exit(400)
+ end
+
+ local data, typ = wb:recv_frame()
+ if data and (typ == "text" or typ == "binary") then
+ local send = typ == "text" and wb.send_text or wb.send_binary
+ send(wb, data)
+ end
+
+ -- returning here, with the connection already hijacked by
+ -- resty.websocket.server, drops the raw TCP connection without a
+ -- close handshake
+end
+
+
-- keep the session open until the peer goes away, so that the request stays in
-- flight in the balancer the way a real WebSocket session does. An idle
timeout is
-- the normal state of such a session, not an error: keep waiting, and only
give up
diff --git a/t/node/websocket-proxy.spec.mts b/t/node/websocket-proxy.spec.mts
new file mode 100644
index 0000000000..7b85d68699
--- /dev/null
+++ b/t/node/websocket-proxy.spec.mts
@@ -0,0 +1,479 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+import { describe, expect, it, jest } from '@jest/globals';
+import axios from 'axios';
+import WS from 'ws';
+
+import { request as requestAdminAPI } from '../ts/admin_api';
+import { wait } from '../ts/utils';
+
+// Every test here does at least one real websocket handshake plus etcd sync
+// round trip, which the shared 5s Jest default leaves little room for on a
+// loaded machine; the handful of tests that need more than this still set
+// their own per-test timeout on top of it.
+jest.setTimeout(15000);
+
+const PROXY_BASE = 'ws://localhost:1984';
+// a loopback address nothing listens on, used as an unreachable upstream node
+const DEAD_NODE = '127.0.0.1:1';
+const DEAD_NODE_2 = '127.0.0.1:2';
+// accepts the connection but never responds, so it reliably exercises the
+// "timeout" (504) branch instead of "tcp failure" - unlike an unroutable
+// address, this doesn't depend on how a given network treats one
+const BLACKHOLE_NODE = '127.0.0.1:1986';
+const ECHO_NODE = '127.0.0.1:1980';
+
+let nextRouteId = 1;
+
+// Routes with a URI no other test in this file reuses can just be created
+// once and left behind (the whole test-nginx instance goes away at the end
+// of the run anyway).
+const createRoute = async (
+ path: string,
+ upstream: object,
+ plugins?: object,
+) => {
+ const id = `ws-proxy-${nextRouteId++}`;
+ const res = await requestAdminAPI(`/apisix/admin/routes/${id}`, 'PUT', {
+ uri: path,
+ upstream,
+ plugins,
+ });
+ expect(res.status).toBe(res.status < 300 ? res.status : 200);
+ // give etcd -> apisix config sync a moment to land before the first request
+ await wait(300);
+ return id;
+};
+
+// Most tests below share the /websocket_echo URI across very different
+// upstream/plugin configs. Deleting and recreating a route under the same
+// URI between tests leaves a window where the delete has been sent but
+// hasn't synced yet when the next test's create lands, and the two can race
+// - the client then gets whichever half-applied state the router held at
+// 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',
+ upstream,
+ plugins,
+ });
+ expect(res.status).toBe(res.status < 300 ? res.status : 200);
+ await wait(300);
+};
+
+// Opens a websocket connection, sends one text frame, resolves with the
+// first frame received in reply (or rejects on error/close-before-reply).
+const sendAndReceive = (path: string, payload: string) =>
+ new Promise<string>((resolve, reject) => {
+ const ws = new WebSocket(`${PROXY_BASE}${path}`);
+ ws.addEventListener('open', () => ws.send(payload));
+ ws.addEventListener('message', (ev) => {
+ resolve(ev.data as string);
+ ws.close();
+ });
+ ws.addEventListener('error', (ev) =>
+ reject(new Error((ev as unknown as { message?: string }).message ??
'websocket error')),
+ );
+ });
+
+// Resolves with the close event's code. onOpen fires right after connecting,
+// so it can send a frame or otherwise trigger whatever leads to the close.
+//
+// The WebSocket spec requires an abnormal closure to fire an error event
+// before its close event, not instead of it, so an expected-to-fail
+// connection (tolerateError: true) must not treat that error as a failure
+// and must instead keep waiting for the close event that follows it.
+const waitForClose = (
+ path: string,
+ onOpen?: (ws: WebSocket) => void,
+ tolerateError = false,
+) =>
+ new Promise<number>((resolve, reject) => {
+ const ws = new WebSocket(`${PROXY_BASE}${path}`);
+ ws.addEventListener('open', () => onOpen?.(ws));
+ ws.addEventListener('close', (ev) => resolve(ev.code));
+ ws.addEventListener('error', (ev) => {
+ if (!tolerateError) {
+ reject(new Error((ev as unknown as { message?: string }).message ??
'websocket error'));
+ }
+ });
+ });
+
+describe('websocket-proxy (ws/wss upstream scheme)', () => {
+ describe('frame-level plugin hooks', () => {
+ it('lets a plugin rewrite the client frame before it reaches the upstream,
and the upstream frame before it reaches the client', async () => {
+ await putEchoRoute(
+ { type: 'roundrobin', scheme: 'ws', nodes: { [ECHO_NODE]: 1 } },
+ {
+ // 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 echo backend bounces whatever it received back unchanged, so the
+ // round trip proves both directions were actually rewritten in flight.
+ const reply = await sendAndReceive('/websocket_echo', 'hello');
+ expect(reply).toBe('hello-client-upstream');
+ });
+
+ it('does not touch binary frames (the hooks only rewrite text frames)',
async () => {
+ await putEchoRoute(
+ { type: 'roundrobin', scheme: 'ws', nodes: { [ECHO_NODE]: 1 } },
+ { 'example-plugin': { i: 1 } },
+ );
+
+ const reply = await new Promise<ArrayBuffer>((resolve, reject) => {
+ const ws = new WebSocket(`${PROXY_BASE}/websocket_echo`);
+ ws.binaryType = 'arraybuffer';
+ ws.addEventListener('open', () => ws.send(new Uint8Array([1, 2, 3,
4])));
+ ws.addEventListener('message', (ev) => {
+ resolve(ev.data as ArrayBuffer);
+ ws.close();
+ });
+ ws.addEventListener('error', (ev) =>
+ reject(new Error((ev as unknown as { message?: string }).message ??
'websocket error')),
+ );
+ });
+ expect(new Uint8Array(reply)).toEqual(new Uint8Array([1, 2, 3, 4]));
+ });
+ });
+
+ describe('fragmented frames', () => {
+ it('reassembles a fragmented message into one frame before invoking plugin
hooks', async () => {
+ await createRoute('/websocket_fragment', {
+ type: 'roundrobin',
+ scheme: 'ws',
+ nodes: { [ECHO_NODE]: 1 },
+ }, {
+ 'example-plugin': { i: 1 },
+ });
+
+ // websocket_fragment sends "hello " and "world" as two continuation
+ // frames of the same message; if aggregate_fragments works, the client
+ // (and the ws_upstream_frame hook in between) see exactly one frame
+ // with the joined payload, not two separate ones.
+ const messages: string[] = [];
+ await new Promise<void>((resolve, reject) => {
+ const ws = new WebSocket(`${PROXY_BASE}/websocket_fragment`);
+ ws.addEventListener('message', (ev) => {
+ messages.push(ev.data as string);
+ ws.close();
+ });
+ ws.addEventListener('close', () => resolve());
+ ws.addEventListener('error', (ev) =>
+ reject(new Error((ev as unknown as { message?: string }).message ??
'websocket error')),
+ );
+ });
+ expect(messages).toEqual(['hello world-upstream']);
+ });
+ });
+
+ describe('ping/pong', () => {
+ it('forwards a client ping to the upstream and the upstream pong back to
the client', async () => {
+ await putEchoRoute({ type: 'roundrobin', scheme: 'ws', nodes: {
[ECHO_NODE]: 1 } });
+
+ await new Promise<void>((resolve, reject) => {
+ const ws = new WS(`${PROXY_BASE}/websocket_echo`);
+ ws.on('open', () => ws.ping());
+ ws.on('pong', () => {
+ ws.terminate();
+ resolve();
+ });
+ ws.on('error', reject);
+ });
+ });
+ });
+
+ describe('close handshake', () => {
+ it('lets the client close cleanly and the upstream echoes the close code
back', async () => {
+ await putEchoRoute({ type: 'roundrobin', scheme: 'ws', nodes: {
[ECHO_NODE]: 1 } });
+
+ const code = await waitForClose('/websocket_echo', (ws) =>
ws.close(1000, 'bye'));
+ expect(code).toBe(1000);
+ });
+
+ it('forwards an upstream-initiated close to the client', async () => {
+ await createRoute('/websocket_close_upstream_initiated', {
+ type: 'roundrobin',
+ scheme: 'ws',
+ nodes: { [ECHO_NODE]: 1 },
+ });
+
+ const code = await waitForClose('/websocket_close_upstream_initiated');
+ expect(code).toBe(1000);
+ });
+ });
+
+ describe('abrupt disconnects', () => {
+ it('reports an abnormal closure (1006) to the client when the upstream
vanishes mid-session', async () => {
+ await createRoute('/websocket_abrupt_close', {
+ type: 'roundrobin',
+ scheme: 'ws',
+ nodes: { [ECHO_NODE]: 1 },
+ });
+
+ const code = await waitForClose('/websocket_abrupt_close', (ws) =>
ws.send('hi'), true);
+ expect(code).toBe(1006);
+ });
+
+ it('cleans up the upstream side when the client vanishes without closing',
async () => {
+ await putEchoRoute({ type: 'roundrobin', scheme: 'ws', nodes: {
[ECHO_NODE]: 1 } });
+
+ const activeConnections = async () => {
+ const res = await
axios.get('http://localhost:1984/apisix/nginx_status');
+ const match = /Active connections:\s*(\d+)/.exec(res.data as string);
+ return match ? Number(match[1]) : NaN;
+ };
+
+ const baseline = await activeConnections();
+
+ // open and forcibly kill a handful of connections without a close
+ // handshake (ws's .terminate() drops the TCP connection directly,
+ // which the standard WebSocket API has no equivalent for)
+ for (let i = 0; i < 10; i++) {
+ await new Promise<void>((resolve, reject) => {
+ const ws = new WS(`${PROXY_BASE}/websocket_echo`);
+ ws.on('open', () => {
+ ws.terminate();
+ resolve();
+ });
+ ws.on('error', reject);
+ });
+ }
+
+ // give the proxy's forwarder coroutines a moment to notice the dead
+ // sockets and tear themselves down
+ await wait(1000);
+
+ const after = await activeConnections();
+ // a leak would grow roughly linearly with the number of terminated
+ // connections (10 here); allow some slack for unrelated background
+ // activity in the shared test-nginx instance instead of an exact match
+ expect(after).toBeLessThan(baseline + 5);
+ }, 10000);
+ });
+
+ describe('upstream retry', () => {
+ it('retries the next node when the first one refuses the connection',
async () => {
+ await putEchoRoute({
+ type: 'roundrobin',
+ scheme: 'ws',
+ retries: 3,
+ nodes: { [DEAD_NODE]: 100, [ECHO_NODE]: 1 },
+ });
+
+ const reply = await sendAndReceive('/websocket_echo', 'hello');
+ expect(reply).toBe('hello');
+ });
+
+ it('returns 502 once every node has been tried and failed', async () => {
+ await putEchoRoute({
+ type: 'roundrobin',
+ scheme: 'ws',
+ retries: 2,
+ nodes: { [DEAD_NODE]: 1, [DEAD_NODE_2]: 1 },
+ });
+
+ await expect(
+ axios.get('http://localhost:1984/websocket_echo', {
+ headers: {
+ Connection: 'Upgrade',
+ Upgrade: 'websocket',
+ 'Sec-WebSocket-Key': 'dGhlIHNhbXBsZSBub25jZQ==',
+ 'Sec-WebSocket-Version': '13',
+ },
+ }),
+ ).rejects.toMatchObject({
+ response: { status: 502 },
+ });
+ });
+
+ it('retries past a connect timeout, not just a refused connection', async
() => {
+ await putEchoRoute({
+ type: 'roundrobin',
+ scheme: 'ws',
+ retries: 2,
+ timeout: { connect: 1, send: 5, read: 5 },
+ nodes: { [BLACKHOLE_NODE]: 100, [ECHO_NODE]: 1 },
+ });
+
+ const start = Date.now();
+ const reply = await sendAndReceive('/websocket_echo', 'hello');
+ const elapsed = Date.now() - start;
+ expect(reply).toBe('hello');
+ // an instant refusal (tcp_failure) would resolve in a few ms; only a
+ // real connect timeout takes close to the configured 1s
+ expect(elapsed).toBeGreaterThanOrEqual(900);
+ }, 10000);
+
+ it('retries across more than one dead node before reaching a healthy one',
async () => {
+ await putEchoRoute({
+ type: 'roundrobin',
+ scheme: 'ws',
+ retries: 3,
+ nodes: { [DEAD_NODE]: 100, [DEAD_NODE_2]: 100, [ECHO_NODE]: 1 },
+ });
+
+ const reply = await sendAndReceive('/websocket_echo', 'hello');
+ expect(reply).toBe('hello');
+ });
+
+ it('retries the same way for a least_conn upstream, not just roundrobin',
async () => {
+ await putEchoRoute({
+ type: 'least_conn',
+ scheme: 'ws',
+ retries: 3,
+ nodes: { [DEAD_NODE]: 100, [ECHO_NODE]: 1 },
+ });
+
+ const reply = await sendAndReceive('/websocket_echo', 'hello');
+ expect(reply).toBe('hello');
+ });
+ });
+
+ describe('passive health check', () => {
+ it('marks a node unhealthy after enough failed connection attempts', async
() => {
+ await putEchoRoute({
+ type: 'roundrobin',
+ scheme: 'ws',
+ retries: 1,
+ nodes: { [DEAD_NODE]: 1, [ECHO_NODE]: 1 },
+ checks: {
+ active: { type: 'tcp', http_path: '/', timeout: 1, healthy: {
interval: 1 } },
+ passive: { unhealthy: { tcp_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++) {
+ 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');
+ }
+
+ expect(unhealthyFound).toBe(true);
+ }, 15000);
+ });
+
+ describe('upstream URI forwarding', () => {
+ it("forwards the client's request URI, including the query string", async
() => {
+ await createRoute('/websocket_echo_uri', {
+ type: 'roundrobin',
+ scheme: 'ws',
+ nodes: { [ECHO_NODE]: 1 },
+ });
+
+ const reply = await new Promise<string>((resolve, reject) => {
+ const ws = new WebSocket(`${PROXY_BASE}/websocket_echo_uri?foo=bar`);
+ ws.addEventListener('message', (ev) => {
+ resolve(ev.data as string);
+ ws.close();
+ });
+ ws.addEventListener('error', (ev) =>
+ reject(new Error((ev as unknown as { message?: string }).message ??
'websocket error')),
+ );
+ });
+ expect(reply).toBe('/websocket_echo_uri?foo=bar');
+ });
+
+ it("forwards the proxy-rewrite plugin's rewritten URI instead of the
original one", async () => {
+ await createRoute(
+ '/websocket_proxy_rewrite_uri',
+ {
+ type: 'roundrobin',
+ scheme: 'ws',
+ nodes: { [ECHO_NODE]: 1 },
+ },
+ {
+ 'proxy-rewrite': { uri: '/websocket_echo_uri' },
+ },
+ );
+
+ const reply = await new Promise<string>((resolve, reject) => {
+ const ws = new WebSocket(`${PROXY_BASE}/websocket_proxy_rewrite_uri`);
+ ws.addEventListener('message', (ev) => {
+ resolve(ev.data as string);
+ ws.close();
+ });
+ ws.addEventListener('error', (ev) =>
+ reject(new Error((ev as unknown as { message?: string }).message ??
'websocket error')),
+ );
+ });
+ expect(reply).toBe('/websocket_echo_uri');
+ });
+ });
+
+ 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', {
+ type: 'roundrobin',
+ scheme: 'ws',
+ nodes: { [ECHO_NODE]: 1 },
+ });
+
+ // connect via the literal loopback address, not PROXY_BASE's
+ // "localhost", so $remote_addr is deterministically 127.0.0.1
+ const reply = await new Promise<string>((resolve, reject) => {
+ const ws = new WS('ws://127.0.0.1:1984/websocket_echo_headers', {
+ headers: { 'X-Real-IP': '1.2.3.4', 'X-Forwarded-For': '5.6.7.8' },
+ });
+ ws.on('message', (data) => {
+ resolve(data.toString());
+ ws.close();
+ });
+ ws.on('error', reject);
+ });
+
+ const seen = JSON.parse(reply);
+ expect(seen.x_real_ip).toBe('127.0.0.1');
+ expect(seen.x_forwarded_for).toBe('5.6.7.8, 127.0.0.1');
+ });
+ });
+
+ describe('concurrent connections', () => {
+ it("keeps two simultaneous connections' frame data isolated from each
other", async () => {
+ await putEchoRoute(
+ { type: 'roundrobin', scheme: 'ws', nodes: { [ECHO_NODE]: 1 } },
+ { 'example-plugin': { i: 1 } },
+ );
+
+ const open = (payload: string) =>
+ new Promise<string>((resolve, reject) => {
+ const ws = new WebSocket(`${PROXY_BASE}/websocket_echo`);
+ ws.addEventListener('open', () => ws.send(payload));
+ ws.addEventListener('message', (ev) => {
+ resolve(ev.data as string);
+ ws.close();
+ });
+ ws.addEventListener('error', (ev) =>
+ reject(new Error((ev as unknown as { message?: string }).message
?? 'websocket error')),
+ );
+ });
+
+ const [replyA, replyB] = await Promise.all([open('alpha'),
open('beta')]);
+ expect(replyA).toBe('alpha-client-upstream');
+ expect(replyB).toBe('beta-client-upstream');
+ });
+ });
+});
diff --git a/t/node/websocket-proxy.t b/t/node/websocket-proxy.t
new file mode 100644
index 0000000000..73b8fe9251
--- /dev/null
+++ b/t/node/websocket-proxy.t
@@ -0,0 +1,35 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+#
+use t::APISIX 'no_plan';
+
+repeat_each(1);
+no_long_string();
+no_root_location();
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: test
+--- timeout: 60
+--- max_size: 2048000
+--- exec
+cd t && pnpm test node/websocket-proxy.spec.mts 2>&1
+--- no_error_log
+failed to execute the script with status
+--- response_body eval
+qr/PASS node\/websocket-proxy.spec.mts/
diff --git a/t/package.json b/t/package.json
index 7661c6eaba..948dd9cba0 100644
--- a/t/package.json
+++ b/t/package.json
@@ -12,6 +12,7 @@
"@types/google-protobuf": "^3.15.12",
"@types/jest": "29.5.14",
"@types/node": "22.14.1",
+ "@types/ws": "^8.18.1",
"axios": "^1.16.0",
"docker-compose": "^1.2.0",
"google-protobuf": "^3.21.4",
@@ -23,6 +24,7 @@
"simple-git": "^3.27.0",
"ts-jest": "29.3.2",
"ts-node": "10.9.2",
+ "ws": "^8.21.3",
"xhr2": "^0.2.1",
"yaml": "^2.7.1"
},
diff --git a/t/pnpm-lock.yaml b/t/pnpm-lock.yaml
index 46bedf29a1..6e8ca05578 100644
--- a/t/pnpm-lock.yaml
+++ b/t/pnpm-lock.yaml
@@ -32,6 +32,9 @@ importers:
'@types/node':
specifier: 22.14.1
version: 22.14.1
+ '@types/ws':
+ specifier: ^8.18.1
+ version: 8.18.1
axios:
specifier: ^1.16.0
version: 1.16.0
@@ -65,6 +68,9 @@ importers:
ts-node:
specifier: 10.9.2
version: 10.9.2(@types/[email protected])([email protected])
+ ws:
+ specifier: ^8.21.3
+ version: 8.21.3
xhr2:
specifier: ^0.2.1
version: 0.2.1
@@ -509,6 +515,9 @@ packages:
'@types/[email protected]':
resolution: {integrity:
sha512-9aEbYZ3TbYMznPdcdr3SmIrLXwC/AKZXQeCf9Pgao5CKb8CyHuEX5jzWPTkvregvhRJHcpRO6BFoGW9ycaOkYw==}
+ '@types/[email protected]':
+ resolution: {integrity:
sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==}
+
'@types/[email protected]':
resolution: {integrity:
sha512-I4q9QU9MQv4oEOz4tAHJtNz1cwuLxn2F3xcc2iV5WdqLPpUnj30aUuxt1mAxYTG+oe8CZMV/+6rU4S4gRDzqtQ==}
@@ -1764,6 +1773,18 @@ packages:
resolution: {integrity:
sha512-7KxauUdBmSdWnmpaGFg+ppNjKF8uNLry8LyzjauQDOVONfFLNKrKvQOxZ/VuTIcS/gge/YNahf5RIIQWTSarlg==}
engines: {node: ^12.13.0 || ^14.15.0 || >=16.0.0}
+ [email protected]:
+ resolution: {integrity:
sha512-201TZ/kPWxoPr/OKWjquZR1SWKXcvxdH+e1xrx89b3YbmzLMFCLfnaG1HFIgWzJOEWZ7MvpK++odZufgYR50Rw==}
+ engines: {node: '>=10.0.0'}
+ peerDependencies:
+ bufferutil: ^4.0.1
+ utf-8-validate: '>=5.0.2'
+ peerDependenciesMeta:
+ bufferutil:
+ optional: true
+ utf-8-validate:
+ optional: true
+
[email protected]:
resolution: {integrity:
sha512-sID0rrVCqkVNUn8t6xuv9+6FViXjUVXq8H5rWOH2rz9fDNQEd4g0EA2XlcEdJXRz5BMEn4O1pJFdT+z4YHhoWw==}
engines: {node: '>= 6'}
@@ -2402,6 +2423,10 @@ snapshots:
'@types/[email protected]': {}
+ '@types/[email protected]':
+ dependencies:
+ '@types/node': 22.14.1
+
'@types/[email protected]': {}
'@types/[email protected]':
@@ -3792,6 +3817,8 @@ snapshots:
imurmurhash: 0.1.4
signal-exit: 3.0.7
+ [email protected]: {}
+
[email protected]: {}
[email protected]: {}
![https://static.apiseven.com/uploads/2023/03/09/ZsH5C8Og_plugins-phases.png"](https://static.apiseven.com/uploads/2023/03/09/ZsH5C8Og_plugins-phases.png"){kind=link}