Nginx Lua OpenResty 反向代理websocket并对响应内容进行过滤

71 阅读 0 评论 47 点赞

我是靠谱客的博主有魅力月饼，这篇文章主要介绍Nginx Lua OpenResty 反向代理websocket并对响应内容进行过滤，现在分享给大家，希望可以做个参考。

【需求】

目前是有一个日志系统的Websocket接口持续不断输出日志，目前用户可以调用，但是Websocket Server发送的日志数据有一些比较细致对用户无用，我们需要过滤掉这些对用户无用的信息，让包含这些信息的data不发送给用户。

【前期尝试】

首先我知道nginx可以直接反向代理 websocket。
其次我也知道可以通过openresty中的body_filter_by_lua对响应进行处理。
那么将这两者结合起来能否实现websocket的消息过滤呢？
答案是不能。或者说可以实现但是非常复杂，涉及到一系列协议，因此放弃了。

【最终方案】

经过广泛搜集资料：发现了一个日本小哥写的Gists websocketをフックする
他使用了 lua-resty-websocket 这个库，通过一个桥连接非常优雅的实现了websocket上下游通讯。
通过对小哥的方案进行修改，非常完美的达到了自己的目的。

【lua-resty-websocket库】

https://github.com/openresty/lua-resty-websocket
大家先去lua-resty-websocket官网把websocket依赖下载下来放到自己的resty目录下。

【nginx.conf】

复制代码

1
2
3
4
5
location = /ws_filter
{
	content_by_lua_file "ws_filter.lua";
}

【ws_filter.lua】

复制代码

local server = require "resty/websocket/server"
local client = require "resty/websocket/client"
local protocol = require "resty/websocket/protocol"

-- websocket server
local sock_server, err = server:new()

if not sock_server then
    ngx.log(ngx.ERR, "failed to new websocket: ", err)
    return ngx.exit(403)
end

local sock_client, err = client:new()
-- 此处是代理的目标 ws 地址，可根据自己的需求进行重写等操作。
-- local uri = "ws://" ..  ngx.var.http_host ..  ngx.var.uri ..  ngx.var.is_args ..  ngx.var.args
local uri = "ws://www.myhost.com/api/websocket?pass=pass"
local ok, err = sock_client:connect(uri)

if not ok then
    ngx.log(ngx.ERR, "failed to connect remote: ", err)
    ngx.exit(403)
end

local function ws_proxy(sock_from, sock_to, flip_masking)
    local opcode_mapper = {
        ["continuation"] = 0x0,
        ["text"] = 0x1,
        ["binary"] = 0x2,
        ["close"] = 0x8,
        ["ping"] = 0x9,
        ["pong"] = 0x9,
    }

while true do
        local data, typ, err = sock_from:recv_frame(flip_masking)

if data == nil then
            -- socket already closed
            sock_to:send_close()
            break
        else
            ngx.log(ngx.INFO, data .. " (" .. typ .. ")")

local fin = (typ ~= "continuation")

if typ == "close" then
                sock_from:send_close()
            end
            
            -- 在这里对data进行处理
            -- data = 'filterdata'

local bytes, err = sock_to:send_frame(fin, opcode_mapper[typ], data, flip_masking)

if bytes == nil then
                sock_from:send_close()
                break
            end
        end

end
end

local s2c = ngx.thread.spawn(ws_proxy, sock_client, sock_server, false)
local c2s = ngx.thread.spawn(ws_proxy, sock_server, sock_client, true)

if not ngx.thread.wait(s2c) then
    ngx.log(ngx.ERR, "s2c wait failed")
end

if not ngx.thread.wait(c2s) then
    ngx.log(ngx.ERR, "c2s wait failed")
end

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
local server = require "resty/websocket/server"
local client = require "resty/websocket/client"
local protocol = require "resty/websocket/protocol"

-- websocket server
local sock_server, err = server:new()

if not sock_server then
    ngx.log(ngx.ERR, "failed to new websocket: ", err)
    return ngx.exit(403)
end

local sock_client, err = client:new()
-- 此处是代理的目标 ws 地址，可根据自己的需求进行重写等操作。
-- local uri = "ws://" ..  ngx.var.http_host ..  ngx.var.uri ..  ngx.var.is_args ..  ngx.var.args
local uri = "ws://www.myhost.com/api/websocket?pass=pass"
local ok, err = sock_client:connect(uri)

if not ok then
    ngx.log(ngx.ERR, "failed to connect remote: ", err)
    ngx.exit(403)
end

local function ws_proxy(sock_from, sock_to, flip_masking)
    local opcode_mapper = {
        ["continuation"] = 0x0,
        ["text"] = 0x1,
        ["binary"] = 0x2,
        ["close"] = 0x8,
        ["ping"] = 0x9,
        ["pong"] = 0x9,
    }

    while true do
        local data, typ, err = sock_from:recv_frame(flip_masking)

        if data == nil then
            -- socket already closed
            sock_to:send_close()
            break
        else
            ngx.log(ngx.INFO, data .. " (" .. typ .. ")")

            local fin = (typ ~= "continuation")

            if typ == "close" then
                sock_from:send_close()
            end
            
            -- 在这里对data进行处理
            -- data = 'filterdata'

            local bytes, err = sock_to:send_frame(fin, opcode_mapper[typ], data, flip_masking)

            if bytes == nil then
                sock_from:send_close()
                break
            end
        end

    end
end

local s2c = ngx.thread.spawn(ws_proxy, sock_client, sock_server, false)
local c2s = ngx.thread.spawn(ws_proxy, sock_server, sock_client, true)

if not ngx.thread.wait(s2c) then
    ngx.log(ngx.ERR, "s2c wait failed")
end

if not ngx.thread.wait(c2s) then
    ngx.log(ngx.ERR, "c2s wait failed")
end