sync: cherry-pick feature commits from Kong/lua-resty-websocket

.github/workflows/ci.yml

@@ -62,6 +62,7 @@ jobs:
           git clone https://github.com/openresty/nginx-devel-utils.git
           git clone https://github.com/simpl/ngx_devel_kit.git ../ndk-nginx-module
           git clone https://github.com/openresty/lua-nginx-module.git ../lua-nginx-module
+          git clone https://github.com/openresty/stream-lua-nginx-module.git ../stream-lua-nginx-module
           git clone https://github.com/openresty/lua-resty-core.git ../lua-resty-core
           git clone https://github.com/openresty/lua-resty-lrucache.git ../lua-resty-lrucache
           git clone https://github.com/openresty/no-pool-nginx.git ../no-pool-nginx
@@ -76,7 +77,7 @@ jobs:
           export LD_LIBRARY_PATH=$PWD/mockeagain:$LD_LIBRARY_PATH
           export TEST_NGINX_RESOLVER=8.8.4.4
           export NGX_BUILD_CC=$CC
-          ngx-build $NGINX_VERSION --with-ipv6 --with-http_realip_module --with-http_ssl_module --with-cc-opt="-I$PCRE_INC -I$OPENSSL_INC" --with-ld-opt="-L$PCRE_LIB -L$OPENSSL_LIB -Wl,-rpath,$PCRE_LIB:$OPENSSL_LIB" --add-module=../ndk-nginx-module --add-module=../lua-nginx-module --with-debug > build.log 2>&1 || (cat build.log && exit 1)
+          ngx-build $NGINX_VERSION --with-ipv6 --with-http_realip_module --with-http_ssl_module --with-stream --with-stream_ssl_module --with-stream_ssl_preread_module --with-cc-opt="-I$PCRE_INC -I$OPENSSL_INC" --with-ld-opt="-L$PCRE_LIB -L$OPENSSL_LIB -Wl,-rpath,$PCRE_LIB:$OPENSSL_LIB" --add-module=../ndk-nginx-module --add-module=../lua-nginx-module --add-module=../stream-lua-nginx-module --with-debug > build.log 2>&1 || (cat build.log && exit 1)
           nginx -V
           ldd `which nginx`|grep -E 'luajit|ssl|pcre'
           prove -I. -r t

lib/resty/websocket/client.lua

@@ -15,10 +15,13 @@ local new_tab = wbproto.new_tab
 local tcp = ngx.socket.tcp
 local re_match = ngx.re.match
 local re_find  = ngx.re.find
+local re_gmatch = ngx.re.gmatch
 local encode_base64 = ngx.encode_base64
 local concat = table.concat
+local insert = table.insert
 local char = string.char
 local str_find = string.find
+local str_sub = string.sub
 local rand = math.random
 local rshift = bit.rshift
 local band = bit.band
@@ -27,6 +30,7 @@ local type = type
 local debug = ngx.config.debug
 local ngx_log = ngx.log
 local ngx_DEBUG = ngx.DEBUG
+local tostring = tostring
 local assert = assert
 local ssl_support = true
 
@@ -79,6 +83,13 @@ end
 
 
 function _M.connect(self, uri, opts)
+    -- a client instance may be reused across multiple connect() attempts
+    -- (e.g. a failed connect followed by a retry); clear any response
+    -- metadata from a previous attempt so callers never observe stale data.
+    self.resp_status_code = nil
+    self.resp_header = nil
+    self.resp_headers = nil
+
     local sock = self.sock
     if not sock then
         return nil, "not initialized"
@@ -191,11 +202,59 @@ function _M.connect(self, uri, opts)
         end
     end
 
+    local connect_addr, connect_port = addr, port
+    local connect_is_unix = is_unix
+    local proxy_opts = opts and opts.proxy_opts
+    local proxy_url
+
+    if scheme == "wss" and proxy_opts and proxy_opts.wss_proxy then
+        proxy_url = proxy_opts.wss_proxy
+    end
+
+    if proxy_url then
+        if str_sub(proxy_url, 1, 6) == "unix:/" then
+            connect_addr = proxy_url
+            connect_port = nil
+            connect_is_unix = true
+
+        else
+            -- https://github.com/ledgetech/lua-resty-http/blob/master/lib/resty/http.lua
+            local m, err = re_match(
+                proxy_url,
+                [[^(?:(http[s]?):)?//((?:[^\[\]:/\?]+)|(?:\[.+\]))(?::(\d+))?([^\?]*)\??(.*)]],
+                "jo"
+            )
+            if err then
+                return nil, "error parsing proxy_url: " .. err
+
+            elseif not m then
+                return nil, "invalid proxy url"
+
+            elseif m[1] == "https" then
+                -- TLS to the proxy itself (as opposed to the tunnelled TLS
+                -- handshake with the target once CONNECT succeeds) is not
+                -- implemented; fail loudly instead of silently sending the
+                -- CONNECT request (and any Proxy-Authorization) in the clear.
+                return nil, "https proxy (TLS to the proxy itself) is not implemented"
+
+            elseif m[1] ~= "http" then
+                return nil, "only proxy with scheme \"http\" is supported"
+            end
+
+            connect_addr = m[2]
+            connect_port = m[3] or 80
+        end
+
+        if not connect_addr then
+            return nil, "invalid proxy url"
+        end
+    end
+
     local ok, err
-    if is_unix then
-        ok, err = sock:connect(addr, sock_opts)
+    if connect_is_unix then
+        ok, err = sock:connect(connect_addr, sock_opts)
     else
-        ok, err = sock:connect(addr, port, sock_opts)
+        ok, err = sock:connect(connect_addr, connect_port, sock_opts)
     end
     if not ok then
         return nil, "failed to connect: " .. err
@@ -213,6 +272,42 @@ function _M.connect(self, uri, opts)
     end
 
     if ssl then
+        if proxy_url then
+            local req = "CONNECT " .. addr .. ":" .. port .. " HTTP/1.1"
+            .. "\r\nHost: " .. addr .. ":" .. port
+            .. "\r\nProxy-Connection: Keep-Alive"
+
+            if proxy_opts.wss_proxy_authorization then
+                req = req .. "\r\nProxy-Authorization: " .. proxy_opts.wss_proxy_authorization
+            end
+
+            req = req  .. "\r\n\r\n"
+
+            local bytes, err = sock:send(req)
+            if not bytes then
+                return nil, "failed to send the handshake request: " .. err
+            end
+
+            local header_reader = sock:receiveuntil("\r\n\r\n")
+            -- FIXME: check for too big response headers
+            local header, err, _ = header_reader()
+            if not header then
+                return nil, "failed to receive response header: " .. err
+            end
+
+            -- error("header: " .. header)
+
+            -- FIXME: verify the response headers
+
+            local m, _ = re_match(header, [[^\s*HTTP/1\.[01]\s+(\d+)]], "jo")
+            if not m then
+                return nil, "bad HTTP response status line: " .. header
+            elseif m[1] ~= "200" then
+                return nil, "error establishing a connection to "..
+                            "the proxy server, got status " .. tostring(m[1])
+            end
+        end
+
         if client_cert then
             ok, err = sock:setclientcert(client_cert, client_priv_key)
             if not ok then
@@ -281,11 +376,23 @@ function _M.connect(self, uri, opts)
         return nil, "bad HTTP response status line: " .. header
     end
 
-    -- RFC 6455 section 4.1: a status code other than 101 means the server
-    -- has not accepted the upgrade, so the client must fail the connection
-    if m[1] ~= "101" then
-        return nil, "failed websocket handshake: unexpected response status: "
-                    .. m[1], header
+    self.resp_status_code = m[1]
+    self.resp_header = header
+    if self.resp_status_code ~= "101" then
+        -- RFC 6455 §4.1: a non-101 response means the WebSocket connection
+        -- was never established; mark fatal unconditionally so that no WS
+        -- frames can be sent on what is still a plain HTTP connection.
+        -- When keep_response=true the caller intends to read the HTTP
+        -- response body, so leave the raw socket open for that purpose.
+        if not (opts and opts.keep_response) then
+            local closing_ok, closing_err = sock:close()
+            if not closing_ok then
+                ngx_log(ngx_DEBUG, "failed to close the underlying socket: ",
+                    closing_err, " when handling a non-101 response")
+            end
+        end
+        self.fatal = true
+        return nil, "unexpected HTTP response code: " .. m[1], header
     end
 
     return 1, nil, header
@@ -423,4 +530,57 @@ function _M.set_keepalive(self, ...)
 end
 
 
+function _M.get_resp_headers(self)
+    if self.resp_headers then
+        return self.resp_headers
+    end
+
+    if not self.resp_header then
+        return nil, "response header not available"
+    end
+
+    local iter, err = re_gmatch(self.resp_header .. "\r\n", "([^:\\s]+):\\s*(.*?)\r\n", "jo")
+    if err then
+        return nil, "failed to parse response header: " .. err
+    end
+
+    -- gather all response headers
+
+    local resp_headers = {}
+
+    while true do
+        local m, err = iter()
+        if err then
+            return nil, "failed to parse response header: " .. err
+        end
+
+        if not m then
+            -- no match found (any more)
+            break
+        end
+
+        local key = m[1]:lower():gsub("-", "_")
+        local val = m[2]
+
+        if resp_headers[key] then
+            if type(resp_headers[key]) ~= "table" then
+                resp_headers[key] = { resp_headers[key] }
+            end
+
+            insert(resp_headers[key], tostring(val))
+
+        else
+            resp_headers[key] = tostring(val)
+        end
+    end
+
+    self.resp_headers = resp_headers
+
+    return resp_headers
+end
+
+function _M.get_resp_status_code(self)
+    return self.resp_status_code
+end
+
 return _M

t/count.t

@@ -63,7 +63,7 @@ size: 5
 --- request
 GET /t
 --- response_body
-size: 13
+size: 15
 --- no_error_log
 [error]