wmproxy

wmproxy是由Rust编写，已实现http/https代理，socks5代理，反向代理，静态文件服务器，内网穿透，配置热更新等，后续将实现websocket代理等，同时会将实现过程分享出来，感兴趣的可以一起造个轮子法

项目 ++wmproxy++

gite: https://gitee.com/tickbh/wmproxy

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

修改header参数

但凡代理之类，基本上都有修改头参数的需求，就比如要获取客户端的真实IP，需要写入x-forward-for表示客户端的真实IP，要不然经过转发后的HTTP无法获取真实的客户端地址。

所以需要在转发的同时能进行处理头部信息的相关参数。故内网端不能仅做流量转发。而且客户端可能直接以纯HTTP2的协议请求内网的数据，所以同时需要支持HTTP/1.1及HTTP2，由于以上需求，我们把之前的简单的转发逻辑改成以服务端接收客户端请求的模式对数据进行重加工。

新流程如下

以下是数据从外网进入到内网服务器的加工流程

graph TD
A[外网客户端]
B[代理服务端-外网]
C[请求端]
D[新的请求端]
E[代理客户端--内网]
F[内网服务器]
A-->|请求http端口| B
B -->|解析成Request| C
C -->|修改Request中的Header|D
D -->|发送HTTP请求数据给CenterClient|E
E -->|请求内网服务器转发数据|F

以下是内网服务器返回数据给外网客户端的流程

graph TD
A[外网客户端]
B[代理服务端-外网]
C[新的返回端]
D[返回端]
E[代理客户端--内网]
F[内网服务器]
F -->|返回Response|E
E -->|发送HTTP数据给CenterServer|D
D -->|修改头信息加工|C
C -->|将数据转发给|B
B -->|返回数据|A

转发中的注意事项

我们可以获取完整的Request再进行请求吗？

如果我们这么操作，当数据包非常的大的时候例如1G，我们此时在内存中将有完整的1G内存，那么此时只需有数个同一类的请求，将会耗尽我们的内存，所以我们必须不能这么处理。

超大文件下载的转发

超大文件必须将得到的数据及时的转发给客户端，此时在内存中的值才不至于太大，又能及时的传输给客户端，要不然可能大文件下载到中转服务器的时间内客户端得不到任何数据就会空耗掉这时间。

http/1.1中的chunked的处理

因为http/1.1的chunked协议，由RFC 2616定义，

分块编码(Transfer-Encoding: chunked)是超文本传输协议(HTTP)中的一种数据传输机制,允许HTTP由网页服务器发送给客户端的数据可以分成多个部分。分块传输编码只在HTTP协议1.1版本(HTTP/1.1)中提供，如果头部中有该选项，则代表数据包是chunked格式。数据分解成一系列数据块,并以一个或多个块发送,这样服务器可以发送数据而不需要预先知道发送内容的总大小。

比如我们常看到的

rust
for data in res.chunk() {
}

就是表示的是数据分段接收，对于大数据这个尤为重要。

此种报文的示例这时,报文中的实体需要改为用一系列分块来传输。每个分块包含十六进制的长度值和数据,长度值独占一行,长度不包括它结尾的 CRLF(\r\n),也不包括分块数据结尾的 CRLF。最后一个分块长度值必须为 0,对应的分块数据没有内容,表示实体结束。例:

http
HTTP/1.1 200 OK
Content-Type: text/plain
Transfer-Encoding: chunked

a\r\n
01234567890\r\n
1e\r\n
wmproxy is very good nat tool\r\n
0\r\n
\r\n

此种报文中我们必须进行解析，因为客户端可能是keep-alive选项，可以连续进行多发。所以收到的Request和Response都是连续的。必须知道何处结束才能继续解析下一个Request/Response。http2不需要，因为http2自带的data分包机制就有这些数据的处理

header数据的定义

header的修改分为两部分，一部分是对请求Request的重写，另一部分是对返回Response的重写。所以我们必须同时支持这两种，且将其区分出来。每条header信息我们将定定义一个可变长的数组，如第一个字符为proxy则表示对Request修改。
关于修改的动作有
1. 添加，如x-forward-for需要末尾添加，我们用操作符+，比如[proxy, +, x-forward-for, $client_ip]
2. 删除，我们用操作符+，如[-, hidden]
3. 设置，设置我们默认不做任何参数，直接以header_name开头，如[custom-key, custom-value]
4. 默认值，有些值有了参数我们就不将其重写，如果没有我们则设为默认值，我们用操作符?，如[?, server, wmproxy]

所以我们client.yaml的配置新增至如下：

yaml
# 连接服务端地址
server: 127.0.0.1:8091
# 连接服务端是否加密
ts: true

# 内网映射配置的数组
mappings:
  #将localhost的域名转发到本地的127.0.0.1:8080
  - name: web
    mode: http
    local_addr: 127.0.0.1:8080
    domain: localhost
    headers:
      - [proxy, +, x-forward-for, $client_ip]
      - [-, hidden]
      - [custom-key, custom-value]
      - [?, server, wmproxy]

mappings的结构修改

rust
pub struct MappingConfig {
    pub name: String,
    pub mode: String,
    pub local_addr: Option<SocketAddr>,
    #[serde(default = "default_domain")]
    pub domain: String,
    #[serde(default = "default_header")]
    pub headers: Vec<Vec<String>>,
}

我们把headers定义成一个动态的数组。根据不同的类型做不同的数据，因为长度有变化所以做不定长参数。以下是代码解析

rust
pub fn parse<T: Buf>(header: ProtFrameHeader, mut buf: T) -> ProxyResult<ProtMapping> {
    must_have!(buf, 2)?;
    let len = buf.get_u16() as usize;
    let mut mappings = vec![];
    
    for _ in 0..len {
        let name = read_short_string(&mut buf)?;
        let mode = read_short_string(&mut buf)?;
        let domain = read_short_string(&mut buf)?;
        let mut headers = vec![];
        must_have!(buf, 2)?;
        let len = buf.get_u16();
        for _ in 0 .. len {
            let mut header = vec![];
            must_have!(buf, 1)?;
            let sub_len = buf.get_u8();
            for _ in 0..sub_len {
                header.push(read_short_string(&mut buf)?);
            }
            headers.push(header);
        }
        mappings.push(MappingConfig::new(name, mode, domain, headers));
    }
    Ok(ProtMapping {
        sock_map: header.sock_map(),
        mappings,
    })
}

如此解析成一个完整的对应域名的结构，因为服务端用不到local_addr所以不做传输。

核心代码的实现

核心处理代码在trans/http.rs下，外部传入一个可读可写的stream，可能是TcpStream也可能是TlsStream<TcpStream>或者其它，同时把接收的SocketAddr传入，以方便后续获取$client_ip的头文件信息。

预处理

rust
pub async fn process<T>(self, inbound: T, addr: SocketAddr) -> Result<(), ProxyError<T>>
where
    T: AsyncRead + AsyncWrite + Unpin + Debug,
{
    println!("new process {:?}", inbound);
    let build = Client::builder();
    let (virtual_sender, virtual_receiver) = channel::<ProtFrame>(10);
    let stream = VirtualStream::new(self.sock_map, self.sender.clone(), virtual_receiver);
    let mut client = Client::new(build.value().ok().unwrap(), stream);
    let (receiver, sender) = client.split().unwrap();
    let oper = HttpOper {
        receiver,
        sender,
        sender_work: self.sender_work.clone(),
        virtual_sender: Some(virtual_sender),
        sock_map: self.sock_map,
        mappings: self.mappings.clone(),
        http_map: None,
    };
    let mut server = Server::new(inbound, Some(addr), oper);
    tokio::spawn( async move {
        let _ = client.wait_operate().await;
    });
    let _ret = server.incoming(Self::operate).await;
    if _ret.is_err() {
        println!("ret = {:?}", _ret);
    }
    Ok(())
}

此时我们创建一个虚拟的Stream来做双边互传，但是此时我们还没有收到任何的Request请求，我们并不知道当前的Host，此时我们还未发送ProtCreate，等真正处理请求的时候做处理，HttpOper是处理每个操作时均会带的参数，我们可以根据自己需要带上该参数。

后续处理，其中我们读和写都用RecvStream，做到读多少数据转发多少数据，以保证数据处理的及时性

rust
async fn inner_operate(
    mut req: Request<RecvStream>,
    data: Arc<Mutex<HttpOper>>,
) -> ProtResult<Option<Response<RecvStream>>> {
    println!("receiver req = {:?}", req.url());
    let mut value = data.lock().await;
    let sender = value.virtual_sender.take();
    // 传在该参数则为第一次, 第一次的时候发送Create创建绑定连接
    if sender.is_some() {
        let host_name = req.get_host().unwrap_or(String::new());
        // 取得相关的host数据，对内网的映射端做匹配，如果未匹配到返回错误，表示不支持
        {
            let mut config = None;
            let mut is_find = false;
            {
                let read = value.mappings.read().await;
                for v in &*read {
                    if v.domain == host_name {
                        is_find = true;
                        config = Some(v.clone());
                    }
                }
            }
            if !is_find {
                return Ok(Some(Response::builder().status(404).body("not found").ok().unwrap().into_type()));
            }
            value.http_map = config;
        }

        println!("do create prot {}, host = {:?}", value.sock_map, req.get_host());

        let create = ProtCreate::new(value.sock_map, Some(req.get_host().unwrap_or(String::new())));
        let _ = value.sender_work.send((create, sender.unwrap())).await;
    }

    if let Some(config) = &value.http_map {
        // 复写Request的头文件信息
        HeaderHelper::rewrite_request(&mut req, &config.headers);
    }

    // 将请求发送出去
    value.sender.send(req).await?;
    // 等待返回数据的到来
    let mut res = value.receiver.recv().await;
    if res.is_some() {
        if let Some(config) = &value.http_map {
            // 复写Response的头文件信息
            HeaderHelper::rewrite_response(res.as_mut().unwrap(), &config.headers);
        }
        return Ok(res);
    } else {
        return Ok(Some(Response::builder().status(503).body("cant trans").ok().unwrap().into_type()));
    }
}