diff --git a/apisix-master-0.rockspec b/apisix-master-0.rockspec index 9a6a9fa9d98e..256ffab1a6fe 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 35f015da1b45..de7d07430c23 100644 --- a/apisix/balancer.lua +++ b/apisix/balancer.lua @@ -243,10 +243,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 @@ -283,18 +303,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 @@ -371,6 +386,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 42833de45835..6661eaf1d8e8 100644 --- a/apisix/cli/ngx_tpl.lua +++ b/apisix/cli/ngx_tpl.lua @@ -1179,6 +1179,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 fceb7d6a0d95..2c2bc2d23c90 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 000000000000..64aebf57549f --- /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 20d0caa1f1d5..7881b0f74a98 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 @@ -304,6 +308,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 @@ -316,12 +337,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 @@ -330,6 +346,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() @@ -674,6 +730,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") @@ -988,6 +1062,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 767ccfae72a9..e76ce72e9e84 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 85dd40905299..b35396941f46 100644 --- a/apisix/schema_def.lua +++ b/apisix/schema_def.lua @@ -506,9 +506,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 99e1e857d37e..760e43cfe52b 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 abd9965d3703..3cfac2ea7f37 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. | {"version":"v2","build":"16","env":"production"} | | 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. | | @@ -1041,6 +1041,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 connection never reaches the `proxy_pass` path it configures. + 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 cdf837cc7523..e384ec18bd19 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 such a route: + +* `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 0aee6fc13bbe..3fe7cd603827 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. +
Routes Diagram diff --git a/docs/zh/latest/admin-api.md b/docs/zh/latest/admin-api.md index aae200b88f85..510f50c136b4 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 证书私钥 | 设置跟上游通信时的客户端私钥，详细信息请参考下文。 | | @@ -1048,6 +1048,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 76dc94530a92..07ce0d5a8fa6 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_; @@ -775,6 +787,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 { @@ -1008,6 +1034,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 c21975ff6691..757c3c584acb 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 000000000000..7b85d68699e2 --- /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((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((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((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((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((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((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((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((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((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((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 000000000000..73b8fe9251c6 --- /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 7661c6eabaca..948dd9cba015 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 46bedf29a19c..6e8ca05578d4 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/node@22.14.1)(typescript@5.8.3) + ws: + specifier: ^8.21.3 + version: 8.21.3 xhr2: specifier: ^0.2.1 version: 0.2.1 @@ -509,6 +515,9 @@ packages: '@types/stack-utils@2.0.3': resolution: {integrity: sha512-9aEbYZ3TbYMznPdcdr3SmIrLXwC/AKZXQeCf9Pgao5CKb8CyHuEX5jzWPTkvregvhRJHcpRO6BFoGW9ycaOkYw==} + '@types/ws@8.18.1': + resolution: {integrity: sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==} + '@types/yargs-parser@21.0.3': 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} + ws@8.21.3: + 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 + xhr2@0.2.1: resolution: {integrity: sha512-sID0rrVCqkVNUn8t6xuv9+6FViXjUVXq8H5rWOH2rz9fDNQEd4g0EA2XlcEdJXRz5BMEn4O1pJFdT+z4YHhoWw==} engines: {node: '>= 6'} @@ -2402,6 +2423,10 @@ snapshots: '@types/stack-utils@2.0.3': {} + '@types/ws@8.18.1': + dependencies: + '@types/node': 22.14.1 + '@types/yargs-parser@21.0.3': {} '@types/yargs@17.0.33': @@ -3792,6 +3817,8 @@ snapshots: imurmurhash: 0.1.4 signal-exit: 3.0.7 + ws@8.21.3: {} + xhr2@0.2.1: {} y18n@5.0.8: {}