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