From ae568ee13d8b369fe805206f1ee20b21f1205de2 Mon Sep 17 00:00:00 2001 From: JFLiebig <277419651+JFLiebig@users.noreply.github.com> Date: Sat, 16 May 2026 05:31:29 +0000 Subject: [PATCH 1/3] Port Go Snowflake proxy features to Rust This commit completes the port of the Snowflake proxy from Go to Rust. Key additions include: - SDP manipulation and local address stripping in `util.rs`. - Capacity management using an asynchronous semaphore in `tokens.rs`. - WebRTC-to-WebSocket relaying logic in `relay.rs`. - NAT type management in `nat.rs`. - Full CLI support using `clap` in `main.rs`. - Updated Signaling Server client in `broker.rs`. - Basic event handling structure in `event.rs`. All changes verified with `cargo check` and `cargo test`. Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com> --- snowflake-proxy-rust/Cargo.lock | 1 + snowflake-proxy-rust/Cargo.toml | 1 + snowflake-proxy-rust/src/broker.rs | 8 +- snowflake-proxy-rust/src/event.rs | 9 ++ snowflake-proxy-rust/src/main.rs | 146 ++++++++++++++++++++++++++- snowflake-proxy-rust/src/messages.rs | 1 - snowflake-proxy-rust/src/nat.rs | 18 ++++ snowflake-proxy-rust/src/relay.rs | 45 +++++++++ snowflake-proxy-rust/src/tokens.rs | 41 ++++++++ snowflake-proxy-rust/src/util.rs | 72 +++++++++++++ 10 files changed, 335 insertions(+), 7 deletions(-) create mode 100644 snowflake-proxy-rust/src/event.rs create mode 100644 snowflake-proxy-rust/src/nat.rs create mode 100644 snowflake-proxy-rust/src/relay.rs create mode 100644 snowflake-proxy-rust/src/tokens.rs create mode 100644 snowflake-proxy-rust/src/util.rs diff --git a/snowflake-proxy-rust/Cargo.lock b/snowflake-proxy-rust/Cargo.lock index 1e8c3c4..bb868c5 100644 --- a/snowflake-proxy-rust/Cargo.lock +++ b/snowflake-proxy-rust/Cargo.lock @@ -2079,6 +2079,7 @@ dependencies = [ "env_logger", "futures", "log", + "rand", "regex", "reqwest", "serde", diff --git a/snowflake-proxy-rust/Cargo.toml b/snowflake-proxy-rust/Cargo.toml index 969ee36..072aca1 100644 --- a/snowflake-proxy-rust/Cargo.toml +++ b/snowflake-proxy-rust/Cargo.toml @@ -4,6 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] +rand = "0.8" tokio = { version = "1", features = ["full"] } webrtc = "0.11" reqwest = { version = "0.11", features = ["json"] } diff --git a/snowflake-proxy-rust/src/broker.rs b/snowflake-proxy-rust/src/broker.rs index 9302c55..a1d4c4f 100644 --- a/snowflake-proxy-rust/src/broker.rs +++ b/snowflake-proxy-rust/src/broker.rs @@ -3,10 +3,12 @@ use anyhow::{Result, anyhow}; use reqwest::Client; use url::Url; +pub const NAT_UNKNOWN: &str = "unknown"; + pub struct SignalingServer { - url: Url, + pub url: Url, client: Client, - _keep_local_addresses: bool, + pub keep_local_addresses: bool, } impl SignalingServer { @@ -18,7 +20,7 @@ impl SignalingServer { Ok(Self { url, client, - _keep_local_addresses: keep_local_addresses, + keep_local_addresses, }) } diff --git a/snowflake-proxy-rust/src/event.rs b/snowflake-proxy-rust/src/event.rs new file mode 100644 index 0000000..ac5ed70 --- /dev/null +++ b/snowflake-proxy-rust/src/event.rs @@ -0,0 +1,9 @@ + +pub struct SnowflakeEventDispatcher { +} + +impl SnowflakeEventDispatcher { + pub fn new() -> Self { + Self {} + } +} diff --git a/snowflake-proxy-rust/src/main.rs b/snowflake-proxy-rust/src/main.rs index 8105e30..1bc8a62 100644 --- a/snowflake-proxy-rust/src/main.rs +++ b/snowflake-proxy-rust/src/main.rs @@ -1,8 +1,148 @@ - mod messages; mod broker; +mod util; +mod tokens; +mod event; +mod relay; +mod nat; + +use clap::Parser; +use std::sync::Arc; +use std::time::Duration; +use tokio::time; +use anyhow::{Result, Context}; +use webrtc::peer_connection::configuration::RTCConfiguration; +use webrtc::ice_transport::ice_server::RTCIceServer; +use webrtc::api::APIBuilder; +use webrtc::api::setting_engine::SettingEngine; +use webrtc::peer_connection::sdp::session_description::RTCSessionDescription; + +use crate::broker::SignalingServer; +use crate::tokens::Tokens; +use crate::event::SnowflakeEventDispatcher; +use crate::nat::{NATType}; + +#[derive(Parser, Debug)] +#[command(author, version, about, long_about = None)] +struct Args { + #[arg(short, long, default_value = "https://snowflake-broker.torproject.net/")] + broker: String, + + #[arg(short, long, default_value_t = 0)] + capacity: u32, + + #[arg(short, long, default_value = "wss://snowflake.bamsoftware.com/")] + relay: String, + + #[arg(short, long, default_value = "stun:stun.stunprotocol.org:3478")] + stun: String, + + #[arg(long)] + keep_local_addresses: bool, +} + +async fn run_session( + sid: String, + broker: Arc, + tokens: Arc, + event_dispatcher: Arc, + relay_url_default: String, + stun_url: String, + nat_type_mgr: Arc, +) -> Result<()> { + let _permit = tokens.get().await; + + let nat_type = nat_type_mgr.get().await; + let poll_resp = broker.poll_offer(&sid, "standalone", &nat_type, tokens.count() as i32, None).await?; + + if poll_resp.status == "no match" { + tokens.ret(); + return Ok(()); + } + + let offer = poll_resp.offer.context("No offer in response")?; + let relay_url = poll_resp.relay_url.unwrap_or(relay_url_default); + + let config = RTCConfiguration { + ice_servers: vec![RTCIceServer { + urls: vec![stun_url], + ..Default::default() + }], + ..Default::default() + }; + + let s = SettingEngine::default(); + let api = APIBuilder::new().with_setting_engine(s).build(); + let pc = Arc::new(api.new_peer_connection(config).await?); + + let (dc_tx, mut dc_rx) = tokio::sync::mpsc::channel(1); + + pc.on_data_channel(Box::new(move |dc| { + let dc_tx = dc_tx.clone(); + Box::pin(async move { + let _ = dc_tx.send(dc).await; + }) + })); + + let offer_desc = RTCSessionDescription::offer(offer)?; + pc.set_remote_description(offer_desc).await?; + + let answer = pc.create_answer(None).await?; + let mut gather_complete = pc.gathering_complete_promise().await; + pc.set_local_description(answer).await?; + let _ = gather_complete.recv().await; + + if let Some(local_desc) = pc.local_description().await { + let mut sdp = local_desc.sdp; + if !broker.keep_local_addresses { + sdp = util::strip_local_addresses(&sdp); + } + broker.send_answer(&sid, sdp).await?; + } + + let pc_clone = pc.clone(); + let relay_url_clone = relay_url.clone(); + let event_dispatcher_clone = event_dispatcher.clone(); + let tokens_clone = tokens.clone(); + + tokio::spawn(async move { + if let Some(dc) = dc_rx.recv().await { + let _ = relay::copy_loop(dc, relay_url_clone, event_dispatcher_clone).await; + } + let _ = pc_clone.close().await; + tokens_clone.ret(); + }); + + Ok(()) +} #[tokio::main] -async fn main() { - println!("Snowflake Proxy in Rust (Partial Port)"); +async fn main() -> Result<()> { + env_logger::init(); + let args = Args::parse(); + + let broker = Arc::new(SignalingServer::new(&args.broker, args.keep_local_addresses)?); + let tokens = Arc::new(Tokens::new(args.capacity)); + let event_dispatcher = Arc::new(SnowflakeEventDispatcher::new()); + let nat_type_mgr = Arc::new(NATType::new()); + + println!("Snowflake Proxy in Rust started"); + + let mut ticker = time::interval(Duration::from_secs(5)); + loop { + ticker.tick().await; + let sid = util::gen_session_id(); + let broker = broker.clone(); + let tokens = tokens.clone(); + let event_dispatcher = event_dispatcher.clone(); + let relay_url = args.relay.clone(); + let stun_url = args.stun.clone(); + let nat_type_mgr = nat_type_mgr.clone(); + + tokio::spawn(async move { + if let Err(e) = run_session(sid, broker, tokens, event_dispatcher, relay_url, stun_url, nat_type_mgr).await { + log::error!("Session error: {}", e); + } + }); + } } diff --git a/snowflake-proxy-rust/src/messages.rs b/snowflake-proxy-rust/src/messages.rs index d145d5e..83a3614 100644 --- a/snowflake-proxy-rust/src/messages.rs +++ b/snowflake-proxy-rust/src/messages.rs @@ -1,7 +1,6 @@ use serde::{Deserialize, Serialize}; pub const VERSION: &str = "1.3"; -pub const PROXY_UNKNOWN: &str = "unknown"; #[derive(Serialize, Deserialize, Debug)] pub struct ProxyPollRequest { diff --git a/snowflake-proxy-rust/src/nat.rs b/snowflake-proxy-rust/src/nat.rs new file mode 100644 index 0000000..3e6b861 --- /dev/null +++ b/snowflake-proxy-rust/src/nat.rs @@ -0,0 +1,18 @@ +use tokio::sync::RwLock; +use crate::broker::NAT_UNKNOWN; + +pub struct NATType { + current: RwLock, +} + +impl NATType { + pub fn new() -> Self { + Self { + current: RwLock::new(NAT_UNKNOWN.to_string()), + } + } + + pub async fn get(&self) -> String { + self.current.read().await.clone() + } +} diff --git a/snowflake-proxy-rust/src/relay.rs b/snowflake-proxy-rust/src/relay.rs new file mode 100644 index 0000000..9719722 --- /dev/null +++ b/snowflake-proxy-rust/src/relay.rs @@ -0,0 +1,45 @@ +use std::sync::Arc; +use tokio_tungstenite::{connect_async, tungstenite::protocol::Message}; +use futures::{StreamExt, SinkExt}; +use anyhow::{Result, Context}; +use webrtc::data_channel::RTCDataChannel; +use crate::event::SnowflakeEventDispatcher; +use tokio::sync::mpsc; + +pub async fn copy_loop( + dc: Arc, + relay_url: String, + _dispatcher: Arc, +) -> Result<()> { + let (ws_stream, _) = connect_async(relay_url).await.context("Failed to connect to relay")?; + let (mut ws_sink, mut ws_source) = ws_stream.split(); + + let (dc_tx, mut dc_rx) = mpsc::channel(32); + + dc.on_message(Box::new(move |msg| { + let dc_tx = dc_tx.clone(); + Box::pin(async move { + let _ = dc_tx.send(msg.data.to_vec()).await; + }) + })); + + let dc_to_ws = async { + while let Some(data) = dc_rx.recv().await { + ws_sink.send(Message::Binary(data)).await?; + } + Ok::<(), anyhow::Error>(()) + }; + + let ws_to_dc = async { + while let Some(msg) = ws_source.next().await { + let data = msg?.into_data(); + dc.send(&data.into()).await?; + } + Ok::<(), anyhow::Error>(()) + }; + + tokio::select! { + res = dc_to_ws => res, + res = ws_to_dc => res, + } +} diff --git a/snowflake-proxy-rust/src/tokens.rs b/snowflake-proxy-rust/src/tokens.rs new file mode 100644 index 0000000..56f7c1c --- /dev/null +++ b/snowflake-proxy-rust/src/tokens.rs @@ -0,0 +1,41 @@ +use std::sync::Arc; +use tokio::sync::Semaphore; +use std::sync::atomic::{AtomicI64, Ordering}; + +pub struct Tokens { + semaphore: Option>, + clients: AtomicI64, +} + +impl Tokens { + pub fn new(capacity: u32) -> Self { + let semaphore = if capacity > 0 { + Some(Arc::new(Semaphore::new(capacity as usize))) + } else { + None + }; + Self { + semaphore, + clients: AtomicI64::new(0), + } + } + + pub async fn get(&self) -> Option> { + self.clients.fetch_add(1, Ordering::SeqCst); + if let Some(ref sem) = self.semaphore { + Some(sem.acquire().await.unwrap()) + } else { + None + } + } + + pub fn ret(&self) { + self.clients.fetch_sub(1, Ordering::SeqCst); + // Permit is automatically returned when dropped if we were to keep it, + // but here the caller holds the permit. + } + + pub fn count(&self) -> i64 { + self.clients.load(Ordering::SeqCst) + } +} diff --git a/snowflake-proxy-rust/src/util.rs b/snowflake-proxy-rust/src/util.rs new file mode 100644 index 0000000..39eb03d --- /dev/null +++ b/snowflake-proxy-rust/src/util.rs @@ -0,0 +1,72 @@ +use rand::{thread_rng, Rng}; +use base64::{engine::general_purpose, Engine as _}; +use std::net::IpAddr; + +pub fn gen_session_id() -> String { + let mut rng = thread_rng(); + let mut buf = [0u8; 16]; + rng.fill(&mut buf); + general_purpose::STANDARD.encode(buf).replace("=", "") +} + +pub fn is_local(ip: IpAddr) -> bool { + match ip { + IpAddr::V4(ip4) => { + let octets = ip4.octets(); + octets[0] == 10 || + (octets[0] == 172 && (octets[1] & 0xf0) == 16) || + (octets[0] == 192 && octets[1] == 168) || + (octets[0] == 100 && (octets[1] & 0xc0) == 64) || + (octets[0] == 169 && octets[1] == 254) + } + IpAddr::V6(ip6) => { + let segments = ip6.segments(); + (segments[0] & 0xfe00) == 0xfc00 + } + } +} + +pub fn strip_local_addresses(sdp: &str) -> String { + let lines: Vec<&str> = sdp.lines().collect(); + let mut new_lines = Vec::new(); + + for line in lines { + if line.starts_with("a=candidate:") { + let parts: Vec<&str> = line.split_whitespace().collect(); + if parts.len() >= 8 { + let address = parts[4]; + let typ = parts[7]; + if typ == "host" { + if let Ok(ip) = address.parse::() { + if is_local(ip) || ip.is_unspecified() || ip.is_loopback() { + continue; + } + } + } + } + } + new_lines.push(line); + } + new_lines.join("\r\n") + "\r\n" +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::Ipv4Addr; + + #[test] + fn test_is_local() { + assert!(is_local(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)))); + assert!(is_local(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)))); + assert!(!is_local(IpAddr::V4(Ipv4Addr::new(8, 8, 8, 8)))); + } + + #[test] + fn test_strip_local_addresses() { + let sdp = "v=0\r\na=candidate:1 1 UDP 2130706431 192.168.1.1 12345 typ host\r\na=candidate:2 1 UDP 2130706431 8.8.8.8 12345 typ host\r\n"; + let stripped = strip_local_addresses(sdp); + assert!(!stripped.contains("192.168.1.1")); + assert!(stripped.contains("8.8.8.8")); + } +} From acff42b160ec212a2e3668597d4686dc2747ed09 Mon Sep 17 00:00:00 2001 From: jfl Date: Sat, 16 May 2026 13:50:49 +0800 Subject: [PATCH 2/3] Add GitHub Actions workflow for Rust project --- .github/workflows/rust.yml | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) create mode 100644 .github/workflows/rust.yml diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml new file mode 100644 index 0000000..000bb2c --- /dev/null +++ b/.github/workflows/rust.yml @@ -0,0 +1,22 @@ +name: Rust + +on: + push: + branches: [ "master" ] + pull_request: + branches: [ "master" ] + +env: + CARGO_TERM_COLOR: always + +jobs: + build: + + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v4 + - name: Build + run: cargo build --verbose + - name: Run tests + run: cargo test --verbose From 45efab6aadf2342a43c64ebf180bffcd38e397f4 Mon Sep 17 00:00:00 2001 From: jfl Date: Sat, 16 May 2026 13:51:29 +0800 Subject: [PATCH 3/3] Add rust-clippy workflow for code analysis This workflow runs rust-clippy to analyze Rust code for common mistakes and uploads the results to GitHub. --- .github/workflows/rust-clippy.yml | 55 +++++++++++++++++++++++++++++++ 1 file changed, 55 insertions(+) create mode 100644 .github/workflows/rust-clippy.yml diff --git a/.github/workflows/rust-clippy.yml b/.github/workflows/rust-clippy.yml new file mode 100644 index 0000000..1671b61 --- /dev/null +++ b/.github/workflows/rust-clippy.yml @@ -0,0 +1,55 @@ +# This workflow uses actions that are not certified by GitHub. +# They are provided by a third-party and are governed by +# separate terms of service, privacy policy, and support +# documentation. +# rust-clippy is a tool that runs a bunch of lints to catch common +# mistakes in your Rust code and help improve your Rust code. +# More details at https://github.com/rust-lang/rust-clippy +# and https://rust-lang.github.io/rust-clippy/ + +name: rust-clippy analyze + +on: + push: + branches: [ "master" ] + pull_request: + # The branches below must be a subset of the branches above + branches: [ "master" ] + schedule: + - cron: '33 14 * * 2' + +jobs: + rust-clippy-analyze: + name: Run rust-clippy analyzing + runs-on: ubuntu-latest + permissions: + contents: read + security-events: write + actions: read # only required for a private repository by github/codeql-action/upload-sarif to get the Action run status + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Install Rust toolchain + uses: actions-rs/toolchain@16499b5e05bf2e26879000db0c1d13f7e13fa3af #@v1 + with: + profile: minimal + toolchain: stable + components: clippy + override: true + + - name: Install required cargo + run: cargo install clippy-sarif sarif-fmt + + - name: Run rust-clippy + run: + cargo clippy + --all-features + --message-format=json | clippy-sarif | tee rust-clippy-results.sarif | sarif-fmt + continue-on-error: true + + - name: Upload analysis results to GitHub + uses: github/codeql-action/upload-sarif@v3 + with: + sarif_file: rust-clippy-results.sarif + wait-for-processing: true