From 4cdec3e0c16244347ea0269d500230ed22ab6826 Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Mon, 14 Nov 2022 22:39:31 -0800
Subject: [PATCH 01/25] wip

---
 kong-3.1.0-0.rockspec            |   1 +
 kong.conf.default                |   8 +
 kong/clustering/utils.lua        |  43 +++-
 kong/conf_loader/init.lua        |  35 +++
 kong/resty/websocket/client.lua  | 371 +++++++++++++++++++++++++++++++
 kong/templates/kong_defaults.lua |   2 +
 6 files changed, 457 insertions(+), 3 deletions(-)
 create mode 100644 kong/resty/websocket/client.lua

diff --git a/kong-3.1.0-0.rockspec b/kong-3.1.0-0.rockspec
index 334e512ee5e2..45694cd03472 100644
--- a/kong-3.1.0-0.rockspec
+++ b/kong-3.1.0-0.rockspec
@@ -92,6 +92,7 @@ build = {
     ["kong.resty.dns.client"] = "kong/resty/dns/client.lua",
     ["kong.resty.dns.utils"] = "kong/resty/dns/utils.lua",
     ["kong.resty.ctx"] = "kong/resty/ctx.lua",
+    ["kong.resty.websocket.client"] = "kong/resty/websocket/client.lua",
 
     ["kong.cmd"] = "kong/cmd/init.lua",
     ["kong.cmd.roar"] = "kong/cmd/roar.lua",
diff --git a/kong.conf.default b/kong.conf.default
index 646da9219ed7..c9c6450ddfab 100644
--- a/kong.conf.default
+++ b/kong.conf.default
@@ -221,6 +221,10 @@
 #anonymous_reports = on          # Send anonymous usage data such as error
                                  # stack traces to help improve Kong.
 
+
+#proxy_server                    # Proxy server defined as a URL. Kong will only use
+                                 # option if any component is explictly configured
+                                 # to use proxy.
 #------------------------------------------------------------------------------
 # HYBRID MODE
 #------------------------------------------------------------------------------
@@ -390,6 +394,10 @@
                          #   found inside DP provided certificate
                          #   or communication with the OCSP responder
                          #   failed, then DP is still allowed through.
+#cluster_use_proxy = off
+                         # Whether to turn on HTTP CONNECT proxy support for
+                         # hybrid mode connections. `proxy_server` will be used
+                         # if this option is turned on.
 #------------------------------------------------------------------------------
 # NGINX
 #------------------------------------------------------------------------------
diff --git a/kong/clustering/utils.lua b/kong/clustering/utils.lua
index 0df232a2a34f..31eb9928bd83 100644
--- a/kong/clustering/utils.lua
+++ b/kong/clustering/utils.lua
@@ -5,10 +5,9 @@ local openssl_x509 = require("resty.openssl.x509")
 local ssl = require("ngx.ssl")
 local ocsp = require("ngx.ocsp")
 local http = require("resty.http")
-local ws_client = require("resty.websocket.client")
+local ws_client = require("kong.resty.websocket.client")
 local ws_server = require("resty.websocket.server")
-local utils = require("kong.tools.utils")
-local meta = require("kong.meta")
+local socket_url = require("socket.url")
 
 local type = type
 local tonumber = tonumber
@@ -17,6 +16,8 @@ local table_insert = table.insert
 local table_concat = table.concat
 local gsub = string.gsub
 local process_type = require("ngx.process").type
+local encode_base64 = ngx.encode_base64
+local fmt = string.format
 
 local kong = kong
 
@@ -324,6 +325,25 @@ function _M.check_configuration_compatibility(obj, dp_plugin_map)
   return true, nil, CLUSTERING_SYNC_STATUS.NORMAL
 end
 
+local function parse_proxy_url(conf)
+  local ret = {}
+  local proxy_server = conf.proxy_server
+  if proxy_server then
+    -- assume proxy_server is validated in conf_loader
+    local parsed = socket_url.parse(proxy_server)
+    ret.proxy_url = fmt("%s://%s:%s", parsed.scheme, parsed.host, parsed.port or 80)
+    ret.scheme = parsed.scheme
+    ret.host = parsed.host
+    ret.port = parsed.port
+
+    if parsed.user and parsed.password then
+      ret.proxy_authorization = "Basic " .. encode_base64(parsed.user  .. ":" .. parsed.password)
+    end
+  end
+
+  return ret
+end
+
 
 --- Return the highest supported Hybrid mode protocol version.
 function _M.check_protocol_support(conf, cert, cert_key)
@@ -347,6 +367,15 @@ function _M.check_protocol_support(conf, cert, cert_key)
   end
 
   local c = http.new()
+
+  if conf.cluster_use_proxy then
+    local proxy_opts = parse_proxy_url(conf)
+    c:set_proxy_options({
+      https_proxy = proxy_opts.proxy_url,
+      https_proxy_authorization = proxy_opts.proxy_authorization,
+    })
+  end
+
   local res, err = c:request_uri(
     "https://" .. conf.cluster_control_plane .. "/v1/wrpc", params)
   if not res then
@@ -383,6 +412,14 @@ function _M.connect_cp(endpoint, conf, cert, cert_key, protocols)
     protocols = protocols,
   }
 
+  if conf.cluster_use_proxy then
+    local proxy_opts = parse_proxy_url(conf)
+    opts.proxy_opts = {
+      wss_proxy = proxy_opts.proxy_url,
+      wss_proxy_authorization = proxy_opts.proxy_authorization,
+    }
+  end
+
   if conf.cluster_mtls == "shared" then
     opts.server_name = "kong_clustering"
 
diff --git a/kong/conf_loader/init.lua b/kong/conf_loader/init.lua
index c5cfb1a78bbd..0be84583c996 100644
--- a/kong/conf_loader/init.lua
+++ b/kong/conf_loader/init.lua
@@ -7,6 +7,7 @@ local openssl_pkey = require "resty.openssl.pkey"
 local openssl_x509 = require "resty.openssl.x509"
 local pl_stringio = require "pl.stringio"
 local pl_stringx = require "pl.stringx"
+local socket_url = require "socket.url"
 local constants = require "kong.constants"
 local listeners = require "kong.conf_loader.listeners"
 local pl_pretty = require "pl.pretty"
@@ -524,6 +525,7 @@ local CONF_INFERENCES = {
   cluster_data_plane_purge_delay = { typ = "number" },
   cluster_ocsp = { enum = { "on", "off", "optional" } },
   cluster_max_payload = { typ = "number" },
+  cluster_use_proxy = { typ = "boolean" },
 
   kic = { typ = "boolean" },
   pluginserver_names = { typ = "array" },
@@ -539,6 +541,8 @@ local CONF_INFERENCES = {
 
   opentelemetry_tracing = { typ = "array" },
   opentelemetry_tracing_sampling_rate = { typ = "number" },
+
+  proxy_server = { typ = "string" },
 }
 
 
@@ -549,6 +553,7 @@ local CONF_SENSITIVE = {
   pg_password = true,
   pg_ro_password = true,
   cassandra_password = true,
+  proxy_server = true, -- hide proxy server URL as it may contain credentials
 }
 
 
@@ -982,6 +987,28 @@ local function check_and_infer(conf, opts)
     errors[#errors + 1] = "worker_state_update_frequency must be greater than 0"
   end
 
+  if conf.proxy_server then
+    local parsed, err = socket_url.parse(conf.proxy_server)
+    if err then
+      errors[#errors + 1] = "proxy_server is invalid: " .. err
+
+    elseif not parsed.scheme then
+      errors[#errors + 1] = "proxy_server missing scheme"
+
+    elseif parsed.scheme ~= "http" then
+      errors[#errors + 1] = "proxy_server only supports \"http\", got " .. parsed.scheme
+
+    elseif not parsed.host then
+      errors[#errors + 1] = "proxy_server missing host"
+
+    elseif parsed.fragment or parsed.query or parsed.params then
+      errors[#errors + 1] = "proxy_server does not support fragments, query strings or parameters"
+
+    elseif (not not parsed.user) ~= (not not parsed.password) then
+      errors[#errors + 1] = "proxy_server must define or undefined username and password at same time"
+    end
+  end
+
   if conf.role == "control_plane" then
     if #conf.admin_listen < 1 or strip(conf.admin_listen[1]) == "off" then
       errors[#errors + 1] = "admin_listen must be specified when role = \"control_plane\""
@@ -999,6 +1026,10 @@ local function check_and_infer(conf, opts)
       errors[#errors + 1] = "in-memory storage can not be used when role = \"control_plane\""
     end
 
+    if conf.cluster_use_proxy then
+      errors[#errors + 1] = "cluster_use_proxy can not be used when role = \"control_plane\""
+    end
+
   elseif conf.role == "data_plane" then
     if #conf.proxy_listen < 1 or strip(conf.proxy_listen[1]) == "off" then
       errors[#errors + 1] = "proxy_listen must be specified when role = \"data_plane\""
@@ -1019,6 +1050,10 @@ local function check_and_infer(conf, opts)
     elseif conf.cluster_mtls == "pki" then
       insert(conf.lua_ssl_trusted_certificate, conf.cluster_ca_cert)
     end
+
+    if conf.cluster_use_proxy and not conf.proxy_server then
+      errors[#errors + 1] = "cluster_use_proxy is turned on but no proxy_server is configured"
+    end
   end
 
   if conf.cluster_data_plane_purge_delay < 60 then
diff --git a/kong/resty/websocket/client.lua b/kong/resty/websocket/client.lua
new file mode 100644
index 000000000000..4b5c1961b710
--- /dev/null
+++ b/kong/resty/websocket/client.lua
@@ -0,0 +1,371 @@
+-- Copyright (C) Yichun Zhang (agentzh)
+
+
+-- FIXME: this library is very rough and is currently just for testing
+--        the websocket server.
+
+
+local wbproto = require "resty.websocket.protocol"
+local bit = require "bit"
+
+
+local _recv_frame = wbproto.recv_frame
+local _send_frame = wbproto.send_frame
+local new_tab = wbproto.new_tab
+local tcp = ngx.socket.tcp
+local re_match = ngx.re.match
+local encode_base64 = ngx.encode_base64
+local concat = table.concat
+local char = string.char
+local str_find = string.find
+local rand = math.random
+local rshift = bit.rshift
+local band = bit.band
+local setmetatable = setmetatable
+local type = type
+local debug = ngx.config.debug
+local ngx_log = ngx.log
+local ngx_DEBUG = ngx.DEBUG
+local ssl_support = true
+
+if not ngx.config
+    or not ngx.config.ngx_lua_version
+    or ngx.config.ngx_lua_version < 9011
+then
+    ssl_support = false
+end
+
+local _M = new_tab(0, 13)
+_M._VERSION = '0.09'
+
+
+local mt = { __index = _M }
+
+
+function _M.new(self, opts)
+    local sock, err = tcp()
+    if not sock then
+        return nil, err
+    end
+
+    local max_payload_len, send_unmasked, timeout
+    if opts then
+        max_payload_len = opts.max_payload_len
+        send_unmasked = opts.send_unmasked
+        timeout = opts.timeout
+
+        if timeout then
+            sock:settimeout(timeout)
+        end
+    end
+
+    return setmetatable({
+        sock = sock,
+        max_payload_len = max_payload_len or 65535,
+        send_unmasked = send_unmasked,
+    }, mt)
+end
+
+
+function _M.connect(self, uri, opts)
+    local sock = self.sock
+    if not sock then
+        return nil, "not initialized"
+    end
+
+    local m, err = re_match(uri, [[^(wss?)://([^:/]+)(?::(\d+))?(.*)]], "jo")
+    if not m then
+        if err then
+            return nil, "failed to match the uri: " .. err
+        end
+
+        return nil, "bad websocket uri"
+    end
+
+    local scheme = m[1]
+    local host = m[2]
+    local port = m[3]
+    local path = m[4]
+
+    -- ngx.say("host: ", host)
+    -- ngx.say("port: ", port)
+
+    if not port then
+        port = 80
+    end
+
+    if path == "" then
+        path = "/"
+    end
+
+    local ssl_verify, server_name, headers, proto_header, origin_header, sock_opts = false
+    local client_cert, client_priv_key
+
+    if opts then
+        local protos = opts.protocols
+        if protos then
+            if type(protos) == "table" then
+                proto_header = "\r\nSec-WebSocket-Protocol: "
+                               .. concat(protos, ",")
+
+            else
+                proto_header = "\r\nSec-WebSocket-Protocol: " .. protos
+            end
+        end
+
+        local origin = opts.origin
+        if origin then
+            origin_header = "\r\nOrigin: " .. origin
+        end
+
+        local pool = opts.pool
+        if pool then
+            sock_opts = { pool = pool }
+        end
+
+        client_cert = opts.client_cert
+        client_priv_key = opts.client_priv_key
+
+        if client_cert then
+            assert(client_priv_key,
+                   "client_priv_key must be provided with client_cert")
+        end
+
+        if opts.ssl_verify or opts.server_name then
+            if not ssl_support then
+                return nil, "ngx_lua 0.9.11+ required for SSL sockets"
+            end
+            ssl_verify = opts.ssl_verify
+            server_name = opts.server_name or host
+        end
+
+        if opts.headers then
+            headers = opts.headers
+            if type(headers) ~= "table" then
+                return nil, "custom headers must be a table"
+            end
+        end
+    end
+
+    local ok, err
+    if sock_opts then
+        ok, err = sock:connect(host, port, sock_opts)
+    else
+        ok, err = sock:connect(host, port)
+    end
+    if not ok then
+        return nil, "failed to connect: " .. err
+    end
+
+    if scheme == "wss" then
+        if not ssl_support then
+            return nil, "ngx_lua 0.9.11+ required for SSL sockets"
+        end
+
+        if client_cert then
+            ok, err = sock:setclientcert(client_cert, client_priv_key)
+            if not ok then
+                return nil, "ssl client cert failed: " .. err
+            end
+        end
+
+        ok, err = sock:sslhandshake(false, server_name, ssl_verify)
+        if not ok then
+            return nil, "ssl handshake failed: " .. err
+        end
+    end
+
+    -- check for connections from pool:
+
+    local count, err = sock:getreusedtimes()
+    if not count then
+        return nil, "failed to get reused times: " .. err
+    end
+    if count > 0 then
+        -- being a reused connection (must have done handshake)
+        return 1
+    end
+
+    local custom_headers
+    if headers then
+        custom_headers = concat(headers, "\r\n")
+        custom_headers = "\r\n" .. custom_headers
+    end
+
+    -- do the websocket handshake:
+
+    local bytes = char(rand(256) - 1, rand(256) - 1, rand(256) - 1,
+                       rand(256) - 1, rand(256) - 1, rand(256) - 1,
+                       rand(256) - 1, rand(256) - 1, rand(256) - 1,
+                       rand(256) - 1, rand(256) - 1, rand(256) - 1,
+                       rand(256) - 1, rand(256) - 1, rand(256) - 1,
+                       rand(256) - 1)
+
+    local key = encode_base64(bytes)
+    local req = "GET " .. path .. " HTTP/1.1\r\nUpgrade: websocket\r\nHost: "
+                .. host .. ":" .. port
+                .. "\r\nSec-WebSocket-Key: " .. key
+                .. (proto_header or "")
+                .. "\r\nSec-WebSocket-Version: 13"
+                .. (origin_header or "")
+                .. "\r\nConnection: Upgrade"
+                .. (custom_headers or "")
+                .. "\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, partial = header_reader()
+    if not header then
+        return nil, "failed to receive response header: " .. err
+    end
+
+    -- error("header: " .. header)
+
+    -- FIXME: verify the response headers
+
+    m, err = re_match(header, [[^\s*HTTP/1\.1\s+]], "jo")
+    if not m then
+        return nil, "bad HTTP response status line: " .. header
+    end
+
+    return 1
+end
+
+
+function _M.set_timeout(self, time)
+    local sock = self.sock
+    if not sock then
+        return nil, nil, "not initialized yet"
+    end
+
+    return sock:settimeout(time)
+end
+
+
+function _M.recv_frame(self)
+    if self.fatal then
+        return nil, nil, "fatal error already happened"
+    end
+
+    local sock = self.sock
+    if not sock then
+        return nil, nil, "not initialized yet"
+    end
+
+    local data, typ, err =  _recv_frame(sock, self.max_payload_len, false)
+    if not data and not str_find(err, ": timeout", 1, true) then
+        self.fatal = true
+    end
+    return data, typ, err
+end
+
+
+local function send_frame(self, fin, opcode, payload)
+    if self.fatal then
+        return nil, "fatal error already happened"
+    end
+
+    if self.closed then
+        return nil, "already closed"
+    end
+
+    local sock = self.sock
+    if not sock then
+        return nil, "not initialized yet"
+    end
+
+    local bytes, err = _send_frame(sock, fin, opcode, payload,
+                                   self.max_payload_len,
+                                   not self.send_unmasked)
+    if not bytes then
+        self.fatal = true
+    end
+    return bytes, err
+end
+_M.send_frame = send_frame
+
+
+function _M.send_text(self, data)
+    return send_frame(self, true, 0x1, data)
+end
+
+
+function _M.send_binary(self, data)
+    return send_frame(self, true, 0x2, data)
+end
+
+
+local function send_close(self, code, msg)
+    local payload
+    if code then
+        if type(code) ~= "number" or code > 0x7fff then
+            return nil, "bad status code"
+        end
+        payload = char(band(rshift(code, 8), 0xff), band(code, 0xff))
+                        .. (msg or "")
+    end
+
+    if debug then
+        ngx_log(ngx_DEBUG, "sending the close frame")
+    end
+
+    local bytes, err = send_frame(self, true, 0x8, payload)
+
+    if not bytes then
+        self.fatal = true
+    end
+
+    self.closed = true
+
+    return bytes, err
+end
+_M.send_close = send_close
+
+
+function _M.send_ping(self, data)
+    return send_frame(self, true, 0x9, data)
+end
+
+
+function _M.send_pong(self, data)
+    return send_frame(self, true, 0xa, data)
+end
+
+
+function _M.close(self)
+    if self.fatal then
+        return nil, "fatal error already happened"
+    end
+
+    local sock = self.sock
+    if not sock then
+        return nil, "not initialized"
+    end
+
+    if not self.closed then
+        local bytes, err = send_close(self)
+        if not bytes then
+            return nil, "failed to send close frame: " .. err
+        end
+    end
+
+    return sock:close()
+end
+
+
+function _M.set_keepalive(self, ...)
+    local sock = self.sock
+    if not sock then
+        return nil, "not initialized"
+    end
+
+    return sock:setkeepalive(...)
+end
+
+
+return _M
diff --git a/kong/templates/kong_defaults.lua b/kong/templates/kong_defaults.lua
index b92634d146c8..8e0369f143e3 100644
--- a/kong/templates/kong_defaults.lua
+++ b/kong/templates/kong_defaults.lua
@@ -14,6 +14,7 @@ plugins = bundled
 port_maps = NONE
 host_ports = NONE
 anonymous_reports = on
+proxy_server = NONE
 
 proxy_listen = 0.0.0.0:8000 reuseport backlog=16384, 0.0.0.0:8443 http2 ssl reuseport backlog=16384
 stream_listen = off
@@ -29,6 +30,7 @@ cluster_server_name = NONE
 cluster_data_plane_purge_delay = 1209600
 cluster_ocsp = off
 cluster_max_payload = 4194304
+cluster_use_proxy = off
 
 lmdb_environment_path = dbless.lmdb
 lmdb_map_size = 128m

From 5b178c775e47e8b3c5d14a6482f76f01f45212a4 Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Mon, 14 Nov 2022 22:42:20 -0800
Subject: [PATCH 02/25] add kong's patch

---
 kong/resty/websocket/client.lua | 77 +++++++++++++++++++++++++++++++--
 1 file changed, 73 insertions(+), 4 deletions(-)

diff --git a/kong/resty/websocket/client.lua b/kong/resty/websocket/client.lua
index 4b5c1961b710..ec8501b156bd 100644
--- a/kong/resty/websocket/client.lua
+++ b/kong/resty/websocket/client.lua
@@ -4,6 +4,9 @@
 -- FIXME: this library is very rough and is currently just for testing
 --        the websocket server.
 
+-- Modifications by Kong Inc.
+--   * added forward proxy support
+
 
 local wbproto = require "resty.websocket.protocol"
 local bit = require "bit"
@@ -147,11 +150,41 @@ function _M.connect(self, uri, opts)
         end
     end
 
+    local connect_host, connect_port = host, port
+    local proxy_opts = 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
+        -- 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 m[1] ~= "http" then
+            return nil, "only HTTP proxy is supported"
+        end
+
+        connect_host = m[2]
+        connect_port = m[3] or 80 -- hardcode for now as we only support HTTP proxy
+
+        if not connect_host then
+            return nil, "invalid proxy url"
+        end
+    end
+
     local ok, err
     if sock_opts then
-        ok, err = sock:connect(host, port, sock_opts)
+        ok, err = sock:connect(connect_host, connect_port, sock_opts)
     else
-        ok, err = sock:connect(host, port)
+        ok, err = sock:connect(connect_host, connect_port)
     end
     if not ok then
         return nil, "failed to connect: " .. err
@@ -162,6 +195,42 @@ function _M.connect(self, uri, opts)
             return nil, "ngx_lua 0.9.11+ required for SSL sockets"
         end
 
+        if proxy_url then
+            local req = "CONNECT " .. host .. ":" .. port .. " HTTP/1.1"
+            .. "\r\nHost: " .. host .. ":" .. 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\.1\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
@@ -219,7 +288,7 @@ function _M.connect(self, uri, opts)
 
     local header_reader = sock:receiveuntil("\r\n\r\n")
     -- FIXME: check for too big response headers
-    local header, err, partial = header_reader()
+    local header, err, _ = header_reader()
     if not header then
         return nil, "failed to receive response header: " .. err
     end
@@ -228,7 +297,7 @@ function _M.connect(self, uri, opts)
 
     -- FIXME: verify the response headers
 
-    m, err = re_match(header, [[^\s*HTTP/1\.1\s+]], "jo")
+    m, _ = re_match(header, [[^\s*HTTP/1\.1\s+]], "jo")
     if not m then
         return nil, "bad HTTP response status line: " .. header
     end

From 551e16d6d71a1060abae991f8061a4b815e44214 Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Mon, 14 Nov 2022 23:20:49 -0800
Subject: [PATCH 03/25] run less

---
 .github/workflows/build_and_test.yml | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/.github/workflows/build_and_test.yml b/.github/workflows/build_and_test.yml
index 63a6bc9807ad..ba69d2d1477a 100644
--- a/.github/workflows/build_and_test.yml
+++ b/.github/workflows/build_and_test.yml
@@ -145,8 +145,8 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        suite: [integration, plugins]
-        split: [first (01-04), second (>= 05)]
+        suite: [integration]
+        split: [first (01-04)]
 
     env:
       KONG_TEST_PG_DATABASE: kong
@@ -246,6 +246,7 @@ jobs:
     name: DB-less integration tests
     runs-on: ubuntu-22.04
     needs: build
+    if: false
 
     env:
       KONG_TEST_PG_DATABASE: kong
@@ -312,6 +313,7 @@ jobs:
     name: C* ${{ matrix.cassandra_version }} ${{ matrix.suite }} - ${{ matrix.split }} tests
     runs-on: ubuntu-22.04
     needs: build
+    if: false
 
     strategy:
       fail-fast: false
@@ -415,6 +417,7 @@ jobs:
     name: PDK tests
     runs-on: ubuntu-22.04
     needs: build
+    if: false
 
     env:
       TEST_SUITE: pdk

From c957231b1dee41add2f5f596a65fcf9277c79233 Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Tue, 15 Nov 2022 02:02:05 -0800
Subject: [PATCH 04/25] add some tests

---
 kong/conf_loader/init.lua                     |   3 -
 spec/01-unit/03-conf_loader_spec.lua          |  64 +++++++
 .../09-hybrid_mode/09-forward-proxy_spec.lua  | 176 ++++++++++++++++++
 3 files changed, 240 insertions(+), 3 deletions(-)
 create mode 100644 spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua

diff --git a/kong/conf_loader/init.lua b/kong/conf_loader/init.lua
index 0be84583c996..4958d822b9cc 100644
--- a/kong/conf_loader/init.lua
+++ b/kong/conf_loader/init.lua
@@ -1003,9 +1003,6 @@ local function check_and_infer(conf, opts)
 
     elseif parsed.fragment or parsed.query or parsed.params then
       errors[#errors + 1] = "proxy_server does not support fragments, query strings or parameters"
-
-    elseif (not not parsed.user) ~= (not not parsed.password) then
-      errors[#errors + 1] = "proxy_server must define or undefined username and password at same time"
     end
   end
 
diff --git a/spec/01-unit/03-conf_loader_spec.lua b/spec/01-unit/03-conf_loader_spec.lua
index fa8796f0f1c1..642634190367 100644
--- a/spec/01-unit/03-conf_loader_spec.lua
+++ b/spec/01-unit/03-conf_loader_spec.lua
@@ -969,6 +969,70 @@ describe("Configuration loader", function()
           )
           assert.matches(".ca_combined", conf.lua_ssl_trusted_certificate_combined)
         end)
+        it("validates proxy_server", function()
+          local conf, _, errors = conf_loader(nil, {
+            proxy_server = "http://cool:pwd@localhost:2333",
+          })
+          assert.is_nil(errors)
+          assert.is_table(conf)
+
+          local conf, _, errors = conf_loader(nil, {
+            proxy_server = "://localhost:2333",
+          })
+          assert.contains("proxy_server missing scheme", errors)
+          assert.is_nil(conf)
+
+
+          local conf, _, errors = conf_loader(nil, {
+            proxy_server = "cool://localhost:2333",
+          })
+          assert.contains("proxy_server only supports \"http\", got cool", errors)
+          assert.is_nil(conf)
+
+          local conf, _, errors = conf_loader(nil, {
+            proxy_server = "http://:2333",
+          })
+          assert.contains("proxy_server missing host", errors)
+          assert.is_nil(conf)
+
+
+          local conf, _, errors = conf_loader(nil, {
+            proxy_server = "http://localhost:2333/?a=1",
+          })
+          assert.contains("proxy_server does not support fragments, query strings or parameters", errors)
+          assert.is_nil(conf)
+        end)
+        it("doesn't allow cluster_use_proxy on CP but allows on DP", function()
+          local conf, _, errors = conf_loader(nil, {
+            role = "data_plane",
+            database = "off",
+            cluster_cert = "spec/fixtures/kong_clustering.crt",
+            cluster_cert_key = "spec/fixtures/kong_clustering.key",
+            cluster_use_proxy = "on",
+          })
+          assert.contains("cluster_use_proxy is turned on but no proxy_server is configured", errors)
+          assert.is_nil(conf)
+
+          local conf, _, errors = conf_loader(nil, {
+            role = "data_plane",
+            database = "off",
+            cluster_cert = "spec/fixtures/kong_clustering.crt",
+            cluster_cert_key = "spec/fixtures/kong_clustering.key",
+            cluster_use_proxy = "on",
+            proxy_server = "http://user:pass@localhost:2333/",
+          })
+          assert.is_nil(errors)
+          assert.is_table(conf)
+
+          local conf, _, errors = conf_loader(nil, {
+            role = "control_plane",
+            cluster_cert = "spec/fixtures/kong_clustering.crt",
+            cluster_cert_key = "spec/fixtures/kong_clustering.key",
+            cluster_use_proxy = "on",
+          })
+          assert.contains("cluster_use_proxy can not be used when role = \"control_plane\"", errors)
+          assert.is_nil(conf)
+        end)
         it("doen't overwrite lua_ssl_trusted_certificate when autoload cluster_cert or cluster_ca_cert", function()
           local conf, _, errors = conf_loader(nil, {
             role = "data_plane",
diff --git a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
new file mode 100644
index 000000000000..c39ebf80edc5
--- /dev/null
+++ b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
@@ -0,0 +1,176 @@
+local helpers = require "spec.helpers"
+local pl_path      = require "pl.path"
+local pl_file      = require "pl.file"
+
+
+local fixtures = {
+  stream_mock = {
+    forward_proxy = [[
+    server {
+      listen 16797;
+      error_log logs/proxy.log debug;
+
+      content_by_lua_block {
+        require("spec.fixtures.forward-proxy-server").connect()
+      }
+    }
+
+    server {
+      listen 16796;
+      error_log logs/proxy_auth.log debug;
+
+      content_by_lua_block {
+        require("spec.fixtures.forward-proxy-server").connect({
+          basic_auth = ngx.encode_base64("test:konghq"),
+        })
+      }
+    }
+
+    ]],
+  },
+}
+
+
+local confs = helpers.get_clustering_protocols()
+
+local auth_confgs = {
+  ["auth off"] = "http://127.0.0.1:16797",
+  ["auth on"] = "http://test:konghq@127.0.0.1:16796",
+}
+
+
+for _, strategy in helpers.each_strategy() do
+  for auth_desc, proxy_url in pairs(auth_confgs) do
+  for cluster_protocol, conf in pairs(confs) do
+    describe("CP/DP sync through proxy (" .. auth_desc .. ") works with #" .. strategy .. " backend, protocol " .. cluster_protocol, function()
+      lazy_setup(function()
+        helpers.get_db_utils(strategy) -- runs migrations
+
+        assert(helpers.start_kong({
+          role = "control_plane",
+          --legacy_hybrid_protocol = (cluster_protocol == "json (by switch)"),
+          cluster_cert = "spec/fixtures/kong_clustering.crt",
+          cluster_cert_key = "spec/fixtures/kong_clustering.key",
+          database = strategy,
+          db_update_frequency = 0.1,
+          cluster_listen = "127.0.0.1:9005",
+          nginx_conf = conf,
+        }))
+
+        assert(helpers.start_kong({
+          role = "data_plane",
+          --legacy_hybrid_protocol = (cluster_protocol == "json (by switch)"),
+          cluster_protocol = cluster_protocol,
+          database = "off",
+          prefix = "servroot2",
+          cluster_cert = "spec/fixtures/kong_clustering.crt",
+          cluster_cert_key = "spec/fixtures/kong_clustering.key",
+          cluster_control_plane = "127.0.0.1:9005",
+          proxy_listen = "0.0.0.0:9002",
+          log_level = "debug",
+
+          -- cluster_use_proxy = "on",
+          proxy_server = proxy_url,
+
+          -- this is unused, but required for the the template to include a stream {} block
+          stream_listen = "0.0.0.0:5555",
+        }, nil, nil, fixtures))
+
+        for _, plugin in ipairs(helpers.get_plugins_list()) do
+        end
+      end)
+
+      lazy_teardown(function()
+        helpers.stop_kong("servroot2")
+        helpers.stop_kong()
+      end)
+
+      describe("sync works", function()
+        it("pushes first change asap and following changes in a batch", function()
+          local admin_client = helpers.admin_client(10000)
+          local proxy_client = helpers.http_client("127.0.0.1", 9002)
+          finally(function()
+            admin_client:close()
+            proxy_client:close()
+          end)
+
+          local res = admin_client:put("/routes/1", {
+            headers = {
+              ["Content-Type"] = "application/json",
+            },
+            body = {
+              paths = { "/1" },
+            },
+          })
+
+          assert.res_status(200, res)
+
+          helpers.wait_until(function()
+            local proxy_client = helpers.http_client("127.0.0.1", 9002)
+            -- serviceless route should return 503 instead of 404
+            res = proxy_client:get("/1")
+            proxy_client:close()
+            if res and res.status == 503 then
+              return true
+            end
+          end, 10)
+
+          for i = 2, 5 do
+            res = admin_client:put("/routes/" .. i, {
+              headers = {
+                ["Content-Type"] = "application/json",
+              },
+              body = {
+                paths = { "/" .. i },
+              },
+            })
+
+            assert.res_status(200, res)
+          end
+
+          helpers.wait_until(function()
+            local proxy_client = helpers.http_client("127.0.0.1", 9002)
+            -- serviceless route should return 503 instead of 404
+            res = proxy_client:get("/5")
+            proxy_client:close()
+            if res and res.status == 503 then
+              return true
+            end
+          end, 5)
+
+          for i = 4, 2, -1 do
+            res = proxy_client:get("/" .. i)
+            assert.res_status(503, res)
+          end
+
+          for i = 1, 5 do
+            local res = admin_client:delete("/routes/" .. i)
+            assert.res_status(204, res)
+          end
+
+          helpers.wait_until(function()
+            local proxy_client = helpers.http_client("127.0.0.1", 9002)
+            -- deleted route should return 404
+            res = proxy_client:get("/1")
+            proxy_client:close()
+            if res and res.status == 404 then
+              return true
+            end
+          end, 5)
+
+          for i = 5, 2, -1 do
+            res = proxy_client:get("/" .. i)
+            assert.res_status(404, res)
+          end
+
+          -- ensure this goes through proxy
+          local path = pl_path.join("servroot2", "logs", "proxy.log")
+          local contents = pl_file.read(path)
+          assert.matches("CONNECT 127.0.0.1:9005", contents)
+        end)
+      end)
+    end)
+
+  end -- cluster protocols
+  end -- auth configs 
+end

From b2275febc9e7ec2ef05cd307ef9ac96bd47c048e Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Tue, 15 Nov 2022 22:12:55 -0800
Subject: [PATCH 05/25] add basic auth to forward-proxy-server fixture

---
 spec/fixtures/forward-proxy-server.lua | 52 ++++++++++++++++++++++++--
 1 file changed, 48 insertions(+), 4 deletions(-)

diff --git a/spec/fixtures/forward-proxy-server.lua b/spec/fixtures/forward-proxy-server.lua
index 61bda1196a9b..3725cf13e988 100644
--- a/spec/fixtures/forward-proxy-server.lua
+++ b/spec/fixtures/forward-proxy-server.lua
@@ -2,11 +2,25 @@ local _M = {}
 
 local split = require("kong.tools.utils").split
 
+local header_mt = {
+  __index = function(self, name)
+    name = name:lower():gsub("_", "-")
+    return rawget(self, name)
+  end,
+
+  __newindex = function(self, name, value)
+    name = name:lower():gsub("_", "-")
+    rawset(self, name, value)
+  end,
+}
+
+local function new_headers()
+  return setmetatable({}, header_mt)
+end
 
 -- This is a very naive forward proxy, which accepts a CONNECT over HTTP, and
 -- then starts tunnelling the bytes blind (for end-to-end SSL).
-function _M.connect()
-
+function _M.connect(opts)
   local req_sock = ngx.req.socket(true)
   req_sock:settimeouts(1000, 1000, 1000)
 
@@ -21,12 +35,42 @@ function _M.connect()
 
   local upstream_host, upstream_port = unpack(split(host_port, ":"))
 
-  -- receive and discard any headers
+  local headers = new_headers()
+
+  -- receive headers
   repeat
     local line = req_sock:receive("*l")
-    ngx.log(ngx.DEBUG, "request header: ", line)
+    local name, value = line:match("^([^:]+):%s*(.+)$")
+    if name and value then
+      ngx.log(ngx.DEBUG, "header: ", name, " => ", value)
+      headers[name] = value
+    end
   until ngx.re.find(line, "^\\s*$", "jo")
 
+
+  local basic_auth = opts and opts.basic_auth
+  if basic_auth then
+    ngx.log(ngx.DEBUG, "checking proxy-authorization...")
+
+    local found = headers["proxy-authorization"]
+    if not found then
+      ngx.log(ngx.NOTICE, "client did not send proxy-authorization header")
+      ngx.print("HTTP/1.1 401 Unauthorized\r\n\r\n")
+      return ngx.exit(ngx.OK)
+    end
+
+    local auth = ngx.re.gsub(found, [[^Basic\s*]], "", "oji")
+
+    if auth ~= basic_auth then
+      ngx.log(ngx.NOTICE, "client sent incorrect proxy-authorization")
+      ngx.print("HTTP/1.1 403 Forbidden\r\n\r\n")
+      return ngx.exit(ngx.OK)
+    end
+
+    ngx.log(ngx.DEBUG, "accepted basic proxy-authorization")
+  end
+
+
   -- Connect to requested upstream
   local upstream_sock = ngx.socket.tcp()
   upstream_sock:settimeouts(1000, 1000, 1000)

From 0dc4697d78ee2beb52d6212cb0fc55ecb32c00e2 Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Tue, 15 Nov 2022 22:19:55 -0800
Subject: [PATCH 06/25] try a different test

---
 .../09-hybrid_mode/09-forward-proxy_spec.lua  | 80 ++++---------------
 1 file changed, 17 insertions(+), 63 deletions(-)

diff --git a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
index c39ebf80edc5..3fad413da47a 100644
--- a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
+++ b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
@@ -76,8 +76,6 @@ for _, strategy in helpers.each_strategy() do
           stream_listen = "0.0.0.0:5555",
         }, nil, nil, fixtures))
 
-        for _, plugin in ipairs(helpers.get_plugins_list()) do
-        end
       end)
 
       lazy_teardown(function()
@@ -86,82 +84,38 @@ for _, strategy in helpers.each_strategy() do
       end)
 
       describe("sync works", function()
-        it("pushes first change asap and following changes in a batch", function()
+        it("proxy on DP follows CP config", function()
           local admin_client = helpers.admin_client(10000)
-          local proxy_client = helpers.http_client("127.0.0.1", 9002)
           finally(function()
             admin_client:close()
-            proxy_client:close()
           end)
 
-          local res = admin_client:put("/routes/1", {
-            headers = {
-              ["Content-Type"] = "application/json",
-            },
-            body = {
-              paths = { "/1" },
-            },
-          })
+          local res = assert(admin_client:post("/services", {
+            body = { name = "mockbin-service", url = "https://127.0.0.1:15556/request", },
+            headers = {["Content-Type"] = "application/json"}
+          }))
+          assert.res_status(201, res)
 
-          assert.res_status(200, res)
+          res = assert(admin_client:post("/services/mockbin-service/routes", {
+            body = { paths = { "/" }, },
+            headers = {["Content-Type"] = "application/json"}
+          }))
+          assert.res_status(201, res)
 
           helpers.wait_until(function()
             local proxy_client = helpers.http_client("127.0.0.1", 9002)
-            -- serviceless route should return 503 instead of 404
-            res = proxy_client:get("/1")
-            proxy_client:close()
-            if res and res.status == 503 then
-              return true
-            end
-          end, 10)
 
-          for i = 2, 5 do
-            res = admin_client:put("/routes/" .. i, {
-              headers = {
-                ["Content-Type"] = "application/json",
-              },
-              body = {
-                paths = { "/" .. i },
-              },
+            res = proxy_client:send({
+              method  = "GET",
+              path    = "/",
             })
 
-            assert.res_status(200, res)
-          end
-
-          helpers.wait_until(function()
-            local proxy_client = helpers.http_client("127.0.0.1", 9002)
-            -- serviceless route should return 503 instead of 404
-            res = proxy_client:get("/5")
-            proxy_client:close()
-            if res and res.status == 503 then
-              return true
-            end
-          end, 5)
-
-          for i = 4, 2, -1 do
-            res = proxy_client:get("/" .. i)
-            assert.res_status(503, res)
-          end
-
-          for i = 1, 5 do
-            local res = admin_client:delete("/routes/" .. i)
-            assert.res_status(204, res)
-          end
-
-          helpers.wait_until(function()
-            local proxy_client = helpers.http_client("127.0.0.1", 9002)
-            -- deleted route should return 404
-            res = proxy_client:get("/1")
+            local status = res and res.status
             proxy_client:close()
-            if res and res.status == 404 then
+            if status == 200 then
               return true
             end
-          end, 5)
-
-          for i = 5, 2, -1 do
-            res = proxy_client:get("/" .. i)
-            assert.res_status(404, res)
-          end
+          end, 10)
 
           -- ensure this goes through proxy
           local path = pl_path.join("servroot2", "logs", "proxy.log")

From f48f125a0be7a848ace7233fb1e953671017ddbb Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Tue, 15 Nov 2022 22:32:17 -0800
Subject: [PATCH 07/25] run less tests

---
 .ci/run_tests.sh | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.ci/run_tests.sh b/.ci/run_tests.sh
index 865e5bfd6371..15501cc9a287 100755
--- a/.ci/run_tests.sh
+++ b/.ci/run_tests.sh
@@ -33,7 +33,7 @@ fi
 if [ "$TEST_SUITE" == "integration" ]; then
     if [[ "$TEST_SPLIT" == first* ]]; then
         # GitHub Actions, run first batch of integration tests
-        eval "$TEST_CMD" $(ls -d spec/02-integration/* | sort | grep -v 05-proxy)
+        eval "$TEST_CMD" $(ls -d spec/02-integration/* | sort | grep 09-hybrid)
 
     elif [[ "$TEST_SPLIT" == second* ]]; then
         # GitHub Actions, run second batch of integration tests

From 3c504d2019c16cf1379f58066efbb0325edb31db Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Tue, 15 Nov 2022 22:53:44 -0800
Subject: [PATCH 08/25] fix test

---
 spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
index 3fad413da47a..542e50e84809 100644
--- a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
+++ b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
@@ -118,7 +118,8 @@ for _, strategy in helpers.each_strategy() do
           end, 10)
 
           -- ensure this goes through proxy
-          local path = pl_path.join("servroot2", "logs", "proxy.log")
+          local path = pl_path.join("servroot2", "logs",
+                      auth_desc == "auth on" and "proxy_auth.log" or "proxy.log")
           local contents = pl_file.read(path)
           assert.matches("CONNECT 127.0.0.1:9005", contents)
         end)

From a8f0c86b6d324368d7e67bdb0f932c95f95be101 Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Tue, 15 Nov 2022 23:05:40 -0800
Subject: [PATCH 09/25] run only one test

---
 .ci/run_tests.sh | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.ci/run_tests.sh b/.ci/run_tests.sh
index 15501cc9a287..ebb618b15acf 100755
--- a/.ci/run_tests.sh
+++ b/.ci/run_tests.sh
@@ -33,7 +33,7 @@ fi
 if [ "$TEST_SUITE" == "integration" ]; then
     if [[ "$TEST_SPLIT" == first* ]]; then
         # GitHub Actions, run first batch of integration tests
-        eval "$TEST_CMD" $(ls -d spec/02-integration/* | sort | grep 09-hybrid)
+        eval "$TEST_CMD" spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
 
     elif [[ "$TEST_SPLIT" == second* ]]; then
         # GitHub Actions, run second batch of integration tests

From 3e9bf1b4dd969ec6c09bb960166491af2ed9ec5f Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Tue, 15 Nov 2022 23:05:48 -0800
Subject: [PATCH 10/25] test tests

---
 .../09-hybrid_mode/09-forward-proxy_spec.lua          | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
index 542e50e84809..5ab6fac7da9a 100644
--- a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
+++ b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
@@ -11,6 +11,7 @@ local fixtures = {
       error_log logs/proxy.log debug;
 
       content_by_lua_block {
+        ngx.log(ngx.INFO, "started")
         require("spec.fixtures.forward-proxy-server").connect()
       }
     }
@@ -20,6 +21,7 @@ local fixtures = {
       error_log logs/proxy_auth.log debug;
 
       content_by_lua_block {
+        ngx.log(ngx.INFO, "started")
         require("spec.fixtures.forward-proxy-server").connect({
           basic_auth = ngx.encode_base64("test:konghq"),
         })
@@ -117,11 +119,18 @@ for _, strategy in helpers.each_strategy() do
             end
           end, 10)
 
+          os.execute("find servroot2/")
+          os.execute("find servroot/")
+
           -- ensure this goes through proxy
           local path = pl_path.join("servroot2", "logs",
-                      auth_desc == "auth on" and "proxy_auth.log" or "proxy.log")
+                      (auth_desc == "auth on") and "proxy_auth.log" or "proxy.log")
           local contents = pl_file.read(path)
           assert.matches("CONNECT 127.0.0.1:9005", contents)
+
+          if auth_desc == "auth on" then
+            assert.matches("accepted basic proxy-authorization", contents)
+          end
         end)
       end)
     end)

From c4ead1237a8ac14dec48270b7dc0b2dc0188c44c Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Tue, 15 Nov 2022 23:22:28 -0800
Subject: [PATCH 11/25] add nginx_conf

---
 spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
index 5ab6fac7da9a..25e5625e4107 100644
--- a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
+++ b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
@@ -71,6 +71,9 @@ for _, strategy in helpers.each_strategy() do
           proxy_listen = "0.0.0.0:9002",
           log_level = "debug",
 
+          -- used to render the mock fixture
+          nginx_conf = "spec/fixtures/custom_nginx.template",
+
           -- cluster_use_proxy = "on",
           proxy_server = proxy_url,
 

From 75260309b850806c595b7dc93e4e3dae2c85bae5 Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Tue, 15 Nov 2022 23:24:41 -0800
Subject: [PATCH 12/25] debug

---
 .../09-hybrid_mode/09-forward-proxy_spec.lua     | 16 ++++++++++------
 1 file changed, 10 insertions(+), 6 deletions(-)

diff --git a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
index 25e5625e4107..697b4d09fe8d 100644
--- a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
+++ b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
@@ -11,7 +11,6 @@ local fixtures = {
       error_log logs/proxy.log debug;
 
       content_by_lua_block {
-        ngx.log(ngx.INFO, "started")
         require("spec.fixtures.forward-proxy-server").connect()
       }
     }
@@ -21,7 +20,6 @@ local fixtures = {
       error_log logs/proxy_auth.log debug;
 
       content_by_lua_block {
-        ngx.log(ngx.INFO, "started")
         require("spec.fixtures.forward-proxy-server").connect({
           basic_auth = ngx.encode_base64("test:konghq"),
         })
@@ -37,7 +35,7 @@ local confs = helpers.get_clustering_protocols()
 
 local auth_confgs = {
   ["auth off"] = "http://127.0.0.1:16797",
-  ["auth on"] = "http://test:konghq@127.0.0.1:16796",
+  -- ["auth on"] = "http://test:konghq@127.0.0.1:16796",
 }
 
 
@@ -74,7 +72,7 @@ for _, strategy in helpers.each_strategy() do
           -- used to render the mock fixture
           nginx_conf = "spec/fixtures/custom_nginx.template",
 
-          -- cluster_use_proxy = "on",
+          cluster_use_proxy = "on",
           proxy_server = proxy_url,
 
           -- this is unused, but required for the the template to include a stream {} block
@@ -84,6 +82,12 @@ for _, strategy in helpers.each_strategy() do
       end)
 
       lazy_teardown(function()
+
+        os.execute("echo @@@@@@@@@@@@@@; cat servroot2/logs/proxy_auth.log")
+        os.execute("echo @@@@@@@@@@@@@@; cat servroot2/logs/proxy.log")
+
+        os.execute("echo @@@@@@@@@@@@@@; cat servroot2/logs/error.log")
+
         helpers.stop_kong("servroot2")
         helpers.stop_kong()
       end)
@@ -122,8 +126,8 @@ for _, strategy in helpers.each_strategy() do
             end
           end, 10)
 
-          os.execute("find servroot2/")
-          os.execute("find servroot/")
+          os.execute("echo @@@@@@@@@@@@@@; cat servroot2/logs/proxy_auth.log")
+          os.execute("echo @@@@@@@@@@@@@@; cat servroot2/logs/proxy.log")
 
           -- ensure this goes through proxy
           local path = pl_path.join("servroot2", "logs",

From 3f341ce81afebff8d9a8743be9affcf6676fb351 Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Wed, 16 Nov 2022 00:47:49 -0800
Subject: [PATCH 13/25] fix forward proxy fixture

---
 spec/fixtures/forward-proxy-server.lua | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/spec/fixtures/forward-proxy-server.lua b/spec/fixtures/forward-proxy-server.lua
index 3725cf13e988..ec2ff1b1b718 100644
--- a/spec/fixtures/forward-proxy-server.lua
+++ b/spec/fixtures/forward-proxy-server.lua
@@ -82,9 +82,11 @@ function _M.connect(opts)
   end
 
   -- Tell the client we are good to go
-  ngx.print("HTTP/1.1 200 OK\n\n")
+  ngx.print("HTTP/1.1 200 OK\r\n\r\n")
   ngx.flush()
 
+  ngx.log(ngx.DEBUG, "tunneling started")
+
   -- 10Kb in either direction should be plenty
   local max_bytes = 10 * 1024
 
@@ -115,6 +117,8 @@ function _M.connect(opts)
   until not req_data and not res_data -- request socket should be closed
 
   upstream_sock:close()
+
+  ngx.log(ngx.DEBUG, "tunneling ended")
 end
 
 return _M

From 580955770fb9adee81cea6950958d5e72f015f37 Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Wed, 16 Nov 2022 00:47:58 -0800
Subject: [PATCH 14/25] add debnug logging, cleanup test

---
 kong/clustering/utils.lua                           |  7 +++++++
 .../09-hybrid_mode/09-forward-proxy_spec.lua        | 13 ++-----------
 2 files changed, 9 insertions(+), 11 deletions(-)

diff --git a/kong/clustering/utils.lua b/kong/clustering/utils.lua
index 31eb9928bd83..2586deca0f19 100644
--- a/kong/clustering/utils.lua
+++ b/kong/clustering/utils.lua
@@ -24,6 +24,7 @@ local kong = kong
 local ngx = ngx
 local ngx_var = ngx.var
 local ngx_log = ngx.log
+local ngx_DEBUG = ngx.DEBUG
 local ngx_INFO = ngx.INFO
 local ngx_NOTICE = ngx.NOTICE
 local ngx_WARN = ngx.WARN
@@ -374,6 +375,9 @@ function _M.check_protocol_support(conf, cert, cert_key)
       https_proxy = proxy_opts.proxy_url,
       https_proxy_authorization = proxy_opts.proxy_authorization,
     })
+
+    ngx_log(ngx_DEBUG, _log_prefix,
+            "using proxy ", proxy_opts.proxy_url, " to check protocol support ")
   end
 
   local res, err = c:request_uri(
@@ -418,6 +422,9 @@ function _M.connect_cp(endpoint, conf, cert, cert_key, protocols)
       wss_proxy = proxy_opts.proxy_url,
       wss_proxy_authorization = proxy_opts.proxy_authorization,
     }
+
+    ngx_log(ngx_DEBUG, _log_prefix,
+            "using proxy ", proxy_opts.proxy_url, " to connect control plane ")
   end
 
   if conf.cluster_mtls == "shared" then
diff --git a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
index 697b4d09fe8d..6d5fa224785b 100644
--- a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
+++ b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
@@ -35,7 +35,7 @@ local confs = helpers.get_clustering_protocols()
 
 local auth_confgs = {
   ["auth off"] = "http://127.0.0.1:16797",
-  -- ["auth on"] = "http://test:konghq@127.0.0.1:16796",
+  ["auth on"] = "http://test:konghq@127.0.0.1:16796",
 }
 
 
@@ -82,12 +82,6 @@ for _, strategy in helpers.each_strategy() do
       end)
 
       lazy_teardown(function()
-
-        os.execute("echo @@@@@@@@@@@@@@; cat servroot2/logs/proxy_auth.log")
-        os.execute("echo @@@@@@@@@@@@@@; cat servroot2/logs/proxy.log")
-
-        os.execute("echo @@@@@@@@@@@@@@; cat servroot2/logs/error.log")
-
         helpers.stop_kong("servroot2")
         helpers.stop_kong()
       end)
@@ -126,9 +120,6 @@ for _, strategy in helpers.each_strategy() do
             end
           end, 10)
 
-          os.execute("echo @@@@@@@@@@@@@@; cat servroot2/logs/proxy_auth.log")
-          os.execute("echo @@@@@@@@@@@@@@; cat servroot2/logs/proxy.log")
-
           -- ensure this goes through proxy
           local path = pl_path.join("servroot2", "logs",
                       (auth_desc == "auth on") and "proxy_auth.log" or "proxy.log")
@@ -136,7 +127,7 @@ for _, strategy in helpers.each_strategy() do
           assert.matches("CONNECT 127.0.0.1:9005", contents)
 
           if auth_desc == "auth on" then
-            assert.matches("accepted basic proxy-authorization", contents)
+            assert.matches("accepted basic proxy%-authorization", contents)
           end
         end)
       end)

From 5b216cba8cf88235fe01b41ef4ea35d105e0a601 Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Wed, 16 Nov 2022 00:50:37 -0800
Subject: [PATCH 15/25] Revert "run less"

This reverts commit 5b30332860ffb0dc76e0183b96f1951c423e7865.
---
 .github/workflows/build_and_test.yml | 7 ++-----
 1 file changed, 2 insertions(+), 5 deletions(-)

diff --git a/.github/workflows/build_and_test.yml b/.github/workflows/build_and_test.yml
index ba69d2d1477a..63a6bc9807ad 100644
--- a/.github/workflows/build_and_test.yml
+++ b/.github/workflows/build_and_test.yml
@@ -145,8 +145,8 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        suite: [integration]
-        split: [first (01-04)]
+        suite: [integration, plugins]
+        split: [first (01-04), second (>= 05)]
 
     env:
       KONG_TEST_PG_DATABASE: kong
@@ -246,7 +246,6 @@ jobs:
     name: DB-less integration tests
     runs-on: ubuntu-22.04
     needs: build
-    if: false
 
     env:
       KONG_TEST_PG_DATABASE: kong
@@ -313,7 +312,6 @@ jobs:
     name: C* ${{ matrix.cassandra_version }} ${{ matrix.suite }} - ${{ matrix.split }} tests
     runs-on: ubuntu-22.04
     needs: build
-    if: false
 
     strategy:
       fail-fast: false
@@ -417,7 +415,6 @@ jobs:
     name: PDK tests
     runs-on: ubuntu-22.04
     needs: build
-    if: false
 
     env:
       TEST_SUITE: pdk

From 96442e9dc8b68bad5cff529c6d34a0b82ce5884c Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Wed, 16 Nov 2022 00:51:07 -0800
Subject: [PATCH 16/25] Revert "run less tests"

This reverts commit 6ad4fe5d49afcba88f08d6c5628a11a424bc1f3c.
---
 .ci/run_tests.sh | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.ci/run_tests.sh b/.ci/run_tests.sh
index ebb618b15acf..865e5bfd6371 100755
--- a/.ci/run_tests.sh
+++ b/.ci/run_tests.sh
@@ -33,7 +33,7 @@ fi
 if [ "$TEST_SUITE" == "integration" ]; then
     if [[ "$TEST_SPLIT" == first* ]]; then
         # GitHub Actions, run first batch of integration tests
-        eval "$TEST_CMD" spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
+        eval "$TEST_CMD" $(ls -d spec/02-integration/* | sort | grep -v 05-proxy)
 
     elif [[ "$TEST_SPLIT" == second* ]]; then
         # GitHub Actions, run second batch of integration tests

From 19bab949a56d12cf05f304186c57b47ac6a75deb Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Wed, 16 Nov 2022 01:00:14 -0800
Subject: [PATCH 17/25] avoid introducing socket.url dep

---
 kong/clustering/utils.lua |  4 +--
 kong/conf_loader/init.lua |  3 +-
 kong/tools/utils.lua      | 71 +++++++++++++++++++++++++++++++++++++++
 3 files changed, 74 insertions(+), 4 deletions(-)

diff --git a/kong/clustering/utils.lua b/kong/clustering/utils.lua
index 2586deca0f19..935f1f6c12a0 100644
--- a/kong/clustering/utils.lua
+++ b/kong/clustering/utils.lua
@@ -7,7 +7,7 @@ local ocsp = require("ngx.ocsp")
 local http = require("resty.http")
 local ws_client = require("kong.resty.websocket.client")
 local ws_server = require("resty.websocket.server")
-local socket_url = require("socket.url")
+local parse_url = require("kong.tools.utils").parse_url
 
 local type = type
 local tonumber = tonumber
@@ -331,7 +331,7 @@ local function parse_proxy_url(conf)
   local proxy_server = conf.proxy_server
   if proxy_server then
     -- assume proxy_server is validated in conf_loader
-    local parsed = socket_url.parse(proxy_server)
+    local parsed = parse_url(proxy_server)
     ret.proxy_url = fmt("%s://%s:%s", parsed.scheme, parsed.host, parsed.port or 80)
     ret.scheme = parsed.scheme
     ret.host = parsed.host
diff --git a/kong/conf_loader/init.lua b/kong/conf_loader/init.lua
index 4958d822b9cc..98285ed194b4 100644
--- a/kong/conf_loader/init.lua
+++ b/kong/conf_loader/init.lua
@@ -7,7 +7,6 @@ local openssl_pkey = require "resty.openssl.pkey"
 local openssl_x509 = require "resty.openssl.x509"
 local pl_stringio = require "pl.stringio"
 local pl_stringx = require "pl.stringx"
-local socket_url = require "socket.url"
 local constants = require "kong.constants"
 local listeners = require "kong.conf_loader.listeners"
 local pl_pretty = require "pl.pretty"
@@ -988,7 +987,7 @@ local function check_and_infer(conf, opts)
   end
 
   if conf.proxy_server then
-    local parsed, err = socket_url.parse(conf.proxy_server)
+    local parsed, err = utils.parse_url(conf.proxy_server)
     if err then
       errors[#errors + 1] = "proxy_server is invalid: " .. err
 
diff --git a/kong/tools/utils.lua b/kong/tools/utils.lua
index 39f74f31b85b..8409b535a02e 100644
--- a/kong/tools/utils.lua
+++ b/kong/tools/utils.lua
@@ -1566,4 +1566,75 @@ _M.sha256_hex       = sha256_hex
 _M.sha256_base64    = sha256_base64
 _M.sha256_base64url = sha256_base64url
 
+
+-----------------------------------------------------------------------------
+-- From https://github.com/lunarmodules/luasocket/blob/master/src/url.lua
+-- Parses a url and returns a table with all its parts according to RFC 2396
+-- The following grammar describes the names given to the URL parts
+-- <url> ::= <scheme>://<authority>/<path>;<params>?<query>#<fragment>
+-- <authority> ::= <userinfo>@<host>:<port>
+-- <userinfo> ::= <user>[:<password>]
+-- <path> :: = {<segment>/}<segment>
+-- Input
+--   url: uniform resource locator of request
+--   default: table with default values for each field
+-- Returns
+--   table with the following fields, where RFC naming conventions have
+--   been preserved:
+--     scheme, authority, userinfo, user, password, host, port,
+--     path, params, query, fragment
+-- Obs:
+--   the leading '/' in {/<path>} is considered part of <path>
+-----------------------------------------------------------------------------
+_M.parse_url = function(url, default)
+  -- initialize default parameters
+  local parsed = {}
+  for i,v in pairs(default or parsed) do parsed[i] = v end
+  -- empty url is parsed to nil
+  if not url or url == "" then return nil, "invalid url" end
+  -- remove whitespace
+  -- url = string.gsub(url, "%s", "")
+  -- get scheme
+  url = string.gsub(url, "^([%w][%w%+%-%.]*)%:",
+      function(s) parsed.scheme = s; return "" end)
+  -- get authority
+  url = string.gsub(url, "^//([^/]*)", function(n)
+      parsed.authority = n
+      return ""
+  end)
+  -- get fragment
+  url = string.gsub(url, "#(.*)$", function(f)
+      parsed.fragment = f
+      return ""
+  end)
+  -- get query string
+  url = string.gsub(url, "%?(.*)", function(q)
+      parsed.query = q
+      return ""
+  end)
+  -- get params
+  url = string.gsub(url, "%;(.*)", function(p)
+      parsed.params = p
+      return ""
+  end)
+  -- path is whatever was left
+  if url ~= "" then parsed.path = url end
+  local authority = parsed.authority
+  if not authority then return parsed end
+  authority = string.gsub(authority,"^([^@]*)@",
+    function(u) parsed.userinfo = u; return "" end)
+  authority = string.gsub(authority, ":([^:%]]*)$",
+    function(p) parsed.port = p; return "" end)
+  if authority ~= "" then
+    -- IPv6?
+    parsed.host = string.match(authority, "^%[(.+)%]$") or authority
+  end
+  local userinfo = parsed.userinfo
+  if not userinfo then return parsed end
+  userinfo = string.gsub(userinfo, ":([^:]*)$",
+    function(p) parsed.password = p; return "" end)
+  parsed.user = userinfo
+  return parsed
+end
+
 return _M

From 2263d0034561e04faaaa74dec8aad165a0579c45 Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Wed, 16 Nov 2022 01:03:43 -0800
Subject: [PATCH 18/25] add changelog

---
 CHANGELOG.md | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 97485cece8b1..20fc9ab79b37 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -159,6 +159,10 @@
   worker.
   [#9616](https://github.com/Kong/kong/pull/9616)
 
+- Add forward proxy support for Hybrid Mode connectiones. New configuration
+  options `cluster_use_proxy` and `proxy_server` is added.
+  [#9758](https://github.com/Kong/kong/pull/9758)
+
 #### CLI
 
 - Fix slow CLI performance due to pending timer jobs

From 96dde5766e885d348f7ed9d7a7a092dd4cf2fd5e Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Wed, 16 Nov 2022 01:37:47 -0800
Subject: [PATCH 19/25] cover protocol negotiaion

---
 spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
index 6d5fa224785b..b43e267e00e4 100644
--- a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
+++ b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
@@ -48,7 +48,7 @@ for _, strategy in helpers.each_strategy() do
 
         assert(helpers.start_kong({
           role = "control_plane",
-          --legacy_hybrid_protocol = (cluster_protocol == "json (by switch)"),
+          legacy_hybrid_protocol = (cluster_protocol == "json (by switch)"),
           cluster_cert = "spec/fixtures/kong_clustering.crt",
           cluster_cert_key = "spec/fixtures/kong_clustering.key",
           database = strategy,
@@ -59,7 +59,7 @@ for _, strategy in helpers.each_strategy() do
 
         assert(helpers.start_kong({
           role = "data_plane",
-          --legacy_hybrid_protocol = (cluster_protocol == "json (by switch)"),
+          legacy_hybrid_protocol = (cluster_protocol == "json (by switch)"),
           cluster_protocol = cluster_protocol,
           database = "off",
           prefix = "servroot2",

From f636b0bd226100a7081c3352591699e35d1b41af Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Wed, 16 Nov 2022 03:55:15 -0800
Subject: [PATCH 20/25] style

---
 kong/clustering/utils.lua |  2 +-
 kong/tools/utils.lua      | 19 ++++++++++---------
 2 files changed, 11 insertions(+), 10 deletions(-)

diff --git a/kong/clustering/utils.lua b/kong/clustering/utils.lua
index 935f1f6c12a0..a8482047e42a 100644
--- a/kong/clustering/utils.lua
+++ b/kong/clustering/utils.lua
@@ -424,7 +424,7 @@ function _M.connect_cp(endpoint, conf, cert, cert_key, protocols)
     }
 
     ngx_log(ngx_DEBUG, _log_prefix,
-            "using proxy ", proxy_opts.proxy_url, " to connect control plane ")
+            "using proxy ", proxy_opts.proxy_url, " to connect control plane")
   end
 
   if conf.cluster_mtls == "shared" then
diff --git a/kong/tools/utils.lua b/kong/tools/utils.lua
index 8409b535a02e..98cba939cfbb 100644
--- a/kong/tools/utils.lua
+++ b/kong/tools/utils.lua
@@ -33,6 +33,7 @@ local lower         = string.lower
 local fmt           = string.format
 local find          = string.find
 local gsub          = string.gsub
+local match         = string.match
 local split         = pl_stringx.split
 local re_find       = ngx.re.find
 local re_match      = ngx.re.match
@@ -1595,25 +1596,25 @@ _M.parse_url = function(url, default)
   -- remove whitespace
   -- url = string.gsub(url, "%s", "")
   -- get scheme
-  url = string.gsub(url, "^([%w][%w%+%-%.]*)%:",
+  url = gsub(url, "^([%w][%w%+%-%.]*)%:",
       function(s) parsed.scheme = s; return "" end)
   -- get authority
-  url = string.gsub(url, "^//([^/]*)", function(n)
+  url = gsub(url, "^//([^/]*)", function(n)
       parsed.authority = n
       return ""
   end)
   -- get fragment
-  url = string.gsub(url, "#(.*)$", function(f)
+  url = gsub(url, "#(.*)$", function(f)
       parsed.fragment = f
       return ""
   end)
   -- get query string
-  url = string.gsub(url, "%?(.*)", function(q)
+  url = gsub(url, "%?(.*)", function(q)
       parsed.query = q
       return ""
   end)
   -- get params
-  url = string.gsub(url, "%;(.*)", function(p)
+  url = gsub(url, "%;(.*)", function(p)
       parsed.params = p
       return ""
   end)
@@ -1621,17 +1622,17 @@ _M.parse_url = function(url, default)
   if url ~= "" then parsed.path = url end
   local authority = parsed.authority
   if not authority then return parsed end
-  authority = string.gsub(authority,"^([^@]*)@",
+  authority = gsub(authority,"^([^@]*)@",
     function(u) parsed.userinfo = u; return "" end)
-  authority = string.gsub(authority, ":([^:%]]*)$",
+  authority = gsub(authority, ":([^:%]]*)$",
     function(p) parsed.port = p; return "" end)
   if authority ~= "" then
     -- IPv6?
-    parsed.host = string.match(authority, "^%[(.+)%]$") or authority
+    parsed.host = match(authority, "^%[(.+)%]$") or authority
   end
   local userinfo = parsed.userinfo
   if not userinfo then return parsed end
-  userinfo = string.gsub(userinfo, ":([^:]*)$",
+  userinfo = gsub(userinfo, ":([^:]*)$",
     function(p) parsed.password = p; return "" end)
   parsed.user = userinfo
   return parsed

From 587a9f9ba29b93213dd82dfe8e99c33330bbffc9 Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Wed, 16 Nov 2022 20:29:15 -0800
Subject: [PATCH 21/25] address comments

---
 CHANGELOG.md                                      |  4 ++--
 kong.conf.default                                 |  2 +-
 kong/clustering/utils.lua                         |  2 +-
 kong/conf_loader/init.lua                         |  2 +-
 spec/01-unit/03-conf_loader_spec.lua              |  5 ++++-
 .../09-hybrid_mode/09-forward-proxy_spec.lua      | 15 ++++-----------
 6 files changed, 13 insertions(+), 17 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 20fc9ab79b37..156488bf0c2e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -159,8 +159,8 @@
   worker.
   [#9616](https://github.com/Kong/kong/pull/9616)
 
-- Add forward proxy support for Hybrid Mode connectiones. New configuration
-  options `cluster_use_proxy` and `proxy_server` is added.
+- Add HTTP CONNECT forward proxy support for Hybrid Mode connections. New configuration
+  options `cluster_use_proxy` and `proxy_server` are added.
   [#9758](https://github.com/Kong/kong/pull/9758)
 
 #### CLI
diff --git a/kong.conf.default b/kong.conf.default
index c9c6450ddfab..1b849c763942 100644
--- a/kong.conf.default
+++ b/kong.conf.default
@@ -397,7 +397,7 @@
 #cluster_use_proxy = off
                          # Whether to turn on HTTP CONNECT proxy support for
                          # hybrid mode connections. `proxy_server` will be used
-                         # if this option is turned on.
+                         # for Hybrid mode connections if this option is turned on.
 #------------------------------------------------------------------------------
 # NGINX
 #------------------------------------------------------------------------------
diff --git a/kong/clustering/utils.lua b/kong/clustering/utils.lua
index a8482047e42a..d631f8acb01e 100644
--- a/kong/clustering/utils.lua
+++ b/kong/clustering/utils.lua
@@ -332,7 +332,7 @@ local function parse_proxy_url(conf)
   if proxy_server then
     -- assume proxy_server is validated in conf_loader
     local parsed = parse_url(proxy_server)
-    ret.proxy_url = fmt("%s://%s:%s", parsed.scheme, parsed.host, parsed.port or 80)
+    ret.proxy_url = fmt("%s://%s:%s", parsed.scheme, parsed.host, parsed.port or 443)
     ret.scheme = parsed.scheme
     ret.host = parsed.host
     ret.port = parsed.port
diff --git a/kong/conf_loader/init.lua b/kong/conf_loader/init.lua
index 98285ed194b4..d721e9bcfa21 100644
--- a/kong/conf_loader/init.lua
+++ b/kong/conf_loader/init.lua
@@ -1001,7 +1001,7 @@ local function check_and_infer(conf, opts)
       errors[#errors + 1] = "proxy_server missing host"
 
     elseif parsed.fragment or parsed.query or parsed.params then
-      errors[#errors + 1] = "proxy_server does not support fragments, query strings or parameters"
+      errors[#errors + 1] = "fragments, query strings or parameters are meaningless in proxy configuration"
     end
   end
 
diff --git a/spec/01-unit/03-conf_loader_spec.lua b/spec/01-unit/03-conf_loader_spec.lua
index 642634190367..84266bd41849 100644
--- a/spec/01-unit/03-conf_loader_spec.lua
+++ b/spec/01-unit/03-conf_loader_spec.lua
@@ -969,6 +969,7 @@ describe("Configuration loader", function()
           )
           assert.matches(".ca_combined", conf.lua_ssl_trusted_certificate_combined)
         end)
+
         it("validates proxy_server", function()
           local conf, _, errors = conf_loader(nil, {
             proxy_server = "http://cool:pwd@localhost:2333",
@@ -999,9 +1000,10 @@ describe("Configuration loader", function()
           local conf, _, errors = conf_loader(nil, {
             proxy_server = "http://localhost:2333/?a=1",
           })
-          assert.contains("proxy_server does not support fragments, query strings or parameters", errors)
+          assert.contains("fragments, query strings or parameters are meaningless in proxy configuration", errors)
           assert.is_nil(conf)
         end)
+
         it("doesn't allow cluster_use_proxy on CP but allows on DP", function()
           local conf, _, errors = conf_loader(nil, {
             role = "data_plane",
@@ -1033,6 +1035,7 @@ describe("Configuration loader", function()
           assert.contains("cluster_use_proxy can not be used when role = \"control_plane\"", errors)
           assert.is_nil(conf)
         end)
+
         it("doen't overwrite lua_ssl_trusted_certificate when autoload cluster_cert or cluster_ca_cert", function()
           local conf, _, errors = conf_loader(nil, {
             role = "data_plane",
diff --git a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
index b43e267e00e4..0fb0ff5046ac 100644
--- a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
+++ b/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
@@ -1,6 +1,6 @@
 local helpers = require "spec.helpers"
-local pl_path      = require "pl.path"
-local pl_file      = require "pl.file"
+local pl_path = require "pl.path"
+local pl_file = require "pl.file"
 
 
 local fixtures = {
@@ -31,8 +31,6 @@ local fixtures = {
 }
 
 
-local confs = helpers.get_clustering_protocols()
-
 local auth_confgs = {
   ["auth off"] = "http://127.0.0.1:16797",
   ["auth on"] = "http://test:konghq@127.0.0.1:16796",
@@ -41,26 +39,22 @@ local auth_confgs = {
 
 for _, strategy in helpers.each_strategy() do
   for auth_desc, proxy_url in pairs(auth_confgs) do
-  for cluster_protocol, conf in pairs(confs) do
-    describe("CP/DP sync through proxy (" .. auth_desc .. ") works with #" .. strategy .. " backend, protocol " .. cluster_protocol, function()
+    describe("CP/DP sync through proxy (" .. auth_desc .. ") works with #" .. strategy .. " backend", function()
       lazy_setup(function()
         helpers.get_db_utils(strategy) -- runs migrations
 
         assert(helpers.start_kong({
           role = "control_plane",
-          legacy_hybrid_protocol = (cluster_protocol == "json (by switch)"),
           cluster_cert = "spec/fixtures/kong_clustering.crt",
           cluster_cert_key = "spec/fixtures/kong_clustering.key",
           database = strategy,
           db_update_frequency = 0.1,
           cluster_listen = "127.0.0.1:9005",
-          nginx_conf = conf,
+          nginx_conf = "spec/fixtures/custom_nginx.template",
         }))
 
         assert(helpers.start_kong({
           role = "data_plane",
-          legacy_hybrid_protocol = (cluster_protocol == "json (by switch)"),
-          cluster_protocol = cluster_protocol,
           database = "off",
           prefix = "servroot2",
           cluster_cert = "spec/fixtures/kong_clustering.crt",
@@ -133,6 +127,5 @@ for _, strategy in helpers.each_strategy() do
       end)
     end)
 
-  end -- cluster protocols
   end -- auth configs 
 end

From b477a1f503eea3f263ab398e23d5130fadb55d32 Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Wed, 16 Nov 2022 23:04:24 -0800
Subject: [PATCH 22/25] address comments

---
 kong.conf.default         |  2 +-
 kong/clustering/utils.lua |  4 ++-
 kong/conf_loader/init.lua |  3 +-
 kong/tools/utils.lua      | 72 ---------------------------------------
 4 files changed, 6 insertions(+), 75 deletions(-)

diff --git a/kong.conf.default b/kong.conf.default
index 1b849c763942..cdca4f01aa3d 100644
--- a/kong.conf.default
+++ b/kong.conf.default
@@ -222,7 +222,7 @@
                                  # stack traces to help improve Kong.
 
 
-#proxy_server                    # Proxy server defined as a URL. Kong will only use
+#proxy_server =                  # Proxy server defined as a URL. Kong will only use
                                  # option if any component is explictly configured
                                  # to use proxy.
 #------------------------------------------------------------------------------
diff --git a/kong/clustering/utils.lua b/kong/clustering/utils.lua
index d631f8acb01e..79b1fd8ca341 100644
--- a/kong/clustering/utils.lua
+++ b/kong/clustering/utils.lua
@@ -7,7 +7,9 @@ local ocsp = require("ngx.ocsp")
 local http = require("resty.http")
 local ws_client = require("kong.resty.websocket.client")
 local ws_server = require("resty.websocket.server")
-local parse_url = require("kong.tools.utils").parse_url
+local utils = require("kong.tools.utils")
+local meta = require("kong.meta")
+local parse_url = require "socket.url".parse
 
 local type = type
 local tonumber = tonumber
diff --git a/kong/conf_loader/init.lua b/kong/conf_loader/init.lua
index d721e9bcfa21..04ec5ed58fe8 100644
--- a/kong/conf_loader/init.lua
+++ b/kong/conf_loader/init.lua
@@ -7,6 +7,7 @@ local openssl_pkey = require "resty.openssl.pkey"
 local openssl_x509 = require "resty.openssl.x509"
 local pl_stringio = require "pl.stringio"
 local pl_stringx = require "pl.stringx"
+local socket_url = require "socket.url"
 local constants = require "kong.constants"
 local listeners = require "kong.conf_loader.listeners"
 local pl_pretty = require "pl.pretty"
@@ -987,7 +988,7 @@ local function check_and_infer(conf, opts)
   end
 
   if conf.proxy_server then
-    local parsed, err = utils.parse_url(conf.proxy_server)
+    local parsed, err = socket_url.parse(conf.proxy_server)
     if err then
       errors[#errors + 1] = "proxy_server is invalid: " .. err
 
diff --git a/kong/tools/utils.lua b/kong/tools/utils.lua
index 98cba939cfbb..39f74f31b85b 100644
--- a/kong/tools/utils.lua
+++ b/kong/tools/utils.lua
@@ -33,7 +33,6 @@ local lower         = string.lower
 local fmt           = string.format
 local find          = string.find
 local gsub          = string.gsub
-local match         = string.match
 local split         = pl_stringx.split
 local re_find       = ngx.re.find
 local re_match      = ngx.re.match
@@ -1567,75 +1566,4 @@ _M.sha256_hex       = sha256_hex
 _M.sha256_base64    = sha256_base64
 _M.sha256_base64url = sha256_base64url
 
-
------------------------------------------------------------------------------
--- From https://github.com/lunarmodules/luasocket/blob/master/src/url.lua
--- Parses a url and returns a table with all its parts according to RFC 2396
--- The following grammar describes the names given to the URL parts
--- <url> ::= <scheme>://<authority>/<path>;<params>?<query>#<fragment>
--- <authority> ::= <userinfo>@<host>:<port>
--- <userinfo> ::= <user>[:<password>]
--- <path> :: = {<segment>/}<segment>
--- Input
---   url: uniform resource locator of request
---   default: table with default values for each field
--- Returns
---   table with the following fields, where RFC naming conventions have
---   been preserved:
---     scheme, authority, userinfo, user, password, host, port,
---     path, params, query, fragment
--- Obs:
---   the leading '/' in {/<path>} is considered part of <path>
------------------------------------------------------------------------------
-_M.parse_url = function(url, default)
-  -- initialize default parameters
-  local parsed = {}
-  for i,v in pairs(default or parsed) do parsed[i] = v end
-  -- empty url is parsed to nil
-  if not url or url == "" then return nil, "invalid url" end
-  -- remove whitespace
-  -- url = string.gsub(url, "%s", "")
-  -- get scheme
-  url = gsub(url, "^([%w][%w%+%-%.]*)%:",
-      function(s) parsed.scheme = s; return "" end)
-  -- get authority
-  url = gsub(url, "^//([^/]*)", function(n)
-      parsed.authority = n
-      return ""
-  end)
-  -- get fragment
-  url = gsub(url, "#(.*)$", function(f)
-      parsed.fragment = f
-      return ""
-  end)
-  -- get query string
-  url = gsub(url, "%?(.*)", function(q)
-      parsed.query = q
-      return ""
-  end)
-  -- get params
-  url = gsub(url, "%;(.*)", function(p)
-      parsed.params = p
-      return ""
-  end)
-  -- path is whatever was left
-  if url ~= "" then parsed.path = url end
-  local authority = parsed.authority
-  if not authority then return parsed end
-  authority = gsub(authority,"^([^@]*)@",
-    function(u) parsed.userinfo = u; return "" end)
-  authority = gsub(authority, ":([^:%]]*)$",
-    function(p) parsed.port = p; return "" end)
-  if authority ~= "" then
-    -- IPv6?
-    parsed.host = match(authority, "^%[(.+)%]$") or authority
-  end
-  local userinfo = parsed.userinfo
-  if not userinfo then return parsed end
-  userinfo = gsub(userinfo, ":([^:]*)$",
-    function(p) parsed.password = p; return "" end)
-  parsed.user = userinfo
-  return parsed
-end
-
 return _M

From af05bc68a2b22a60200ce15df54011dd94c11bec Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Thu, 17 Nov 2022 15:09:02 +0800
Subject: [PATCH 23/25] Update kong.conf.default

Co-authored-by: Datong Sun <datong.sun@konghq.com>
---
 kong.conf.default | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/kong.conf.default b/kong.conf.default
index cdca4f01aa3d..4fb8c833c379 100644
--- a/kong.conf.default
+++ b/kong.conf.default
@@ -222,7 +222,7 @@
                                  # stack traces to help improve Kong.
 
 
-#proxy_server =                  # Proxy server defined as a URL. Kong will only use
+#proxy_server =                  # Proxy server defined as a URL. Kong will only use this
                                  # option if any component is explictly configured
                                  # to use proxy.
 #------------------------------------------------------------------------------

From bc4f404dbca83d6726c156d855e1a446e4a3c069 Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Thu, 17 Nov 2022 16:19:40 +0800
Subject: [PATCH 24/25] Apply suggestions from code review

Co-authored-by: Datong Sun <datong.sun@konghq.com>
---
 kong/clustering/utils.lua | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/kong/clustering/utils.lua b/kong/clustering/utils.lua
index 79b1fd8ca341..414c78f85d1c 100644
--- a/kong/clustering/utils.lua
+++ b/kong/clustering/utils.lua
@@ -9,7 +9,7 @@ local ws_client = require("kong.resty.websocket.client")
 local ws_server = require("resty.websocket.server")
 local utils = require("kong.tools.utils")
 local meta = require("kong.meta")
-local parse_url = require "socket.url".parse
+local parse_url = require("socket.url").parse
 
 local type = type
 local tonumber = tonumber
@@ -328,6 +328,7 @@ function _M.check_configuration_compatibility(obj, dp_plugin_map)
   return true, nil, CLUSTERING_SYNC_STATUS.NORMAL
 end
 
+
 local function parse_proxy_url(conf)
   local ret = {}
   local proxy_server = conf.proxy_server

From 7eddd1ad083f1160eba874a1f343d445e8e6e3fd Mon Sep 17 00:00:00 2001
From: Wangchong Zhou <wangchong@konghq.com>
Date: Thu, 17 Nov 2022 00:20:28 -0800
Subject: [PATCH 25/25] rename

---
 .../{09-forward-proxy_spec.lua => 10-forward-proxy_spec.lua}      | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 rename spec/02-integration/09-hybrid_mode/{09-forward-proxy_spec.lua => 10-forward-proxy_spec.lua} (100%)

diff --git a/spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua b/spec/02-integration/09-hybrid_mode/10-forward-proxy_spec.lua
similarity index 100%
rename from spec/02-integration/09-hybrid_mode/09-forward-proxy_spec.lua
rename to spec/02-integration/09-hybrid_mode/10-forward-proxy_spec.lua