6. 用Rust手把手編寫一個wmproxy(代理，內網穿透等), 通訊協議源碼解讀篇

来源:https://www.cnblogs.com/luojiawaf/archive/2023/09/30/17737591.html
-Advertisement-
Play Games

用Rust手把手編寫一個wmproxy(代理，內網穿透等), 通訊協議源碼解讀篇 項目 ++wmproxy++ gite: https://gitee.com/tickbh/wmproxy github: https://github.com/tickbh/wmproxy 事件模型的選取 OS線程, ...


用Rust手把手編寫一個wmproxy(代理，內網穿透等), 通訊協議源碼解讀篇

項目 ++wmproxy++

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

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

事件模型的選取

  • OS線程, 簡單的一個IO對應一個系統級別的線程，通常單進程創建的線程數是有限的，線上程與線程間同步數據會相當困難，線程間的調度爭用會相當損耗效率，不適合IO密集的場景。
  • 事件驅動(Event driven), 事件驅動基本上是最早的高併發的IO密集型的編程模式了，如C++的libevent，RUST的MIO，通過監聽IO的可讀可寫從而進行編程設計，缺點通常跟回調( Callback )一起使用，如果使用不好，回調層級過多會有回調地獄的風險。
  • 協程(Coroutines) 可能是目前比較火的併發模型，火遍全球的Go語言的協程設計就非常優秀。協程跟線程類似，無需改變編程模型，同時它也跟async類似，可以支持大量的任務併發運行。
  • actor模型 是erlang的殺手鐧之一，它將所有併發計算分割成一個一個單元，這些單元被稱為actor,單元之間通過消息傳遞的方式進行通信和數據傳遞，跟分散式系統的設計理念非常相像。由於actor模型跟現實很貼近，因此它相對來說更容易實現，但是一旦遇到流控制、失敗重試等場景時，就會變得不太好用
  • async/await， 該模型為非同步編輯模型，async模型的問題就是內部實現機制過於複雜，對於用戶來說，理解和使用起來也沒有線程和協程簡單。主要是等待完成狀態await，就比如讀socket數據，等待系統將數據送達再繼續觸發讀操作的執行，從而答到無損耗的運行。

這裡我們選擇的是async/await的模式

Rust中的async

  • Future 在 Rust 中是惰性的，只有在被輪詢(poll)時才會運行， 因此丟棄一個 future 會阻止它未來再被運行, 你可以將Future理解為一個在未來某個時間點被調度執行的任務。在Rust中調用非同步函數沒有用await會被編輯器警告，因為這不符合預期。
  • Async 在 Rust 中使用開銷是零， 意味著只有你能看到的代碼(自己的代碼)才有性能損耗，你看不到的(async 內部實現)都沒有性能損耗，例如，你可以無需分配任何堆記憶體、也無需任何動態分發來使用 async，這對於熱點路徑的性能有非常大的好處，正是得益於此，Rust 的非同步編程性能才會這麼高。
  • Rust 非同步運行時，Rust社區生態中已經提供了非常優異的運行時實現例如tokio，官方版本的async目前的生態相對tokio會差許多
  • 運行時同時支持單線程和多線程

流代碼的封裝

跟數據通訊相關的代碼均放在streams目錄下麵。

  1. center_client.rs中的CenterClient表示中心客戶端，提供主動連接服務端的能力並可選擇為加密（TLS）或者普通模式，並且將該客戶端收發的消息轉給服務端
  2. center_server.rs中的CenterServer表示中心服務端，接受中心客戶端的連接，並且將信息處理或者轉發
  3. trans_stream.rs中的TransStream表示轉發流量端，提供與中心端綁定的讀出寫入功能，在代理伺服器中客戶端接收的連接因為無需處理任何數據，直接綁定為TransStream將數據完整的轉發給服務端
  4. virtual_stream.rs中的VirtualStream表示虛擬端，虛擬出一個流連接，並實現AsyncRead及AsyncRead，可以和流一樣正常操作，在代理伺服器中服務端接收到新連接，把他虛擬成一個VirtualStream，就可以直接和他連接的伺服器上做雙向綁定。

幾種流式在代碼中的轉化

HTTP代理

下麵展示的是http代理，通過加密TLS中的轉化

flowchart TD A[TcpStream請求到代理]<-->|建立連接/明文|B[代理轉化成TransStream] B<-->|轉發到/內部|C[中心客戶端] C<-->|建立加密連接/加密|D[TlsStream< TcpStream>綁定中心服務端] D<-->|收到Create/內部|E[虛擬出VirtualStream] E<-->|解析到host並連接/明文|F[TcpStream連接到http伺服器]

上述過程實現了程式中實現了http的代理轉發

HTTP內網穿透

以下是http內網穿透在代理中的轉化

flowchart TD A[服務端綁定http對外埠]<-->|接收連接/明文|B[外部的TcpStream] B<-->|轉發到/內部|C[中心服務端並綁定TransStream] C<-->|通過客戶的加密連接推送/加密|D[TlsStream< TcpStream>綁定中心客戶端] D<-->|收到Create/內部|E[虛擬出VirtualStream] E<-->|解析對應的連接信息/明文|F[TcpStream連接到內網的http伺服器]

上述過程可以主動把公網的請求連接轉發到內網，由內網提供完服務後再轉發到公網的請求，從而實現內網穿透。

流代碼的介紹

CenterClient中心客端

下麵是代碼類的定義

/// 中心客戶端
/// 負責與服務端建立連接，斷開後自動再重連
pub struct CenterClient {
    /// tls的客戶端連接信息
    tls_client: Option<Arc<rustls::ClientConfig>>,
    /// tls的客戶端連接功能變數名稱
    domain: Option<String>,
    /// 連接中心伺服器的地址
    server_addr: SocketAddr,
    /// 內網映射的相關消息
    mappings: Vec<MappingConfig>,
    /// 存在普通連接和加密連接，此處不為None則表示普通連接
    stream: Option<TcpStream>,
    /// 存在普通連接和加密連接，此處不為None則表示加密連接
    tls_stream: Option<TlsStream<TcpStream>>,
    /// 綁定的下一個sock_map映射
    next_id: u32,

    /// 發送Create，並將綁定的Sender發到做綁定
    sender_work: Sender<(ProtCreate, Sender<ProtFrame>)>,
    /// 接收的Sender綁定，開始服務時這值move到工作協程中，所以不能二次調用服務
    receiver_work: Option<Receiver<(ProtCreate, Sender<ProtFrame>)>>,

    /// 發送協議數據，接收到服務端的流數據，轉發給相應的Stream
    sender: Sender<ProtFrame>,
    /// 接收協議數據，並轉發到服務端。
    receiver: Option<Receiver<ProtFrame>>,
}

主要的邏輯流程，迴圈監聽數據流的到達，同時等待多個非同步的到達，這裡用的是tokio::select!巨集

loop {
    let _ = tokio::select! {
        // 嚴格的順序流
        biased;
        // 新的流建立，這裡接收Create併進行綁定
        r = receiver_work.recv() => {
            if let Some((create, sender)) = r {
                map.insert(create.sock_map(), sender);
                let _ = create.encode(&mut write_buf);
            }
        }
        // 數據的接收，並將數據寫入給遠程端
        r = receiver.recv() => {
            if let Some(p) = r {
                let _ = p.encode(&mut write_buf);
            }
        }
        // 數據的等待讀取，一旦流可讀則觸發，讀到0則關閉主動關閉所有連接
        r = reader.read(&mut vec) => {
            match r {
                Ok(0)=>{
                    is_closed=true;
                    break;
                }
                Ok(n) => {
                    read_buf.put_slice(&vec[..n]);
                }
                Err(_err) => {
                    is_closed = true;
                    break;
                },
            }
        }
        // 一旦有寫數據，則嘗試寫入數據，寫入成功後扣除相應的數據
        r = writer.write(write_buf.chunk()), if write_buf.has_remaining() => {
            match r {
                Ok(n) => {
                    write_buf.advance(n);
                    if !write_buf.has_remaining() {
                        write_buf.clear();
                    }
                }
                Err(e) => {
                    println!("center_client errrrr = {:?}", e);
                },
            }
        }
    };

    loop {
        // 將讀出來的數據全部解析成ProtFrame併進行相應的處理，如果是0則是自身消息，其它進行轉發
        match Helper::decode_frame(&mut read_buf)? {
            Some(p) => {
                match p {
                    ProtFrame::Create(p) => {
                    }
                    ProtFrame::Close(_) | ProtFrame::Data(_) => {
                    },
                }
            }
            None => {
                break;
            }
        }
    }
}

CenterServer中心服務端

下麵是代碼類的定義

/// 中心服務端
/// 接受中心客戶端的連接，並且將信息處理或者轉發
pub struct CenterServer {
    /// 代理的詳情信息，如用戶密碼這類
    option: ProxyOption,
    
    /// 發送協議數據，接收到服務端的流數據，轉發給相應的Stream
    sender: Sender<ProtFrame>,
    /// 接收協議數據，並轉發到服務端。
    receiver: Option<Receiver<ProtFrame>>,

    /// 發送Create，並將綁定的Sender發到做綁定
    sender_work: Sender<(ProtCreate, Sender<ProtFrame>)>,
    /// 接收的Sender綁定，開始服務時這值move到工作協程中，所以不能二次調用服務
    receiver_work: Option<Receiver<(ProtCreate, Sender<ProtFrame>)>>,
    /// 綁定的下一個sock_map映射，為雙數
    next_id: u32,
}

主要的邏輯流程，迴圈監聽數據流的到達，同時等待多個非同步的到達，這裡用的是tokio::select!巨集，select處理方法與Client相同，均處理相同邏輯，不同的是接收數據包後數據端是處理的proxy的請求，而Client處理的是內網穿透的邏輯

loop {
    // 將讀出來的數據全部解析成ProtFrame併進行相應的處理，如果是0則是自身消息，其它進行轉發
    match Helper::decode_frame(&mut read_buf)? {
        Some(p) => {
            match p {
                ProtFrame::Create(p) => {
                    tokio::spawn(async move {
                        let _ = Proxy::deal_proxy(stream, flag, username, password, udp_bind).await;
                    });
                }
                ProtFrame::Close(_) | ProtFrame::Data(_) => {
                },
            }
        }
        None => {
            break;
        }
    }
}

TransStream轉發流量端

下麵是代碼類的定義

/// 轉發流量端
/// 提供與中心端綁定的讀出寫入功能
pub struct TransStream<T>
where
    T: AsyncRead + AsyncWrite + Unpin,
{
    // 流有相應的AsyncRead + AsyncWrite + Unpin均可
    stream: T,
    // sock綁定的句柄
    id: u32,
    // 讀取的數據緩存，將轉發成ProtFrame
    read: BinaryMut,
    // 寫的數據緩存，直接寫入到stream下，從ProtFrame轉化而來
    write: BinaryMut,
    // 收到數據通過sender發送給中心端
    in_sender: Sender<ProtFrame>,
    // 收到中心端的寫入請求，轉成write
    out_receiver: Receiver<ProtFrame>,
}

主要的邏輯流程，迴圈監聽數據流的到達，同時等待多個非同步的到達，這裡用的是tokio::select!巨集，監聽的對象有stream可讀，可寫，sender的寫發送及receiver的可接收

loop {
    // 有剩餘數據，優先轉化成Prot，因為數據可能從外部直接帶入
    if self.read.has_remaining() {
        link.push_back(ProtFrame::new_data(self.id, self.read.copy_to_binary()));
        self.read.clear();
    }

    tokio::select! {
        n = reader.read(&mut buf) => {
            let n = n?;
            if n == 0 {
                return Ok(())
            } else {
                self.read.put_slice(&buf[..n]);
            }
        },
        r = writer.write(self.write.chunk()), if self.write.has_remaining() => {
            match r {
                Ok(n) => {
                    self.write.advance(n);
                    if !self.write.has_remaining() {
                        self.write.clear();
                    }
                }
                Err(_) => todo!(),
            }
        }
        r = self.out_receiver.recv() => {
            if let Some(v) = r {
                if v.is_close() || v.is_create() {
                    return Ok(())
                } else if v.is_data() {
                    match v {
                        ProtFrame::Data(d) => {
                            self.write.put_slice(&d.data().chunk());
                        }
                        _ => unreachable!(),
                    }
                }
            } else {
                return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid frame"))
            }
        }
        p = self.in_sender.reserve(), if link.len() > 0 => {
            match p {
                Err(_)=>{
                    return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid frame"))
                }
                Ok(p) => {
                    p.send(link.pop_front().unwrap())
                }, 
            }
        }
    }

VirtualStream虛擬端

下麵是代碼類的定義，我們並未有真實的socket，通過虛擬出的端方便後續的操作

/// 虛擬端
/// 虛擬出一個流連接，並實現AsyncRead及AsyncRead，可以和流一樣正常操作
pub struct VirtualStream
{
    // sock綁定的句柄
    id: u32,
    // 收到數據通過sender發送給中心端
    sender: PollSender<ProtFrame>,
    // 收到中心端的寫入請求，轉成write
    receiver: Receiver<ProtFrame>,
    // 讀取的數據緩存，將轉發成ProtFrame
    read: BinaryMut,
    // 寫的數據緩存，直接寫入到stream下，從ProtFrame轉化而來
    write: BinaryMut,
}

虛擬的流主要通過實現AsyncRead及AsyncWrite


impl AsyncRead for VirtualStream
{
    // 有讀取出數據，則返回數據，返回數據0的Ready狀態則表示已關閉
    fn poll_read(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut [std](https://note.youdao.com/)[link](https://note.youdao.com/)::task::Context<'_>,
        buf: &mut tokio::io::ReadBuf<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        loop {
            match self.receiver.poll_recv(cx) {
                Poll::Ready(value) => {
                    if let Some(v) = value {
                        if v.is_close() || v.is_create() {
                            return Poll::Ready(Ok(()))
                        } else if v.is_data() {
                            match v {
                                ProtFrame::Data(d) => {
                                    self.read.put_slice(&d.data().chunk());
                                }
                                _ => unreachable!(),
                            }
                        }
                    } else {
                        return Poll::Ready(Ok(()))
                    }
                },
                Poll::Pending => {
                    if !self.read.has_remaining() {
                        return Poll::Pending;
                    }
                },
            }


            if self.read.has_remaining() {
                let copy = std::cmp::min(self.read.remaining(), buf.remaining());
                buf.put_slice(&self.read.chunk()[..copy]);
                self.read.advance(copy);
                return Poll::Ready(Ok(()));
            }
        }
        
    }
}


impl AsyncWrite for VirtualStream
{
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &[u8],
    ) -> std::task::Poll<Result<usize, std::io::Error>> {
        self.write.put_slice(buf);
        if let Err(_) = ready!(self.sender.poll_reserve(cx)) {
            return Poll::Pending;
        }
        let binary = Binary::from(self.write.chunk().to_vec());
        let id = self.id;
        if let Ok(_) = self.sender.send_item(ProtFrame::Data(ProtData::new(id, binary))) {
            self.write.clear();
        }
        Poll::Ready(Ok(buf.len()))
    }

}

至此基本幾個大類已設置完畢，接下來僅需簡單的拓展就能實現內網穿透功能。


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • Flutter/Dart第05天：Dart特殊特性Mixin詳解
    在Java語言中，子類只能繼承extends單個父類，實現implements多個介面（即單繼承和多實現）。在Dart語言中，所有類型均是Object子類，它們也是單繼承和多實現，但Dart中有個Mixin的高級特性，它可以做到更多的代碼復用（單繼承、多實現、多Mixin代碼復用）…… ...
  • 魅族雲服務自動一鍵所有選擇圖片下載。
    魅族雲服務的相冊功能，沒有一鍵選擇所有的圖片，就挺噁心的。魅族不一直提供雲相冊的服務了，就需要將圖片全部下載。之前有大神寫過油潑猴的腳本。 今天拿來用，發現用不了。 又在網上查一下了，有npm的開源下載工具。附上碼雲地址，沒用過。https://gitee.com/moreant/mpcb 但是部署 ...
  • 使用正則表達式判斷日期字元串格式是否合法遇到的問題(解決)
    我們在使用SpringMVC從前端接受傳遞過來的日期數據時，預設傳遞過來的數據是String類型，如果我們從前端傳遞過來的數據格式是yyyy/MM/dd，SpringMVC有內置類型轉化器會將String類型自動轉化成Date類型。但如果我們從前端傳遞過來的數據格式是yyyy-MM-dd，Sprin... ...
  • 聲明式調用 —— SpringCloud OpenFeign
    Feign 簡介 Spring Cloud Feign 是一個 HTTP 請求調用的輕量級框架，可以以 Java 介面註解的方式調用 HTTP 請求，而不用通過封裝 HTTP 請求報文的方式直接調用 Feign 通過處理註解，將請求模板化，當實際調用的時候傳入參數，根據參數再應用到請求上，進而轉化成 ...
  • Get請求和Post請求
    Get請求和Post請求都是HTTP協議中的兩種常見請求方法，底層都是TCP/IP協議，用於客戶端與伺服器之間的數據傳輸。 ...
  • 洛谷題解 | AT_abc321_c Primes on Interval
    目錄題目翻譯題目描述輸入格式輸出格式樣例 #1樣例輸入 #1樣例輸出 #1樣例 #2樣例輸入 #2樣例輸出 #2樣例 #3樣例輸入 #3樣例輸出 #3題目簡化題目思路AC代碼 題目翻譯 【題目描述】 你決定用素數定理來做一個調查. 眾所周知, 素數又被稱為質數，其含義就是除了數字一和本身之外不能被其 ...
  • Python 布爾類型
    布爾值表示兩個值之一：True（真）或False（假）。 布爾值 在編程中，您經常需要知道一個表達式是否為True或False。 您可以在Python中評估任何表達式，並獲得兩個答案之一：True或False。 當您比較兩個值時，表達式會被評估，Python會返回布爾答案： 示例 print(10 ...
  • 【matplotlib 實戰】--面積圖
    面積圖，或稱區域圖，是一種隨有序變數的變化，反映數值變化的統計圖表。 面積圖也可用於多個系列數據的比較。這時，面積圖的外觀看上去類似層疊的山脈，在錯落有致的外形下表達數據的總量和趨勢。面積圖不僅可以清晰地反映出數據的趨勢變化，也能夠強調不同類別的數據間的差距對比。 面積圖的特點在於，折線與自變數坐標 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類