42 干货系列从零用Rust编写负载均衡及代理，wmproxy中配置tcp转websocket

合集 - wmproxy(44)

1.用Rust手把手编写一个Proxy(代理), 动工2023-09-192.用Rust手把手编写一个Proxy(代理), UDP绑定篇2023-09-213.5. 用Rust手把手编写一个Proxy(代理), 通讯协议建立, 为内网穿透做准备2023-09-284.6. 用Rust手把手编写一个wmproxy(代理，内网穿透等), 通讯协议源码解读篇2023-09-305.7. 用Rust手把手编写一个wmproxy(代理，内网穿透等), HTTP及TCP内网穿透原理及运行篇2023-10-046.8. 用Rust手把手编写一个wmproxy(代理，内网穿透等), HTTP改造篇之HPACK原理2023-10-087. 用Rust手把手编写一个Proxy(代理), 准备篇, 动手造轮子2023-09-168.9. 用Rust手把手编写一个wmproxy(代理，内网穿透等), HTTP2改造篇之HPACK示例, 了解http2头信息如何处理2023-10-099.10. 用Rust手把手编写一个wmproxy(代理，内网穿透等), HTTP内网穿透支持修改头信息2023-10-1010.11. 用Rust手把手编写一个wmproxy(代理，内网穿透等), 实现健康检查2023-10-1211.12. 用Rust手把手编写一个wmproxy(代理，内网穿透等), TLS的双向认证信息及token验证2023-10-1412.13. 从零开始编写一个类nginx工具, HTTP中的压缩gzip,deflate,brotli算法2023-10-1713.14. 从零开始编写一个类nginx工具, HTTP文件服务器的实现过程及参数2023-10-1914.15. 从零开始编写一个类nginx工具, 如果将nginx.conf转成yaml,toml,json会怎么样2023-10-2015.16. 从零开始编写一个类nginx工具, 反向代理upstream源码实现2023-10-2316.17. 从零开始编写一个类nginx工具, Rust中一些功能的实现2023-10-2417.18. 从零开始编写一个类nginx工具, 主动式健康检查源码实现2023-10-2618.19. 从零开始编写一个类nginx工具, 配置数据的热更新原理及实现2023-10-2719.25. 干货系列从零用Rust编写正反向代理，序列化之serde是如何工作的2023-11-1020.27. 干货系列从零用Rust编写正反向代理，Rust中日志库的应用基础准备2023-11-1521.28. 干货系列从零用Rust编写正反向代理，项目日志的源码实现2023-11-1722.30. 干货系列从零用Rust编写正反向代理，HTTP的组装之旅（中间件）2023-11-2423.29. 干货系列从零用Rust编写正反向代理，异步回调(async trait)的使用2023-11-2124.26. 干货系列从零用Rust编写正反向代理，如何发布Rust项目到Docker2023-11-1325.32. 干货系列从零用Rust编写正反向代理，关于堆和栈以及如何解决stack overflow2023-12-0426.33. 干货系列从零用Rust编写正反向代理，关于HTTP客户端代理的源码实现2023-12-1227.34. 干货系列从零用Rust编写负载均衡及代理，异步测试在Rust中的实现2023-12-1628.35. 干货系列从零用Rust编写负载均衡及代理，代理服务器的源码升级改造2023-12-1929.36. 干货系列从零用Rust编写负载均衡及代理，内网穿透中内网代理的实现2023-12-2230.37. 干货系列从零用Rust编写负载均衡及代理，负载均衡中try_files实现2023-12-2731.38. 干货系列从零用Rust编写负载均衡及代理，负载均衡中ip通行与禁止01-0332.39. 干货系列从零用Rust编写负载均衡及代理，正则及格式替换01-0933.40. 干货系列从零用Rust编写负载均衡及代理，websocket的实现01-1134.41. 干货系列从零用Rust编写负载均衡及代理，websocket与tcp的映射，WS与TCP互转01-16

35.42 干货系列从零用Rust编写负载均衡及代理，wmproxy中配置tcp转websocket01-18

36.43 干货系列从零用Rust编写负载均衡及代理，内网穿透方案完整部署01-2237.44从零开始用Rust编写nginx，命令行参数的设计与解析及说明01-2438.45从零开始用Rust编写nginx，静态文件服务器竟然还有这些细节01-2639.46从零开始用Rust编写nginx，数据还能这么传，多层代理（IP多级代理）搭建01-3040.47从零开始用Rust编写nginx，配对还有这么多要求！负载均衡中的路径匹配02-0141.48从零开始用Rust编写nginx，搭建一个简单又好看官方网站02-0642.49从零开始用Rust编写nginx，我竟然在同一个端口上绑定了多少IP02-1943.50从零开始用Rust编写nginx，原来TLS证书还可以这么申请02-2644.51从零开始用Rust编写nginx，江湖救急，TLS证书快过期了02-29

收起

wmproxy

wmproxy已用Rust实现http/https代理, socks5代理, 反向代理, 静态文件服务器，四层TCP/UDP转发，七层负载均衡，内网穿透，后续将实现websocket代理等，会将实现过程分享出来，感兴趣的可以一起造个轮子

项目地址

国内: https://gitee.com/tickbh/wmproxy

github: https://github.com/tickbh/wmproxy

设计目标

通过简单配置方便用户快速使用tcp转websocket及websocket转tcp，也可支持http升级到websocket协议。

改造http升级websocket

因为负载均衡的不确定性，在未读取数据前，未能确定当前的处理逻辑

• /root/proxy.png 访问当前的文件服务器
• /api/up 通过负载均衡访问后端服务器
• /ws 将连接升级成websocket
• 其它情况
  所以我们得预备能支持websocket的可能，那我们将同时设置回调HTTP及websocket，源码在reverse/http.rs

let timeout = oper.servers[0].comm.build_client_timeout();
let mut server = Server::builder()
    .addr(addr)
    .timeout_layer(timeout)
    .stream(inbound);
 
// 设置HTTP回调
server.set_callback_http(Box::new(Operate { inner: oper }));
// 设置websocket回调,客户端有可能升级到websocket协议
server.set_callback_ws(Box::new(ServerWsOperate::new(servers)));
if let Err(e) = server.incoming().await {
    if server.get_req_num() == 0 {
        log::info!("反向代理：未处理任何请求时发生错误：{:?}", e);
    } else {
        if !e.is_io() {
            log::info!("反向代理：处理信息时发生错误：{:?}", e);
        }
    }
}

在ServerWsOperate中定义了服务的内部信息，及向远程websocket发送的sender，以做绑定

pub struct ServerWsOperate {
    inner: InnerWsOper,
    sender: Option>,
}

在on_open的时候建立和远程websocket的双向绑定：

 
#[async_trait]
impl WsTrait for ServerWsOperate {
    /// 握手完成后之后的回调,服务端返回了Response之后就认为握手成功
    async fn on_open(&mut self, shake: WsHandshake) -> ProtResult<Option> {
        if shake.request.is_none() {
            return Err(ProtError::Extension("miss request"));
        }
        let mut option = WsOption::new();
        if let Some(location) =
            ReverseHelper::get_location_by_req(&self.inner.servers, shake.request.as_ref().unwrap())
        {
            if !location.is_ws {
                return Err(ProtError::Extension("Not Support Ws"));
            }
            if let Ok((url, domain)) = location.get_reverse_url() {
                println!("connect url = {}, domain = {:?}", url, domain);
                let mut client = Client::builder()
                    .url(url)?
                    .connect_with_domain(&domain)
                    .await?;
 
                let (serv_sender, serv_receiver) = channel::(10);
                let (cli_sender, cli_receiver) = channel::(10);
                option.set_receiver(serv_receiver);
                self.sender = Some(cli_sender);
 
                client.set_callback_ws(Box::new(ClientWsOperate {
                    sender: Some(serv_sender),
                    receiver: Some(cli_receiver),
                }));
 
                tokio::spawn(async move {
                    if let Err(e) = client
                        .wait_ws_operate_with_req(shake.request.unwrap())
                        .await
                    {
                        println!("error = {:?}", e);
                    };
                    println!("client close!!!!!!!!!!");
                });
            }
            return Ok(Some(option));
        }
        return Err(ProtError::Extension("miss match"));
    }
}

在此地方，我们是用负载均衡来做配置location.get_reverse_url()，远程端的域名和ip要映射成本地的ip，所以这边可能要读取负载的ip而不是从dns中解析ip。

获取正确的连接域名和ip地址。

pub fn get_reverse_url(&self) -> ProtResult<(Url, String)> {
    if let Some(addr) = self.get_upstream_addr() {
        if let Some(r) = &self.comm.proxy_url {
            let mut url = r.clone();
            let domain = url.domain.clone().unwrap_or(String::new());
            url.domain = Some(format!("{}", addr.ip()));
            url.port = Some(addr.port());
            Ok((url, domain))
        } else {
            let url = Url::parse(format!("http://{}/", addr).into_bytes())?;
            let domain = format!("{}", addr.ip());
            Ok((url, domain))
        }
    } else {
        Err(ProtError::Extension("error"))
    }
}

此处的处理方式与nginx不同，nginx是将所有升级请求的头信息全部删除，再根据配置的过行补充

Upgrade: websocket
Connection: Upgrade

所以在nginx中配置支持websocket通常如下配置，也就是通常配置的时候需要查找资料进行copy

proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";

在wmproxy中并不会对客户端的请求做特殊的处理，也就是发了升级远程的websocket服务器接受了升级，我们当前协议就会升级。所以我们在配置中加入了一个字段is_ws，如果升级成websocket但是并不支持websocket的时候直接进行报错，告知不支持协议

[[http.server.location]]
rule = "/ws"
is_ws = true
reverse_proxy = "http://ws"

如此在该问该url的时候就可以转websocket了，比如websocat ws://127.0.0.1/ws

tcp转websocket

利用上章讲述的StreamToWs，且利用stream流的转发，将转发类型配置tcp2ws非安全的ws，或者tcp2wss带tls的wss，实现源码在reverse/stream.rs

[[stream.server]]
bind_addr = "0.0.0.0:85"
proxy_url = "ws://127.0.0.1:8081/"
bind_mode = "tcp2ws"

这样子，我们就可以将本地监听的85端口的地址，流量转发成8081的websocket远程地址。如果远程端验证域名可以配置上相应的domain = "wmproxy.com"

if s.bind_mode == "tcp2ws" {
    let mut stream_to_ws = StreamToWs::new(inbound, format!("ws://{}", addr))?;
    if domain.is_some() {
        stream_to_ws.set_domain(domain.unwrap());
    }
    let _ = stream_to_ws.copy_bidirectional().await;
}

如此我们就可以轻松的获取tcp流量转websocket的能力。

websocket转tcp

利用上章讲述的WsToStream，且利用stream流的转发，将转发类型配置ws2tcp转发为tcp，实现源码在reverse/stream.rs

[[stream.server]]
bind_addr = "0.0.0.0:86"
up_name = "ws1"
proxy_url = "tcp://127.0.0.1:8082"
bind_mode = "ws2tcp"

这样子，我们就可以将本地监听的86端口websocket的地址，流量转发成8082的tcp远程地址。如果远程端验证域名可以配置上相应的domain = "wmproxy.com"

if s.bind_mode == "ws2tcp" {
    let mut ws_to_stream = WsToStream::new(inbound, addr)?;
    if domain.is_some() {
        ws_to_stream.set_domain(domain.unwrap());
    }
    let _ = ws_to_stream.copy_bidirectional().await;
}

如此我们就可以轻松的获取websocket流量转tcp的能力。

小结

利用wmproxy可以轻松的转化tcp到websocket的流量互转，配置简单。可以利用现成的websocket高速通道辅助我们的tcp程序获取更稳定的流量通道。

点击 [关注]，[在看]，[点赞] 是对作者最大的支持