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

## 项目地址

[国内: https://gitee.com/tickbh/wmproxy](https://gitee.com/tickbh/wmproxy)

[github: https://github.com/tickbh/wmproxy](https://github.com/tickbh/wmproxy)

### 项目设计目标
针对有一些应用场景需要将TCP转成websocket的，就比如旧的客户端或者旧的服务端比较不合适进行改造，但是又需要借助阿里的全站加速DCDN等这类服务或者其它可能需要特定浏览器协议的情况下，需要进行协议的转化而服务。

### Tcp转Websocket
##### 流程图
> 以下展示Tcp转Websocket的流程图，就纯粹的Tcp客户端在不经过任何源码修改的情况下成功连接websocket服务端
```mermaid
flowchart TD
 A[tcp客户端] -->|连接服务| B[服务节点]
 B -->|服务转化| C[websocket客户端]
 C -->|连接服务| D[websocket服务端]
```
比较适合原生客户端，又不想引入第三方库，又能在需要的时候直接使用websocket来做配合。

##### 源码实现 
实现源码在[stream_to_ws](https://github.com/tickbh/wenmeng/blob/0.2.3/src/plugins/stream_to_ws.rs)

```RUST
/// 将tcp的流量转化成websocket的流量
pub struct StreamToWs<T: AsyncRead + AsyncWrite + Unpin> {
 url: Url,
 io: T,
}
```
* 需要传入的参数为原生的tcp，此处tcp是具备异步读异步写功能的虚拟tcp
* 传入连接websocket的url地址，可以连接到websocket的服务端地址

定义回调类：
```RUST
struct Operate {
 /// 将tcp来的数据流转发到websocket
 stream_sender: Sender<Vec<u8>>,
 /// 从websocket那接收信息
 receiver: Option<Receiver<OwnedMessage>>,
}
```
* **stream_sender**将数据进行发送到websocket中
* **receiver**从websocket中获取信息流

核心转发逻辑：
```RUST
pub async fn copy_bidirectional(self) -> ProtResult<()> {
 let (ws_sender, ws_receiver) = channel::<OwnedMessage>(10);
 let (stream_sender, stream_receiver) = channel::<Vec<u8>>(10);
 let url = self.url;
 tokio::spawn(async move {
 if let Ok(mut client) = Client::builder().url(url).unwrap().connect().await {
 client.set_callback_ws(Box::new(Operate {
 stream_sender,
 receiver: Some(ws_receiver),
 }));
 let _e = client.wait_ws_operate().await;
 }
 });
 Self::bind(self.io, ws_sender, stream_receiver).await?;
 Ok(())
}
```
创建两对发送接收对分别为`OwnedMessage`及`Vec<u8>`来进行双向绑定，并在协程中发起对websocket的连接请求。更多的逻辑请查看源码。

##### 测试demo
示例文件[`ws_stw`](https://github.com/tickbh/wenmeng/blob/0.2.3/examples/ws_stw.rs)，当下监听`8082`的流量并将其转发到`8081`的`websocket`服务上，测试借助`websocat`做测试服务端
* `cargo run --example ws_stw` 启动转发监听8082
* `websocat -s 8081` 监听8081
* `telnet 127.0.0.1 8082` 手动建立8082的端口

![图片.png](http://47.97.231.149/static/img/532b5f7e966535bd479c1ad65842659f.%C3%A5%C2%9B%C2%BE%C3%A7%C2%89%C2%87.webp)
成功测试转发

### Websocket转Tcp
##### 流程图
> 以下展示Websocket转Tcp的流程图，通常由浏览器环境中发起（因为浏览器的标准全双工就是websocket）。然后服务器这边由TCP的方案
```mermaid
flowchart TD
 A[websocket客户端] -->|连接服务| B[服务节点]
 B -->|服务转化| C[tcp客户端]
 C -->|连接服务| D[tcp服务端]
```
比较适合原生服务端，又不想引入第三方库，又能兼容TCP及websocket协议，适合在这个做个中间层。

##### 源码实现 
实现源码在[ws_to_stream](https://github.com/tickbh/wenmeng/blob/0.2.3/src/plugins/ws_to_stream.rs)

```RUST
/// 将websocket的流量转化成的tcp流量
pub struct WsToStream<T: AsyncRead + AsyncWrite + Unpin + Send + 'static, A: ToSocketAddrs> {
 addr: A,
 io: T,
}
```
* 需要传入的参数为原生的tcp，此处tcp是具备异步读异步写功能的虚拟tcp，其中`'static`表示io为一个类，而不是引用
* 传入连接tcp的SocketAddr地址，可以连接到Tcp的服务端地址

定义回调类：
```RUST
struct Operate {
 /// 将tcp来的数据流转发到websocket
 stream_sender: Sender<Vec<u8>>,
 /// 从websocket那接收信息
 receiver: Option<Receiver<OwnedMessage>>,
}
```
* **stream_sender**将数据进行发送到websocket中
* **receiver**从websocket中获取信息流

核心转发逻辑：
```RUST
pub async fn copy_bidirectional(self) -> ProtResult<()> {
 let (ws_sender, ws_receiver) = channel(10);
 let (stream_sender, stream_receiver) = channel::<Vec<u8>>(10);
 let stream = TcpStream::connect(self.addr).await?;
 let io = self.io;
 tokio::spawn(async move {
 let mut server = Server::new(io, None);
 server.set_callback_ws(Box::new(Operate {
 stream_sender,
 receiver: Some(ws_receiver),
 }));
 let e = server.incoming().await;
 println!("close server ==== addr = {:?} e = {:?}", 0, e);
 });
 Self::bind(stream, ws_sender, stream_receiver).await?;
 Ok(())
}
```
与tcp转websocket类似，但是此时是将io流量转成`Server`的处理函数。

##### 测试demo
示例文件[`ws_wts`](https://github.com/tickbh/wenmeng/blob/0.2.3/examples/ws_wts.rs)，当下监听`8082`的流量并将其转发到`8081`的`websocket`服务上，测试借助`websocat`做测试服务端
新建测试TCP的监听，原样转发的测试代码：
```RUST
#[tokio::main]
async fn main() -> std::io::Result<()> {
 use tokio::{net::TcpListener, io::{AsyncReadExt, AsyncWriteExt}};
 let tcp_listener = TcpListener::bind(format!("127.0.0.1:{}", 8082)).await?;
 loop {
 let mut stream = tcp_listener.accept().await?;
 tokio::spawn(async move {
 let mut buf = vec![0;20480];
 loop {
 if let Ok(size) = stream.0.read(&mut buf).await {
 println!("receiver = {:?} size = {:?}", &buf[..size], size);
 let _ = stream.0.write_all(b"from tcp:").await;
 let _ = stream.0.write_all(&buf[..size]).await;
 } else {
 break;
 }
 }
 });
 }
}
```
* `cargo run --example tcp` 监听8082的端口，收到数据原样转发
* `cargo run --example ws_wts` 启动转发监听8081
* `websocat ws://127.0.0.1:8081` 用websocket的方式连接到8081

![图片.png](http://47.97.231.149/static/img/f50810857b0ee99f0f377bf019f69a78.%C3%A5%C2%9B%C2%BE%C3%A7%C2%89%C2%87.webp)
成功测试转发

### 组合方案
当我们现存的网络方案为`Tcp到Tcp`或者为`Websocket到Websocket`而我们在中间的传输过程中如想利用DCDN做源地址保护，而他只支持Websocket，此时我们就可以利用数据的转化，将我们的数据包通过DCDN做转发：
```mermaid
flowchart TD
 A[TCP客户端] -->|连接服务| B[服务节点]
 B -->|转化成websocket通过加速| C[DCDN全站加速]
 C -->|连接服务| E[服务节点]
 E -->|转化成Tcp并串连到服务端| F[TCP服务端]
```
这样子我们就可以利用基础网络中的CDN或者DCDN等服务，又不用对旧的数据进行修改或者无法修改的程序就比如远程服务通过CDN进行加速等。

### 小结
协议的自由转化可以帮助我们创建更合适的网络环境，可以让运维更自由的构建系统。利用转化可以用好全站加速`DCDN`这类的功能，可以更好的保护源站，防止被DDOS攻击。

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

41. 干货系列从零用Rust编写负载均衡及代理，websocket与tcp的映射，WS与TCP互转

# wmproxy
`wmproxy`已用`Rust`实现`http/https`代理, `socks5`代理, 反向代理, 静态文件服务器，四层TCP/UDP转发，七层负载均衡，内网穿透，后续将实现`websocket`代理等，会将实现过程分享出来，感兴趣的可以一起造个轮子

## 项目地址

[国内: https://gitee.com/tickbh/wmproxy](https://gitee.com/tickbh/wmproxy)

[github: https://github.com/tickbh/wmproxy](https://github.com/tickbh/wmproxy)

### 简单介绍websocket
`WebSocket` 是一种在单个 TCP 连接上进行全双工通信的协议，它使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。WebSocket 通信协议于 2011 年被 IETF 定为标准 [RFC 6455](https://datatracker.ietf.org/doc/html/rfc6455)，并由 [RFC7936](https://datatracker.ietf.org/doc/html/rfc7936) 补充规范。WebSocket API 也被 W3C 定为标准。

也就是在web环境中，websocket就是socket的一种标准形式的体现。类似的还要SSE基于HTTP中的`text/event-stream`

### 源码文件含义
协议层的编码解码主要在[webparse/ws](https://github.com/tickbh/webparse/tree/0.2.3/src/ws)

* **frame_header**协议头的解码与编码
* **dataframe** 基础单位为帧，存在多帧组成一个数据包的情况
* **message** 协议的基本信息，包含`Text`必须为UTF-8字符串文本，`Binary`二进制数据流，`Close`关闭信息，`Ping`，`Pong`用来做心跳包相关的信息。
* **mask** 是否为数据进行基本的加密，服务端要求客户端传来的数据必须加密

网络处理层的源码主要在[wenmeng/ws](https://github.com/tickbh/wenmeng/tree/0.2.3/src/ws)
* **codec/framed_read** 每一帧的读，以帧为单位进行读取
* **codec/framed_write** 每一帧的写，以帧为单位进行写入
* **state/state_handshake** websocket连接内部的握手状态
* **client_connection** 客户端的状态连接
* **server_connection** 服务端的状态连接
* **control** 状态的控制，写入读取的pending等，核心处理源码
* **handshake** 定义`on_open`回调后的`WsHandshake`类
* **option** 定义`on_open`回调后返回的`WsOption`类，当下只包含定时器，即客户端多久时间唤醒一次`interval`
* **ws_trait** websocket的核心回调
```RUST
#[async_trait]
pub trait WsTrait: Send {
 /// 通过请求连接构建出返回的握手连接信息
 #[inline]
 fn on_request(&mut self, req: &RecvRequest) -> ProtResult<RecvResponse> {
 // warn!("Handler received request:\n{}", req);
 WsHandshake::build_request(req)
 }

 /// 握手完成后之后的回调,服务端返回了Response之后就认为握手成功
 fn on_open(&mut self, shake: WsHandshake) -> ProtResult<Option<WsOption>>;

 /// 接受到远端的关闭消息
 async fn on_close(&mut self, reason: &Option<CloseData>) {}

 /// 服务内部出现了错误代码
 async fn on_error(&mut self, err: ProtError) {}

 /// 收到来在远端的ping消息, 默认返回pong消息
 async fn on_ping(&mut self, val: Vec<u8>) -> ProtResult<OwnedMessage> {
 return Ok(OwnedMessage::Pong(val));
 }

 /// 收到来在远端的pong消息, 默认不做任何处理, 可自定义处理如ttl等
 async fn on_pong(&mut self, val: Vec<u8>) {}

 /// 收到来在远端的message消息, 必须覆写该函数
 async fn on_message(&mut self, msg: OwnedMessage) -> ProtResult<()>;

 /// 定时器定时按间隔时间返回
 async fn on_interval(&mut self, option: &mut Option<WsOption>) -> ProtResult<()> {
 Ok(())
 }
 
 /// 将当前trait转化成Any,仅需当需要重新获取回调处理的时候进行处理
 fn as_any(&self) -> Option<&dyn Any> {
 None
 }

 /// 将当前trait转化成mut Any,仅需当需要重新获取回调处理的时候进行处理
 fn as_any_mut(&mut self) -> Option<&mut dyn Any> {
 None
 }
}
```

### 服务端基础demo
> 建立一个本地监听8081的ws端口，完整源码[ws_server](https://github.com/tickbh/wenmeng/blob/0.2.3/examples/ws_server.rs)

建立监听类：
```RUST

struct Operate {
 sender: Option<Sender<OwnedMessage>>,
}

#[async_trait]
impl WsTrait for Operate {
 fn on_open(&mut self, shake: WsHandshake) -> ProtResult<Option<WsOption>> {
 self.sender = Some(shake.sender);
 Ok(Some(WsOption::new(Duration::from_secs(10))))
 }

 async fn on_message(&mut self, msg: OwnedMessage) -> ProtResult<()> {
 println!("callback on message = {:?}", msg);
 let _ = self
 .sender
 .as_mut()
 .unwrap()
 .send(OwnedMessage::Text("from server".to_string()))
 .await;
 let _ = self.sender.as_mut().unwrap().send(msg).await;
 Ok(())
 }

 async fn on_interval(&mut self, _option: &mut Option<WsOption>) -> ProtResult<()> {
 println!("on_interval!!!!!!!");
 Ok(())
 }
}
```
然后启动服务器监听：
```RUST
async fn run_main() -> Result<(), Box<dyn Error>> {
 let addr = "127.0.0.1:8081".to_string();
 let server = TcpListener::bind(&addr).await?;
 println!("Listening on: {}", addr);
 loop {
 let (stream, addr) = server.accept().await?;
 tokio::spawn(async move {
 let mut server = Server::new(stream, Some(addr));
 let operate = Operate { sender: None };
 // 设置服务回调
 server.set_callback_ws(Box::new(operate));
 let e = server.incoming().await;
 println!("close server ==== addr = {:?} e = {:?}", addr, e);
 });
 }
}
```
此时即可实现websocket的监听及处理。

### 客户端demo
> 当下客户端demo需要能接受终端的输入，并向服务器发送数据，所以需要自己构建sender

建立客户端连接，在这里我们手动构建了一个sender/receiver对。
```RUST
async fn run_main() -> ProtResult<()> {
 // 自己手动构建数据对,并将receiver传给服务端
 let (sender, receiver) = channel(10);
 let sender_clone = sender.clone();
 tokio::spawn(async move {
 let url = "ws://127.0.0.1:8081";
 let mut client = Client::builder()
 .url(url)
 .unwrap()
 .connect()
 .await
 .unwrap();
 client.set_callback_ws(Box::new(Operate { sender:Some(sender_clone), receiver: Some(receiver) }));
 client.wait_ws_operate().await.unwrap();
 });
 loop {
 let mut buffer = String::new();
 let stdin = io::stdin(); // We get `Stdin` here.
 stdin.read_line(&mut buffer)?;
 sender.send(OwnedMessage::Text(buffer)).await?;
 }
 Ok(())
}
```
监听实现
```RUST
struct Operate {
 sender: Option<Sender<OwnedMessage>>,
 receiver: Option<Receiver<OwnedMessage>>,
}

#[async_trait]
impl WsTrait for Operate {
 fn on_open(&mut self, shake: WsHandshake) -> ProtResult<Option<WsOption>> {
 // 将receiver传给控制中心, 以让其用该receiver做接收
 let mut option = WsOption::new(Duration::from_secs(1000));
 if self.receiver.is_some() {
 option.set_receiver(self.receiver.take().unwrap());
 }
 if self.sender.is_none() {
 self.sender = Some(shake.sender);
 }
 Ok(Some(option))
 }

 async fn on_message(&mut self, msg: OwnedMessage) -> ProtResult<()> {
 println!("callback on message = {:?}", msg);
 let _ = self
 .sender
 .as_mut()
 .unwrap()
 .send(OwnedMessage::Text("from client".to_string()))
 .await;
 let _ = self.sender.as_mut().unwrap().send(msg).await;
 Ok(())
 }

 async fn on_interval(&mut self, _option: &mut Option<WsOption>) -> ProtResult<()> {
 println!("on_interval!!!!!!!");
 Ok(())
 }
}
```

### 接口说明
`Client`和`Server`为了同时兼容HTTP服务，即握手用的为HTTP的前半段请求，选择了回调用`Box<dyn Trait>`的形式来做回调函数的处理。
```RUST
pub struct Server<T>
where
 T: AsyncRead + AsyncWrite + Unpin + Sized,
{
 /// http的接口回调, 处理http服务器
 callback_http: Option<Box<dyn HttpTrait>>,
 /// websocket的接口回调, 处理websocket服务器
 callback_ws: Option<Box<dyn WsTrait>>,
 // ...
}
```
他们两个可能是同时存在，或者单个存在的，即当作服务的时候，可能仅对`/ws`进行websocket的升级，其它的仅仅是http服务，所以需要能单独又能聚合的处理数据。而单存的websocket仅需`WsTrait`回调。

即在`pub async fn incoming(&mut self) -> ProtResult<()>`处理服务的时候不在传入回调地址，改成预先设置。达到灵活处理的目的。且接口比较清晰。

### 小结
wenmeng库当前已支持HTTP1.1/HTTP2/WEBSOCKET，在浏览器的环境中websocket是必不可缺少的存在，当然有很多原生的服务中用的都是socket，下一章中，我们将实现websocket与tcp的互转，以便一些tcp的程序可以服务web的服务。

点击 **[关注]**，**[在看]**，**[点赞]** 是对作者最大的支持

40干货系列从零用Rust编写负载均衡及代理，websocket的实现

# wmproxy
`wmproxy`已用`Rust`实现`http/https`代理, `socks5`代理, 反向代理, 静态文件服务器，四层TCP/UDP转发，七层负载均衡，内网穿透，后续将实现`websocket`代理等，会将实现过程分享出来，感兴趣的可以一起造个轮子

## 项目地址

[国内: https://gitee.com/tickbh/wmproxy](https://gitee.com/tickbh/wmproxy)

[github: https://github.com/tickbh/wmproxy](https://github.com/tickbh/wmproxy)

### 项目设计目标
利用正则替换的能力，能把指定的字符串替换成想要的字符串。

### 正则库
因为rust官方团队并未将正则正式的加入到std标准库里面，目前我们引用的是[`regex`](https://crates.io/crates/regex)也是`rust-lang`官方出品的正则库。

### 匹配的规则

1. 字符匹配：正则表达式可以匹配单个字符，如字母、数字、标点符号等。常见的字符匹配包括：
 * `\d`：匹配任意数字，等价于`[0-9]`。
 * `\D`：匹配任意非数字字符，等价于`[^0-9]`。
 * `\w`：匹配任意字母、数字或下划线字符，等价于`[A-Za-z0-9_]`。
 * `\W`：匹配任意非字母、数字或下划线字符，等价于`[^A-Za-z0-9_]`。
 * `.`：匹配除换行符（\n、\r）之外的任意字符。
2. 字符类匹配：使用字符类可以匹配指定范围内的字符。常见的字符类匹配包括：
 * `[abc]`：匹配方括号内的任意字符，例如`a`、`b`或`c`。
 * `[^abc]`：匹配除方括号内字符之外的任意字符，例如不是`a`、`b`或`c`的字符。
 * `[a-z]`：匹配任意小写字母。
 * `[A-Z]`：匹配任意大写字母。
 * `[0-9]`：匹配任意数字。
3. 量词匹配：用于指定字符或字符类出现的次数。常见的量词匹配包括：
 * `*`：匹配前一项0次或多次，等价于`{0,}`。
 * `+`：匹配前一项1次或多次，等价于`{1,}`。
 * `?`：匹配前一项0次或1次，也就是说前一项是可选的，等价于`{0,1}`。
 * `{n}`：匹配前一项恰好n次。
 * `{n,}`：匹配前一项至少n次。
 * `{n,m}`：匹配前一项至少n次，但不超过m次。
4. 边界匹配：用于匹配字符串的边界位置。常见的边界匹配包括：
 * `^`：匹配字符串的开头位置。
 * `$`：匹配字符串的结尾位置。
 * `\b`：匹配单词的边界位置，即字与空白间的位置。
 * `\B`：匹配非单词边界的位置。
5. 选择、分组和引用：
 * `|`：选择符号，匹配该符号左边或右边的表达式。
 * `(...)`：将几项组合成一个单元，这个单元可通过"*"、"+"、"?" 和"|" 等符号加以修饰，也可以记住与这个组匹配的字符以便后面引用。
 * `\n`：在正则表达式中，n 是一个正整数，引用匹配到的第n个分组。
6. 预查：预查是一种零宽断言，即匹配的是位置而不是字符。预查包括正向预查和负向预查：
 * `(?=...)`：正向肯定预查，表示要匹配的字符串后面必须紧跟着指定的模式。
 * `(?!...)`：正向否定预查，表示要匹配的字符串后面不能紧跟着指定的模式。
 * `(?<=...)`：反向肯定预查，表示要匹配的字符串前面必须紧跟着指定的模式。
 * `(?<!...)`：反向否定预查，表示要匹配的字符串前面不能紧跟着指定的模式。

### 需求功能
* 需要从`Request`中获取`Url`或者`Path`并将其中间的某一部分替换成另一部分，且需兼容部分不需要替换，两种模式需均能正常的。

* 需要配置中正确的读取分割的信息，如`"{path}/ '/ro(\\w+)/(.*) {path} /ro$1/Cargo.toml' /root/README.md"`需要正确的分割成`{path}/`, `/ro(\\w+)/(.*) {path} /ro$1/Cargo.toml`, `/root/README.md`三个部分。

### 需求实现
以下是一段try_paths的配置
```TOML
[[http.server.location]]
rate_limit = "4m/s"
rule = "/root"
file_server = { browse = true }
proxy_pass = ""
try_paths = "{path}/ '/ro(\\w+)/(.*) {path} /ro$1/Cargo.toml' /root/README.md"
```
我们需要将try_paths做正确的拆分，我们需要将一个字符串按空格做分割，且如果有单引号`'`或者双引号`"`需要找到其对应的结尾，防止将其中的字符串做切割。

我们将利用以下正则，其中小括号括起来是我们将匹配的内容，用`|`则表示并行的匹配规则，当第一个没有匹配到将匹配第二个选项。
```RUST
r#"([^\s'"]+)|"([^"]*)"|'([^']*)'"#
```
其中`([^\s'"]+)`表示非空白字符开头也非单号双引号开头，直到碰到空格或者单引号双引号停止，那我们将获取第一个匹配项`{path}/`

其中`"([^"]*)"`则表示以双引号开头，中间不能添加任何双引号的其它任意字符，直到匹配到另一个双引号停止，取中间的数据，不取双引号。`'([^']*)'`则类似双引号，那么`'/ro(\\w+)/(.*) {path} /ro$1/Cargo.toml'`将是单引号开头的匹配项直到另一个单引号截止，那么匹配的结果为`/ro(\\w+)/(.*) {path} /ro$1/Cargo.toml`

另一个匹配第一个规则`/root/README.md`，此时我们已经正确将数据进行切割，以下是源码实现，用lazy_static是为了只初始化一次，无需重复耗性能。
```RUST
pub fn split_by_whitespace<'a>(key: &'a str) -> Vec<&'a str> {
 lazy_static! {
 static ref RE: Regex = Regex::new(r#"([^\s'"]+)|"([^"]*)"|'([^']*)'"#).unwrap();
 };
 
 let mut vals = vec![];
 for (_, [value]) in RE.captures_iter(key).map(|c| c.extract()) {
 vals.push(value);
 }
 vals
}
```

> 以下是正确的替换，假设我们收到的是
`GET /root/index.html HTTP/1.1\r\n`
那么第一个参数`{path}/`将通过`Request`取得为`/root/index.html/`

那么第二个参数`/ro(\\w+)/(.*) {path} /ro$1/Cargo.toml`，按空格切割，切割结果有`/ro(\\w+)/(.*)`,`{path}`,`/ro$1/Cargo.toml`有三个参数，且第一个参数为正则，那么我们尝试将第一个参数正则化，与第二个参数相匹配，并替换成第三个参数的内容同时将第二个参数格式化为`/root/index.html`，那么与正则相匹配

| 匹配项 | 匹配结果 |
| --- | --- |
| $0 | /root/index.html 正则表达示里0均为整个字符串 |
| $1 | ot (\\w+)的内容 |
| $2 | index.html (.*)的内容 |

那么`/ro$1/Cargo.toml`替换成结果后将为`/root/Cargo.toml`通过这种方法我们可以将任意的字符串按照一定的规则匹配成另一个字符串来达到自定义的目的。

源码实现：
```RUST
pub fn format_req(req: &Request<Body>, formats: &str) -> String {
 let pw = FORMAT_PATTERN_CACHE.with(|m| {
 if !m.borrow().contains_key(&formats) {
 let p = PatternEncoder::new(formats);
 m.borrow_mut().insert(
 Box::leak(formats.to_string().clone().into_boxed_str()),
 Arc::new(p),
 );
 }
 m.borrow()[&formats].clone()
 });

 // 将其转化成Record然后进行encode
 let record = ProxyRecord::new_req(Record::builder().level(Level::Info).build(), req);
 let mut buf = vec![];
 pw.encode(&mut SimpleWriter(&mut buf), &record).unwrap();
 String::from_utf8_lossy(&buf[..]).to_string()
}

fn inner_oper_regex(req: &Request<Body>, re: &Regex, vals: &[&str]) -> String {
 let mut ret = String::new();
 let key = Self::format_req(req, vals[0]);
 for idx in 1..vals.len() {
 if idx != 1 {
 ret += " ";
 }
 let val = re.replace_all(&key, vals[idx]);
 ret += &val;
 }
 ret
}

pub fn format_req_may_regex(req: &Request<Body>, formats: &str) -> String {
 let formats = formats.trim();
 if formats.contains(char::is_whitespace) {
 // 因为均是从配置中读取的数据, 在这里缓存正则表达示会在总量上受到配置的限制
 lazy_static! {
 static ref RE_CACHES: Mutex<HashMap<&'static str, Regex>> =
 Mutex::new(HashMap::new());
 };

 if formats.len() == 0 {
 return String::new();
 }

 let vals = Self::split_by_whitespace(formats);
 if vals.len() < 2 {
 return String::new();
 }

 if let Ok(mut guard) = RE_CACHES.lock() {
 if let Some(re) = guard.get(&vals[1]) {
 return Self::inner_oper_regex(req, re, &vals[1..]);
 } else {
 if let Ok(re) = Regex::new(vals[0]) {
 let ret = Self::inner_oper_regex(req, &re, &vals[1..]);
 guard.insert(Box::leak(vals[0].to_string().into_boxed_str()), re);
 return ret;
 }
 }
 }
 }
 Self::format_req(req, formats)
}
```

### 测试用例
根据Request生成我们任意想要的内容
```rust
mod tests {
 use webparse::Request;
 use wenmeng::Body;

 use crate::Helper;

 fn build_request() -> Request<Body> {
 Request::builder()
 .url("http://127.0.0.1/test/root?query=1&a=b")
 .header("Accept", "text/html")
 .body("ok")
 .unwrap()
 .into_type()
 }

 #[test]
 fn do_test_reg() {
 let req = &build_request();
 let format = r" /test/(.*) {path} /formal/$1 ";
 let val = Helper::format_req_may_regex(req, format);
 assert_eq!(val, "/formal/root");
 
 let format = r" /te(\w+)/(.*) {path} /formal/$1/$2 ";
 let val = Helper::format_req_may_regex(req, format);
 assert_eq!(val, "/formal/st/root");

 let format = r" /te(\w+)/(.*) {url} /formal/$1/$2 ";
 let val = Helper::format_req_may_regex(req, format);
 assert_eq!(val, "http://127.0.0.1/formal/st/root?query=1&a=b");
 }
}
```

### 小结
正则在计算机的处理中是非常的常用的一种技术，具有许多优点，使得它在文本处理和模式匹配方面非常强大和灵活，有强大的文本匹配和搜索功能，跨平台性跨语言，每种语言都有相应的实现，既简洁又高效便捷，是受欢迎的一种又相处较难的字符串处理技术。

点击 **[关注]**，**[在看]**，**[点赞]** 是对作者最大的支持

39干货系列从零用Rust编写负载均衡及代理，正则及格式替换

# wmproxy
`wmproxy`已用`Rust`实现`http/https`代理, `socks5`代理, 反向代理, 静态文件服务器，四层TCP/UDP转发，七层负载均衡，内网穿透，后续将实现`websocket`代理等，会将实现过程分享出来，感兴趣的可以一起造个轮子

## 项目地址

[国内: https://gitee.com/tickbh/wmproxy](https://gitee.com/tickbh/wmproxy)

[github: https://github.com/tickbh/wmproxy](https://github.com/tickbh/wmproxy)

### 设计目标
需要能对针对性的IP地址进行放行或者禁止，从而达到网络限制或者安全等目的，保护系统的整体稳定性。

### IP的作用
IP地址的作用是标示一台在互联网上的主机，就像每个人的住宅地址一样，邮寄东西需要住宅地址，而互联网上一台电脑对另一台电脑发送数据也需要一个可以识别的地址。

早期的IP地址由32位（即IPv4）的数据表示，也就是大概有42亿的地址，以现如今的网络拥有量已经将公有IP即将耗尽的情形。即使有大量的地址并不会占用公网地址，如公司内部的电脑，家庭内部的电脑，手机等并未占用公网地址，可能很多的设备占用了同一个公网出口。所以我们做了IPv4的限定的时候，也有可能将正常的用户做了相关限定。

现在正在推行的IP地址是128位（即IPv6）的数据将可以表示很大的IP数，将可以每个物联网的设备都拥有独立的IP，但是由于IP的升级涉及到大量的基础设备，大量的旧软件，所以当下基本上两种IP都必须支持的阶段。但是国内相对IPv4占了大部分。

### IP中子网掩码
子网掩码就是为了划分同网段的主机数量。每类网段默认是254个。

我们在现实的生活中经常看到路由上`255.255.255.0/24`也经常在云的白名单上看到了`0.0.0.0/0`，那么他们表示的含义又是什么呢？

通常我们可以把IPv4看成一个无符号的32位整型，那么`255.255.255.0/24`后面的24就表示`0xffffff00`，那么我们将某个IP与该值进行`按位与`运算，得到相同的值将表示归属于同一个IP段。
举例：内网常用的ip如`192.168.0.100`按位与后将变成`192.168.0.0`，及`192.168.0.1-254`这254个ip按位与得到的地址均为
`192.168.0.0`，所以他们归属于同一个IP段，也就是表示在他们在同一个路由器下面。

而`0.0.0.0/0`中最后的`0`就是表示`0x00000000`，那么我们将任意的IP与该值做按位与得到的值均为`0.0.0.0`，则表示所有的IP都归属于同一个类，也就是通常配置的白名单对所有的都生效。

IPv6与IPv4类似，只是IPv6的长度更长，子网的长度可以由0-127，而IPv6不叫子网掩码，通常称其为前缀，但其原理都是用位表示，前n位为网络位，则说明IP只要前n位一样，则子网一样，IP的机制通了，涉及IP的问题也就好解决了。 

### 源码的实现
在`Rust`中并不支持子网掩码等，那么我们将在其基础上增加一个8位的无符号型：
```RUST
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IpGate {
 pub ip: IpAddr,
 pub gate: u8,
}
```

序列化我们通过`serde_with`中的`DisplayForStr`实现，如果存在`/`则将其切割，如果不存在那么子网掩码位数为0，兼容两种模式：
```RUST

impl FromStr for IpGate {
 type Err = io::Error;

 fn from_str(s: &str) -> Result<Self, Self::Err> {
 let vals = s.split("/").collect::<Vec<&str>>();
 let ip = vals[0].parse::<IpAddr>().map_err(|_| io::Error::new(io::ErrorKind::Other, "parse ip error"))?;
 let mut gate = 0;
 if vals.len() > 1 {
 gate = vals[1].parse::<u8>().map_err(|_| io::Error::new(io::ErrorKind::Other, "parse ip error"))?;
 if ip.is_ipv4() && gate > 32 {
 return Err(io::Error::new(io::ErrorKind::Other, "too big gate"));
 } else if ip.is_ipv6() && gate > 128 {
 return Err(io::Error::new(io::ErrorKind::Other, "too big gate"));
 }
 }
 Ok(IpGate {
 ip, gate
 })
 }
}

impl Display for IpGate {
 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
 if self.gate > 0 {
 f.write_fmt(format_args!("{}/{}", self.ip, self.gate))
 } else {
 f.write_fmt(format_args!("{}", self.ip))
 }
 }
}
```
查看是否包含：
```RUST
pub fn contains(&self, ip: &IpAddr) -> bool {
 if self.gate == 0 {
 ip == &self.ip
 } else {
 match (&ip, &self.ip) {
 (IpAddr::V4(other), IpAddr::V4(my)) => {
 let other = u32::from_be_bytes(other.octets()) >> (32u8 - self.gate);
 let my = u32::from_be_bytes(my.octets()) >> (32u8 - self.gate);
 other == my
 }
 _ => {
 ip == &self.ip
 }
 }
 }
}
```

### 在负载均衡中的通行及禁止
我们将配置信息转化成可通行的IP段数组或者禁止的IP段数组
* 如果存在可通行的配置那么必须在配置中才可通行
* 如果存在禁止的IP，那么在配置中的将会被禁止

我们在配置的时候，就可以进行如下的配置：
```TOML
[[http.server.location]]
rule = "/try"
# 只允许本地网络通行
allow_ip = "127.0.0.1 192.168.0.0/24"

[[http.server.location]]
rule = "/"
reverse_proxy = "http://server"
# 全面禁止10开头的IP段
deny_ip = "10.0.0.0/8"
```

源码示例：
```RUST

if l.comm.deny_ip.is_some() || l.comm.allow_ip.is_some() {
 if let Some(ip) = req.headers().system_get("{client_ip}") {
 let ip = ip
 .parse::<IpAddr>()
 .map_err(|_| ProtError::Extension("client ip error"))?;
 if let Some(allow) = &l.comm.allow_ip {
 if !allow.contains(&ip) {
 return Ok(Response::status503()
 .body("now allow ip")
 .unwrap()
 .into_type());
 }
 }
 if let Some(deny) = &l.comm.deny_ip {
 if deny.contains(&ip) {
 return Ok(Response::status503().body("deny ip").unwrap().into_type());
 }
 }
 }
}

```

### 小结
后续可能需要在接受连接的时候就直接禁止掉IP，那么我们可以防止客户端持续的发送流量，即可能造成流量被耗尽。

IP的通行及禁止帮我们更好的保护系统的健壮性及私域的隐私性做保证。自动禁止IP的话，将是WAF等进阶能力的，更好的保护源站。


点击 **[关注]**，**[在看]**，**[点赞]** 是对作者最大的支持

38干货系列从零用Rust编写负载均衡及代理，负载均衡中ip通行与禁止

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

## 项目地址

[国内: https://gitee.com/tickbh/wmproxy](https://gitee.com/tickbh/wmproxy)

[github: https://github.com/tickbh/wmproxy](https://github.com/tickbh/wmproxy)

### nginx中的try_files
* 语法：`try_files file … uri;`或 `try_files file … = code;` 
* 作用域：server location
*
 * 首先：按照指定的顺序检查文件是否存在，并使用第一个找到的文件进行请求处理
 * 其次：处理是在当前上下文中执行的。根据 root 和 alias 指令从 file 参数构造文件路径。
 * 然后：可以通过在名称末尾指定一个斜杠来检查目录的存在，例如`"$uri/"`。
 * 最后：如果没有找到任何文件，则进行内部重定向到最后一个参数中指定的 uri。

注：只有最后一个参数可以引起一个内部重定向，之前的参数只设置内部的 URL 的指向。最后一个参数是回退 URL 且必须存在，否则会出现内部 500 错误。命名的 location 也可以使用在最后一个参数中。

> 应用场景

1、前端路由处理：
```NGINX
location / {
 try_files $uri $uri/ /index.html;
 # $uri指请求的uri路径，$uri/表示请求的uri路径加上一个/，例如访问example.com/path，则会依次尝试访问/path,/path/index.html,/index.html
 # /index.html表示如果仍未匹配到则重定向到index.html
}
```
这种场景多用于单页应用，例如vue.js等前端框架的路由管理。当用户在浏览器中访问一个非根路径的路径时，由于这些路径都是由前端路由管理的，nginx无法直接返回正确的静态文件，因此需要将请求重定向到统一的路径，这里是/index.html，由前端路由控制页面的展示。
2、图片服务器：
```NGINX
location /images/ {
 root /data/www;
 error_page 404 = /fetch_image.php;
 try_files $uri $uri/ =404;
}
location /fetch_image.php {
 fastcgi_pass 127.0.0.1:9000;
 set $path_info "";
 fastcgi_param PATH_INFO $path_info;
 fastcgi_param SCRIPT_FILENAME /scripts/fetch_image.php;
 include fastcgi_params;
}
```
这种场景多用于图片服务器，当用户访问图片时，先尝试在本地文件系统中查找是否有该文件，如果找到就返回；如果没有找到则会转发到fetch_image.php进行处理，从远程资源服务器拉取图片并返回给用户。

### 实现方案
当前nginx方案的实现，是基于文件的重试，也就是所谓的`伪静态`，如果跨目录的服务器就很麻烦了，比如：
```NGINX
location /images/ {
 root /data/upload;
 try_files $uri $uri/ =404;
}
location /images2/ {
 root /data/www;
 try_files $uri $uri/ =404;
}
```
上面的我们无法同时索引两个目录下的结构。即我假设我访问`/images/logo.png`无法同时查找`/data/upload/logo.png`及`/data/www/logo.png`是否存在。

> 当前实现方案从`try_files`变成`try_paths`也就是当碰到该选项时，将当前的几个访问地址重新进入路由

例：
```TOML
[[http.server.location]]
rate_limit = "4m/s"
rule = "/root/logo.png"
file_server = { browse = true }
proxy_pass = ""
try_paths = "/data/upload/logo.png /data/www/logo.png /root/README.md"

[[http.server.location]]
rule = "/data/upload"
file_server = { browse = true }

[[http.server.location]]
rule = "/data/www"
file_server = { browse = true }
```

> 除非碰到返回100或者200状态码的，否则将执行到最后一个匹配路由。

### 源码实现
* 1. 要能循环遍历路由
* 2. 当try_paths时要避免递归死循环
* 3. 当try_paths时可能会调用自己本身，需要能重复调用

以下主要源码均在[`reverse/http.rs`](https://github.com/tickbh/wmproxy/blob/0.2.3/src/reverse/http.rs#L217)
* **实现循环**
主要的处理函数为`deal_match_location`，函数的参数为
```RUST
#[async_recursion]
async fn deal_match_location(
 req: &mut Request<Body>,
 // 缓存客户端请求
 cache: &mut HashMap<
 LocationConfig,
 (Sender<Request<Body>>, Receiver<ProtResult<Response<Body>>>),
 >,
 // 该Server的配置选项
 server: Arc<ServerConfig>,
 // 已处理的匹配路由
 deals: &mut HashSet<usize>,
 // 已处理的TryPath匹配路由
 try_deals: &mut HashSet<usize>,
) -> ProtResult<Response<Body>>
```
当前在Rust中的异步递归会报如下错误
```
recursion in an `async fn` requires boxing
a recursive `async fn` must be rewritten to return a boxed `dyn Future`
consider using the `async_recursion` crate: https://crates.io/crates/async_recursion
```
所以需要添加`#[async_recursion]`或者改成Box返回。

参数其中多定义了两组`HashSet`用来存储已处理的路由及已处理的`TryPath`路由。

将循环获取合适的location，如果未找到直接返回503错误。
```RUST
let path = req.path().clone();
let mut l = None;
let mut now = usize::MAX;
for idx in 0..server.location.len() {
 if deals.contains(&idx) {
 continue;
 }
 if server.location[idx].is_match_rule(&path, req.method()) {
 l = Some(&server.location[idx]);
 now = idx;
 break;
 }
}
if l.is_none() {
 return Ok(Response::status503()
 .body("unknow location to deal")
 .unwrap()
 .into_type());
}
```

当该路由存在`try_paths`的情况时：
```RUST
// 判定该try是否处理过, 防止死循环
if !try_deals.contains(&now) && l.try_paths.is_some() {
 let try_paths = l.try_paths.as_ref().unwrap();
 try_deals.insert(now);
 let ori_path = req.path().clone();
 for val in try_paths.list.iter() {
 deals.clear();
 // 重写path好方便做数据格式化
 req.set_path(ori_path.clone());
 let new_path = Helper::format_req(req, &**val);
 // 重写path好方便后续处理无感
 req.set_path(new_path);
 if let Ok(res) = Self::deal_match_location(
 req,
 cache,
 server.clone(),
 deals,
 try_deals,
 )
 .await
 {
 if !res.status().is_client_error() && !res.status().is_server_error() {
 return Ok(res);
 }
 }
 }
 return Ok(Response::builder()
 .status(try_paths.fail_status)
 .body("not valid to try")
 .unwrap()
 .into_type());
}
```
其中会将`req`中的`path`进行格式化的重写以方便处理：
```rust
// 重写path好方便做数据格式化
req.set_path(ori_path.clone());
let new_path = Helper::format_req(req, &**val);
// 重写path好方便后续处理无感
req.set_path(new_path);
```

如果不存在`try_paths`将正常的按照路由的处理逻辑，该文件服务器或者反向代理，并标记该路由已处理。
```RUST
deals.insert(now);
let clone = l.clone_only_hash();
if cache.contains_key(&clone) {
 let mut cache_client = cache.remove(&clone).unwrap();
 if !cache_client.0.is_closed() {
 println!("do req data by cache");
 let _send = cache_client.0.send(req.replace_clone(Body::empty())).await;
 match cache_client.1.recv().await {
 Some(res) => {
 if res.is_ok() {
 log::trace!("cache client receive response");
 cache.insert(clone, cache_client);
 }
 return res;
 }
 None => {
 log::trace!("cache client close response");
 return Ok(Response::status503()
 .body("already lose connection")
 .unwrap()
 .into_type());
 }
 }
 }
} else {
 log::trace!("do req data by new");
 let (res, sender, receiver) = l.deal_request(req).await?;
 if sender.is_some() && receiver.is_some() {
 cache.insert(clone, (sender.unwrap(), receiver.unwrap()));
 }
 return Ok(res);
}
```

### 小结
`try_files`在nginx中提供了更多的可能，也方便了伪静态文件服务器的处理。我们在其中的基础上稍微改造成`try_paths`来适应处理提供多路由映射的可能性。


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

37干货系列从零用Rust编写负载均衡及代理，负载均衡中try_files实现

用Rust实现http/https代理, socks5代理, 负载均衡, 反向代理, 静态文件服务器，四层TCP/UDP转发，websocket转发, 内网穿透nat 