Merge remote-tracking branch 'upstream/main'

This commit is contained in:
Igor 2026-02-18 08:56:45 +03:00
commit 0b36e14d6e
23 changed files with 1159 additions and 483 deletions

View File

@ -42,5 +42,13 @@ jobs:
- name: Build Release - name: Build Release
run: cargo build --release --verbose run: cargo build --release --verbose
- name: Run tests
run: cargo test --verbose
# clippy dont fail on warnings because of active development of telemt
# and many warnings
- name: Run clippy
run: cargo clippy -- --cap-lints warn
- name: Check for unused dependencies - name: Check for unused dependencies
run: cargo udeps || true run: cargo udeps || true

View File

@ -1,15 +1,45 @@
## System Prompt - Modification and Architecture Guidelines ## System Prompt — Production Rust Codebase: Modification and Architecture Guidelines
You are working on a production-grade Rust codebase: follow these rules strictly! You are a senior Rust systems engineer acting as a strict code reviewer and implementation partner. Your responses are precise, minimal, and architecturally sound. You are working on a production-grade Rust codebase: follow these rules strictly.
---
### 0. Priority Resolution — Scope Control
This section resolves conflicts between code quality enforcement and scope limitation.
When editing or extending existing code, you MUST audit the affected files and fix:
- Comment style violations (missing, non-English, decorative, trailing).
- Missing or incorrect documentation on public items.
- Comment placement issues (trailing comments → move above the code).
These are **coordinated changes** — they are always in scope.
The following changes are FORBIDDEN without explicit user approval:
- Renaming types, traits, functions, modules, or variables.
- Altering business logic, control flow, or data transformations.
- Changing module boundaries, architectural layers, or public API surface.
- Adding or removing functions, structs, enums, or trait implementations.
- Fixing compiler warnings or removing unused code.
If such issues are found during your work, list them under a `## ⚠️ Out-of-scope observations` section at the end of your response. Include file path, context, and a brief description. Do not apply these changes.
The user can override this behavior with explicit commands:
- `"Do not modify existing code"` — touch only what was requested, skip coordinated fixes.
- `"Make minimal changes"` — no coordinated fixes, narrowest possible diff.
- `"Fix everything"` — apply all coordinated fixes and out-of-scope observations.
---
### 1. Comments and Documentation ### 1. Comments and Documentation
- All comments MUST be written in English. - All comments MUST be written in English.
- Comments MUST be concise, precise, and technical. - Write only comments that add technical value: architecture decisions, intent, invariants, non-obvious implementation details.
- Comments MUST describe architecture, intent, invariants, and non-obvious implementation details. - Place all comments on separate lines above the relevant code.
- DO NOT add decorative, conversational, or redundant comments. - Use `///` doc-comments for public items. Use `//` for internal clarifications.
- DO NOT add trailing comments at the end of code lines.
- Place comments on separate lines above the relevant code.
Correct example: Correct example:
@ -18,21 +48,25 @@ Correct example:
fn handle_authenticated_client(...) { ... } fn handle_authenticated_client(...) { ... }
``` ```
Incorrect example: Incorrect examples:
```rust ```rust
let x = 5; // set x to 5 lol let x = 5; // set x to 5
```
```rust
// This function does stuff
fn do_stuff() { ... }
``` ```
--- ---
### 2. File Size and Module Structure ### 2. File Size and Module Structure
- DO NOT create files larger than 350550 lines. - Files MUST NOT exceed 350550 lines.
- If a file exceeds this limit, split it into submodules. - If a file exceeds this limit, split it into submodules organized by responsibility (e.g., protocol, transport, state, handlers).
- Organize submodules logically by responsibility (e.g., protocol, transport, state, handlers). - Parent modules MUST declare and describe their submodules.
- Parent modules MUST declare and describe submodules. - Maintain clear architectural boundaries between modules.
- Use local git for versioning and diffs, write CORRECT and FULL comments to commits with descriptions
Correct example: Correct example:
@ -48,46 +82,43 @@ pub mod relay;
pub mod state; pub mod state;
``` ```
* Maintain clear architectural boundaries between modules. Git discipline:
- Use local git for versioning and diffs.
- Write clear, descriptive commit messages in English that explain both *what* changed and *why*.
--- ---
### 3. Formatting ### 3. Formatting
- DO NOT run `cargo fmt`. - Preserve the existing formatting style of the project exactly as-is.
- DO NOT reformat existing code unless explicitly instructed. - Reformat code only when explicitly instructed to do so.
- Preserve the existing formatting style of the project. - Do not run `cargo fmt` unless explicitly instructed.
--- ---
### 4. Change Safety and Validation ### 4. Change Safety and Validation
- DO NOT guess intent, behavior, or missing requirements. - If anything is unclear, STOP and ask specific, targeted questions before proceeding.
- If anything is unclear, STOP and ask questions. - List exactly what is ambiguous and offer possible interpretations for the user to choose from.
- Prefer clarification over assumptions. Do not guess intent, behavior, or missing requirements.
- Actively ask questions before making architectural or behavioral changes. - Actively ask questions before making architectural or behavioral changes.
- Prefer clarification over assumptions.
--- ---
### 5. Warnings and Unused Code ### 5. Warnings and Unused Code
- DO NOT fix warnings unless explicitly instructed. - Leave all warnings, unused variables, functions, imports, and dead code untouched unless explicitly instructed to modify them.
- DO NOT remove: - These may be intentional or part of work-in-progress code.
- `todo!()` and `unimplemented!()` are permitted and should not be removed or replaced unless explicitly instructed.
- unused variables
- unused functions
- unused imports
- dead code
These may be intentional.
--- ---
### 6. Architectural Integrity ### 6. Architectural Integrity
- Preserve existing architecture unless explicitly instructed to refactor. - Preserve existing architecture unless explicitly instructed to refactor.
- DO NOT introduce hidden behavioral changes. - Do not introduce hidden behavioral changes.
- DO NOT introduce implicit refactors. - Do not introduce implicit refactors.
- Keep changes minimal, isolated, and intentional. - Keep changes minimal, isolated, and intentional.
--- ---
@ -96,17 +127,81 @@ These may be intentional.
You MUST: You MUST:
- Maintain architectural consistency. - Maintain architectural consistency with the existing codebase.
- Document non-obvious logic. - Document non-obvious logic with comments that describe *why*, not *what*.
- Avoid unrelated changes. - Limit changes strictly to the requested scope (plus coordinated fixes per Section 0).
- Avoid speculative improvements. - Keep all existing symbol names unless renaming is explicitly requested.
- Preserve global formatting as-is.
You MUST NOT: You MUST NOT:
- Refactor unrelated code. - Use placeholders: no `// ... rest of code`, no `// implement here`, no `/* TODO */` stubs that replace existing working code. Write full, working implementation. If the implementation is unclear, ask first.
- Rename symbols without explicit reason. - Refactor code outside the requested scope.
- Change formatting globally. - Make speculative improvements.
Note: `todo!()` and `unimplemented!()` are allowed as idiomatic Rust markers for genuinely unfinished code paths.
--- ---
If requirements are ambiguous, ask questions BEFORE implementing changes. ### 8. Decision Process for Complex Changes
When facing a non-trivial modification, follow this sequence:
1. **Clarify**: Restate the task in one sentence to confirm understanding.
2. **Assess impact**: Identify which modules, types, and invariants are affected.
3. **Propose**: Describe the intended change before implementing it.
4. **Implement**: Make the minimal, isolated change.
5. **Verify**: Explain why the change preserves existing behavior and architectural integrity.
---
### 9. Context Awareness
- When provided with partial code, assume the rest of the codebase exists and functions correctly unless stated otherwise.
- Reference existing types, functions, and module structures by their actual names as shown in the provided code.
- When the provided context is insufficient to make a safe change, request the missing context explicitly.
---
### 10. Response Format
#### Language Policy
- Code, comments, commit messages, documentation: **English**.
- Reasoning and explanations in response text: **Russian**.
#### Response Structure
Your response MUST consist of two sections:
**Section 1: `## Reasoning` (in Russian)**
- What needs to be done and why.
- Which files and modules are affected.
- Architectural decisions and their rationale.
- Potential risks or side effects.
**Section 2: `## Changes`**
- For each modified or created file: the filename on a separate line in backticks, followed by the code block.
- For files **under 200 lines**: return the full file with all changes applied.
- For files **over 200 lines**: return only the changed functions/blocks with at least 3 lines of surrounding context above and below. If the user requests the full file, provide it.
- New files: full file content.
- End with a suggested git commit message in English.
#### Reporting Out-of-Scope Issues
If during modification you discover issues outside the requested scope (potential bugs, unsafe code, architectural concerns, missing error handling, unused imports, dead code):
- Do not fix them silently.
- List them under `## ⚠️ Out-of-scope observations` at the end of your response.
- Include: file path, line/function context, brief description of the issue, and severity estimate.
#### Splitting Protocol
If the response exceeds the output limit:
1. End the current part with: **SPLIT: PART N — CONTINUE? (remaining: file_list)**
2. List the files that will be provided in subsequent parts.
3. Wait for user confirmation before continuing.
4. No single file may be split across parts.

35
Cargo.lock generated
View File

@ -591,6 +591,25 @@ dependencies = [
"wasip3", "wasip3",
] ]
[[package]]
name = "h2"
version = "0.4.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2f44da3a8150a6703ed5d34e164b875fd14c2cdab9af1252a9a1020bde2bdc54"
dependencies = [
"atomic-waker",
"bytes",
"fnv",
"futures-core",
"futures-sink",
"http",
"indexmap",
"slab",
"tokio",
"tokio-util",
"tracing",
]
[[package]] [[package]]
name = "half" name = "half"
version = "2.7.1" version = "2.7.1"
@ -694,6 +713,12 @@ version = "1.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"
[[package]]
name = "httpdate"
version = "1.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
[[package]] [[package]]
name = "hyper" name = "hyper"
version = "1.8.1" version = "1.8.1"
@ -704,9 +729,11 @@ dependencies = [
"bytes", "bytes",
"futures-channel", "futures-channel",
"futures-core", "futures-core",
"h2",
"http", "http",
"http-body", "http-body",
"httparse", "httparse",
"httpdate",
"itoa", "itoa",
"pin-project-lite", "pin-project-lite",
"pin-utils", "pin-utils",
@ -1757,6 +1784,10 @@ dependencies = [
"futures", "futures",
"hex", "hex",
"hmac", "hmac",
"http-body-util",
"httpdate",
"hyper",
"hyper-util",
"libc", "libc",
"lru", "lru",
"md-5", "md-5",
@ -1929,8 +1960,12 @@ checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098"
dependencies = [ dependencies = [
"bytes", "bytes",
"futures-core", "futures-core",
"futures-io",
"futures-sink", "futures-sink",
"futures-util",
"hashbrown 0.15.5",
"pin-project-lite", "pin-project-lite",
"slab",
"tokio", "tokio",
] ]

View File

@ -1,6 +1,6 @@
[package] [package]
name = "telemt" name = "telemt"
version = "3.0.0" version = "3.0.3"
edition = "2024" edition = "2024"
[dependencies] [dependencies]

View File

@ -210,11 +210,18 @@ then Ctrl+X -> Y -> Enter to save
```toml ```toml
# === General Settings === # === General Settings ===
[general] [general]
# prefer_ipv6 is deprecated; use [network].prefer
prefer_ipv6 = false prefer_ipv6 = false
fast_mode = true fast_mode = true
use_middle_proxy = false use_middle_proxy = false
# ad_tag = "..." # ad_tag = "..."
[network]
ipv4 = true
ipv6 = true # set false to disable, omit for auto
prefer = 4 # 4 or 6
multipath = false
[general.modes] [general.modes]
classic = false classic = false
secure = false secure = false

View File

@ -1,10 +1,19 @@
# === General Settings === # === General Settings ===
[general] [general]
prefer_ipv6 = true # prefer_ipv6 is deprecated; use [network].prefer instead
prefer_ipv6 = false
fast_mode = true fast_mode = true
use_middle_proxy = true use_middle_proxy = true
#ad_tag = "00000000000000000000000000000000" #ad_tag = "00000000000000000000000000000000"
[network]
# Enable/disable families; ipv6 = true/false/auto(None)
ipv4 = true
ipv6 = true
# prefer = 4 or 6
prefer = 4
multipath = false
# Log level: debug | verbose | normal | silent # Log level: debug | verbose | normal | silent
# Can be overridden with --silent or --log-level CLI flags # Can be overridden with --silent or --log-level CLI flags
# RUST_LOG env var takes absolute priority over all of these # RUST_LOG env var takes absolute priority over all of these

View File

@ -189,11 +189,18 @@ r#"# Telemt MTProxy — auto-generated config
show_link = ["{username}"] show_link = ["{username}"]
[general] [general]
# prefer_ipv6 is deprecated; use [network].prefer
prefer_ipv6 = false prefer_ipv6 = false
fast_mode = true fast_mode = true
use_middle_proxy = false use_middle_proxy = false
log_level = "normal" log_level = "normal"
[network]
ipv4 = true
ipv6 = true
prefer = 4
multipath = false
[general.modes] [general.modes]
classic = false classic = false
secure = false secure = false

View File

@ -54,6 +54,10 @@ fn default_metrics_whitelist() -> Vec<IpAddr> {
vec!["127.0.0.1".parse().unwrap(), "::1".parse().unwrap()] vec!["127.0.0.1".parse().unwrap(), "::1".parse().unwrap()]
} }
fn default_prefer_4() -> u8 {
4
}
fn default_unknown_dc_log_path() -> Option<String> { fn default_unknown_dc_log_path() -> Option<String> {
Some("unknown-dc.txt".to_string()) Some("unknown-dc.txt".to_string())
} }
@ -185,6 +189,32 @@ impl std::fmt::Display for LogLevel {
} }
} }
fn validate_network_cfg(net: &mut NetworkConfig) -> Result<()> {
if !net.ipv4 && matches!(net.ipv6, Some(false)) {
return Err(ProxyError::Config(
"Both ipv4 and ipv6 are disabled in [network]".to_string(),
));
}
if net.prefer != 4 && net.prefer != 6 {
return Err(ProxyError::Config(
"network.prefer must be 4 or 6".to_string(),
));
}
if !net.ipv4 && net.prefer == 4 {
warn!("prefer=4 but ipv4=false; forcing prefer=6");
net.prefer = 6;
}
if matches!(net.ipv6, Some(false)) && net.prefer == 6 {
warn!("prefer=6 but ipv6=false; forcing prefer=4");
net.prefer = 4;
}
Ok(())
}
// ============= Sub-Configs ============= // ============= Sub-Configs =============
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
@ -207,6 +237,34 @@ impl Default for ProxyModes {
} }
} }
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NetworkConfig {
#[serde(default = "default_true")]
pub ipv4: bool,
/// None = auto-detect IPv6 availability
#[serde(default)]
pub ipv6: Option<bool>,
/// 4 or 6
#[serde(default = "default_prefer_4")]
pub prefer: u8,
#[serde(default)]
pub multipath: bool,
}
impl Default for NetworkConfig {
fn default() -> Self {
Self {
ipv4: true,
ipv6: None,
prefer: 4,
multipath: false,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GeneralConfig { pub struct GeneralConfig {
#[serde(default)] #[serde(default)]
@ -609,6 +667,9 @@ pub struct ProxyConfig {
#[serde(default)] #[serde(default)]
pub general: GeneralConfig, pub general: GeneralConfig,
#[serde(default)]
pub network: NetworkConfig,
#[serde(default)] #[serde(default)]
pub server: ServerConfig, pub server: ServerConfig,
@ -697,6 +758,16 @@ impl ProxyConfig {
config.censorship.mask_host = Some(config.censorship.tls_domain.clone()); config.censorship.mask_host = Some(config.censorship.tls_domain.clone());
} }
// Migration: prefer_ipv6 -> network.prefer
if config.general.prefer_ipv6 {
if config.network.prefer == 4 {
config.network.prefer = 6;
}
warn!("prefer_ipv6 is deprecated, use [network].prefer = 6");
}
validate_network_cfg(&mut config.network)?;
// Random fake_cert_len // Random fake_cert_len
use rand::Rng; use rand::Rng;
config.censorship.fake_cert_len = rand::rng().gen_range(1024..4096); config.censorship.fake_cert_len = rand::rng().gen_range(1024..4096);

View File

@ -16,6 +16,7 @@ mod config;
mod crypto; mod crypto;
mod error; mod error;
mod ip_tracker; mod ip_tracker;
mod network;
mod metrics; mod metrics;
mod protocol; mod protocol;
mod proxy; mod proxy;
@ -27,16 +28,14 @@ mod util;
use crate::config::{LogLevel, ProxyConfig}; use crate::config::{LogLevel, ProxyConfig};
use crate::crypto::SecureRandom; use crate::crypto::SecureRandom;
use crate::ip_tracker::UserIpTracker; use crate::ip_tracker::UserIpTracker;
use crate::network::probe::{decide_network_capabilities, log_probe_result, run_probe};
use crate::proxy::ClientHandler; use crate::proxy::ClientHandler;
use crate::stats::{ReplayChecker, Stats}; use crate::stats::{ReplayChecker, Stats};
use crate::stream::BufferPool; use crate::stream::BufferPool;
use crate::transport::middle_proxy::{ use crate::transport::middle_proxy::{
MePool, fetch_proxy_config, run_me_ping, MePingFamily, MePingSample, format_sample_line, MePool, fetch_proxy_config, run_me_ping, MePingFamily, MePingSample, format_sample_line,
stun_probe,
}; };
use crate::transport::{ListenOptions, UpstreamManager, create_listener}; use crate::transport::{ListenOptions, UpstreamManager, create_listener};
use crate::util::ip::detect_ip;
use crate::protocol::constants::{TG_MIDDLE_PROXIES_V4, TG_MIDDLE_PROXIES_V6};
fn parse_cli() -> (String, bool, Option<String>) { fn parse_cli() -> (String, bool, Option<String>) {
let mut config_path = "config.toml".to_string(); let mut config_path = "config.toml".to_string();
@ -219,8 +218,17 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
warn!("Using default tls_domain. Consider setting a custom domain."); warn!("Using default tls_domain. Consider setting a custom domain.");
} }
let prefer_ipv6 = config.general.prefer_ipv6; let probe = run_probe(
let mut use_middle_proxy = config.general.use_middle_proxy; &config.network,
config.general.middle_proxy_nat_stun.clone(),
config.general.middle_proxy_nat_probe,
)
.await?;
let decision = decide_network_capabilities(&config.network, &probe);
log_probe_result(&probe, &decision);
let prefer_ipv6 = decision.prefer_ipv6();
let mut use_middle_proxy = config.general.use_middle_proxy && (decision.ipv4_me || decision.ipv6_me);
let config = Arc::new(config); let config = Arc::new(config);
let stats = Arc::new(Stats::new()); let stats = Arc::new(Stats::new());
let rng = Arc::new(SecureRandom::new()); let rng = Arc::new(SecureRandom::new());
@ -244,40 +252,10 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
// Connection concurrency limit // Connection concurrency limit
let _max_connections = Arc::new(Semaphore::new(10_000)); let _max_connections = Arc::new(Semaphore::new(10_000));
// STUN check before choosing transport if use_middle_proxy && !decision.ipv4_me && !decision.ipv6_me {
if use_middle_proxy { warn!("No usable IP family for Middle Proxy detected; falling back to direct DC");
match stun_probe(config.general.middle_proxy_nat_stun.clone()).await {
Ok(Some(probe)) => {
info!(
local_ip = %probe.local_addr.ip(),
reflected_ip = %probe.reflected_addr.ip(),
"STUN Autodetect:"
);
if probe.local_addr.ip() != probe.reflected_addr.ip()
&& !config.general.stun_iface_mismatch_ignore
{
match crate::transport::middle_proxy::detect_public_ip().await {
Some(ip) => {
info!(
local_ip = %probe.local_addr.ip(),
reflected_ip = %probe.reflected_addr.ip(),
public_ip = %ip,
"STUN mismatch but public IP auto-detected, continuing with middle proxy"
);
}
None => {
warn!(
"STUN/IP-on-Interface mismatch and public IP auto-detect failed -> fallback to direct-DC"
);
use_middle_proxy = false; use_middle_proxy = false;
} }
}
}
}
Ok(None) => warn!("STUN probe returned no address; continuing"),
Err(e) => warn!(error = %e, "STUN probe failed; continuing"),
}
}
// ===================================================================== // =====================================================================
// Middle Proxy initialization (if enabled) // Middle Proxy initialization (if enabled)
@ -351,6 +329,8 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
cfg_v4.map.clone(), cfg_v4.map.clone(),
cfg_v6.map.clone(), cfg_v6.map.clone(),
cfg_v4.default_dc.or(cfg_v6.default_dc), cfg_v4.default_dc.or(cfg_v6.default_dc),
decision.clone(),
rng.clone(),
); );
match pool.init(2, &rng).await { match pool.init(2, &rng).await {
@ -482,7 +462,12 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
info!("================= Telegram DC Connectivity ================="); info!("================= Telegram DC Connectivity =================");
let ping_results = upstream_manager let ping_results = upstream_manager
.ping_all_dcs(prefer_ipv6, &config.dc_overrides) .ping_all_dcs(
prefer_ipv6,
&config.dc_overrides,
decision.ipv4_dc,
decision.ipv6_dc,
)
.await; .await;
for upstream_result in &ping_results { for upstream_result in &ping_results {
@ -559,8 +544,15 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
// Background tasks // Background tasks
let um_clone = upstream_manager.clone(); let um_clone = upstream_manager.clone();
let decision_clone = decision.clone();
tokio::spawn(async move { tokio::spawn(async move {
um_clone.run_health_checks(prefer_ipv6).await; um_clone
.run_health_checks(
prefer_ipv6,
decision_clone.ipv4_dc,
decision_clone.ipv6_dc,
)
.await;
}); });
let rc_clone = replay_checker.clone(); let rc_clone = replay_checker.clone();
@ -568,16 +560,31 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
rc_clone.run_periodic_cleanup().await; rc_clone.run_periodic_cleanup().await;
}); });
let detected_ip = detect_ip().await; let detected_ip_v4: Option<std::net::IpAddr> = probe
.reflected_ipv4
.map(|s| s.ip())
.or_else(|| probe.detected_ipv4.map(std::net::IpAddr::V4));
let detected_ip_v6: Option<std::net::IpAddr> = probe
.reflected_ipv6
.map(|s| s.ip())
.or_else(|| probe.detected_ipv6.map(std::net::IpAddr::V6));
debug!( debug!(
"Detected IPs: v4={:?} v6={:?}", "Detected IPs: v4={:?} v6={:?}",
detected_ip.ipv4, detected_ip.ipv6 detected_ip_v4, detected_ip_v6
); );
let mut listeners = Vec::new(); let mut listeners = Vec::new();
for listener_conf in &config.server.listeners { for listener_conf in &config.server.listeners {
let addr = SocketAddr::new(listener_conf.ip, config.server.port); let addr = SocketAddr::new(listener_conf.ip, config.server.port);
if addr.is_ipv4() && !decision.ipv4_dc {
warn!(%addr, "Skipping IPv4 listener: IPv4 disabled by [network]");
continue;
}
if addr.is_ipv6() && !decision.ipv6_dc {
warn!(%addr, "Skipping IPv6 listener: IPv6 disabled by [network]");
continue;
}
let options = ListenOptions { let options = ListenOptions {
ipv6_only: listener_conf.ip.is_ipv6(), ipv6_only: listener_conf.ip.is_ipv6(),
..Default::default() ..Default::default()
@ -594,11 +601,11 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
} else if listener_conf.ip.is_unspecified() { } else if listener_conf.ip.is_unspecified() {
// Auto-detect for unspecified addresses // Auto-detect for unspecified addresses
if listener_conf.ip.is_ipv4() { if listener_conf.ip.is_ipv4() {
detected_ip.ipv4 detected_ip_v4
.map(|ip| ip.to_string()) .map(|ip| ip.to_string())
.unwrap_or_else(|| listener_conf.ip.to_string()) .unwrap_or_else(|| listener_conf.ip.to_string())
} else { } else {
detected_ip.ipv6 detected_ip_v6
.map(|ip| ip.to_string()) .map(|ip| ip.to_string())
.unwrap_or_else(|| listener_conf.ip.to_string()) .unwrap_or_else(|| listener_conf.ip.to_string())
} }
@ -626,9 +633,8 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
let (host, port) = if let Some(ref h) = config.general.links.public_host { let (host, port) = if let Some(ref h) = config.general.links.public_host {
(h.clone(), config.general.links.public_port.unwrap_or(config.server.port)) (h.clone(), config.general.links.public_port.unwrap_or(config.server.port))
} else { } else {
let ip = detected_ip let ip = detected_ip_v4
.ipv4 .or(detected_ip_v6)
.or(detected_ip.ipv6)
.map(|ip| ip.to_string()); .map(|ip| ip.to_string());
if ip.is_none() { if ip.is_none() {
warn!("show_link is configured but public IP could not be detected. Set public_host in config."); warn!("show_link is configured but public IP could not be detected. Set public_host in config.");

4
src/network/mod.rs Normal file
View File

@ -0,0 +1,4 @@
pub mod probe;
pub mod stun;
pub use stun::IpFamily;

225
src/network/probe.rs Normal file
View File

@ -0,0 +1,225 @@
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, UdpSocket};
use tracing::{info, warn};
use crate::config::NetworkConfig;
use crate::error::Result;
use crate::network::stun::{stun_probe_dual, DualStunResult, IpFamily};
#[derive(Debug, Clone, Default)]
pub struct NetworkProbe {
pub detected_ipv4: Option<Ipv4Addr>,
pub detected_ipv6: Option<Ipv6Addr>,
pub reflected_ipv4: Option<SocketAddr>,
pub reflected_ipv6: Option<SocketAddr>,
pub ipv4_is_bogon: bool,
pub ipv6_is_bogon: bool,
pub ipv4_nat_detected: bool,
pub ipv6_nat_detected: bool,
pub ipv4_usable: bool,
pub ipv6_usable: bool,
}
#[derive(Debug, Clone, Default)]
pub struct NetworkDecision {
pub ipv4_dc: bool,
pub ipv6_dc: bool,
pub ipv4_me: bool,
pub ipv6_me: bool,
pub effective_prefer: u8,
pub effective_multipath: bool,
}
impl NetworkDecision {
pub fn prefer_ipv6(&self) -> bool {
self.effective_prefer == 6
}
pub fn me_families(&self) -> Vec<IpFamily> {
let mut res = Vec::new();
if self.ipv4_me {
res.push(IpFamily::V4);
}
if self.ipv6_me {
res.push(IpFamily::V6);
}
res
}
}
pub async fn run_probe(config: &NetworkConfig, stun_addr: Option<String>, nat_probe: bool) -> Result<NetworkProbe> {
let mut probe = NetworkProbe::default();
probe.detected_ipv4 = detect_local_ip_v4();
probe.detected_ipv6 = detect_local_ip_v6();
probe.ipv4_is_bogon = probe.detected_ipv4.map(is_bogon_v4).unwrap_or(false);
probe.ipv6_is_bogon = probe.detected_ipv6.map(is_bogon_v6).unwrap_or(false);
let stun_server = stun_addr.unwrap_or_else(|| "stun.l.google.com:19302".to_string());
let stun_res = if nat_probe {
stun_probe_dual(&stun_server).await?
} else {
DualStunResult::default()
};
probe.reflected_ipv4 = stun_res.v4.map(|r| r.reflected_addr);
probe.reflected_ipv6 = stun_res.v6.map(|r| r.reflected_addr);
probe.ipv4_nat_detected = match (probe.detected_ipv4, probe.reflected_ipv4) {
(Some(det), Some(reflected)) => det != reflected.ip(),
_ => false,
};
probe.ipv6_nat_detected = match (probe.detected_ipv6, probe.reflected_ipv6) {
(Some(det), Some(reflected)) => det != reflected.ip(),
_ => false,
};
probe.ipv4_usable = config.ipv4
&& probe.detected_ipv4.is_some()
&& (!probe.ipv4_is_bogon || probe.reflected_ipv4.map(|r| !is_bogon(r.ip())).unwrap_or(false));
let ipv6_enabled = config.ipv6.unwrap_or(probe.detected_ipv6.is_some());
probe.ipv6_usable = ipv6_enabled
&& probe.detected_ipv6.is_some()
&& (!probe.ipv6_is_bogon || probe.reflected_ipv6.map(|r| !is_bogon(r.ip())).unwrap_or(false));
Ok(probe)
}
pub fn decide_network_capabilities(config: &NetworkConfig, probe: &NetworkProbe) -> NetworkDecision {
let mut decision = NetworkDecision::default();
decision.ipv4_dc = config.ipv4 && probe.detected_ipv4.is_some();
decision.ipv6_dc = config.ipv6.unwrap_or(probe.detected_ipv6.is_some()) && probe.detected_ipv6.is_some();
decision.ipv4_me = config.ipv4
&& probe.detected_ipv4.is_some()
&& (!probe.ipv4_is_bogon || probe.reflected_ipv4.is_some());
let ipv6_enabled = config.ipv6.unwrap_or(probe.detected_ipv6.is_some());
decision.ipv6_me = ipv6_enabled
&& probe.detected_ipv6.is_some()
&& (!probe.ipv6_is_bogon || probe.reflected_ipv6.is_some());
decision.effective_prefer = match config.prefer {
6 if decision.ipv6_me || decision.ipv6_dc => 6,
4 if decision.ipv4_me || decision.ipv4_dc => 4,
6 => {
warn!("prefer=6 requested but IPv6 unavailable; falling back to IPv4");
4
}
_ => 4,
};
let me_families = decision.ipv4_me as u8 + decision.ipv6_me as u8;
decision.effective_multipath = config.multipath && me_families >= 2;
decision
}
fn detect_local_ip_v4() -> Option<Ipv4Addr> {
let socket = UdpSocket::bind("0.0.0.0:0").ok()?;
socket.connect("8.8.8.8:80").ok()?;
match socket.local_addr().ok()?.ip() {
IpAddr::V4(v4) => Some(v4),
_ => None,
}
}
fn detect_local_ip_v6() -> Option<Ipv6Addr> {
let socket = UdpSocket::bind("[::]:0").ok()?;
socket.connect("[2001:4860:4860::8888]:80").ok()?;
match socket.local_addr().ok()?.ip() {
IpAddr::V6(v6) => Some(v6),
_ => None,
}
}
pub fn is_bogon(ip: IpAddr) -> bool {
match ip {
IpAddr::V4(v4) => is_bogon_v4(v4),
IpAddr::V6(v6) => is_bogon_v6(v6),
}
}
pub fn is_bogon_v4(ip: Ipv4Addr) -> bool {
let octets = ip.octets();
if ip.is_private() || ip.is_loopback() || ip.is_link_local() {
return true;
}
if octets[0] == 0 {
return true;
}
if octets[0] == 100 && (octets[1] & 0xC0) == 64 {
return true;
}
if octets[0] == 192 && octets[1] == 0 && octets[2] == 0 {
return true;
}
if octets[0] == 192 && octets[1] == 0 && octets[2] == 2 {
return true;
}
if octets[0] == 198 && (octets[1] & 0xFE) == 18 {
return true;
}
if octets[0] == 198 && octets[1] == 51 && octets[2] == 100 {
return true;
}
if octets[0] == 203 && octets[1] == 0 && octets[2] == 113 {
return true;
}
if ip.is_multicast() {
return true;
}
if octets[0] >= 240 {
return true;
}
if ip.is_broadcast() {
return true;
}
false
}
pub fn is_bogon_v6(ip: Ipv6Addr) -> bool {
if ip.is_unspecified() || ip.is_loopback() || ip.is_unique_local() {
return true;
}
let segs = ip.segments();
if (segs[0] & 0xFFC0) == 0xFE80 {
return true;
}
if segs[0..5] == [0, 0, 0, 0, 0] && segs[5] == 0xFFFF {
return true;
}
if segs[0] == 0x0100 && segs[1..4] == [0, 0, 0] {
return true;
}
if segs[0] == 0x2001 && segs[1] == 0x0db8 {
return true;
}
if segs[0] == 0x2002 {
return true;
}
if ip.is_multicast() {
return true;
}
false
}
pub fn log_probe_result(probe: &NetworkProbe, decision: &NetworkDecision) {
info!(
ipv4 = probe.detected_ipv4.as_ref().map(|v| v.to_string()).unwrap_or_else(|| "-".into()),
ipv6 = probe.detected_ipv6.as_ref().map(|v| v.to_string()).unwrap_or_else(|| "-".into()),
reflected_v4 = probe.reflected_ipv4.as_ref().map(|v| v.ip().to_string()).unwrap_or_else(|| "-".into()),
reflected_v6 = probe.reflected_ipv6.as_ref().map(|v| v.ip().to_string()).unwrap_or_else(|| "-".into()),
ipv4_bogon = probe.ipv4_is_bogon,
ipv6_bogon = probe.ipv6_is_bogon,
ipv4_me = decision.ipv4_me,
ipv6_me = decision.ipv6_me,
ipv4_dc = decision.ipv4_dc,
ipv6_dc = decision.ipv6_dc,
prefer = decision.effective_prefer,
multipath = decision.effective_multipath,
"Network capabilities resolved"
);
}

186
src/network/stun.rs Normal file
View File

@ -0,0 +1,186 @@
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use tokio::net::{lookup_host, UdpSocket};
use crate::error::{ProxyError, Result};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum IpFamily {
V4,
V6,
}
#[derive(Debug, Clone, Copy)]
pub struct StunProbeResult {
pub local_addr: SocketAddr,
pub reflected_addr: SocketAddr,
pub family: IpFamily,
}
#[derive(Debug, Default, Clone)]
pub struct DualStunResult {
pub v4: Option<StunProbeResult>,
pub v6: Option<StunProbeResult>,
}
pub async fn stun_probe_dual(stun_addr: &str) -> Result<DualStunResult> {
let (v4, v6) = tokio::join!(
stun_probe_family(stun_addr, IpFamily::V4),
stun_probe_family(stun_addr, IpFamily::V6),
);
Ok(DualStunResult {
v4: v4?,
v6: v6?,
})
}
pub async fn stun_probe_family(stun_addr: &str, family: IpFamily) -> Result<Option<StunProbeResult>> {
use rand::RngCore;
let bind_addr = match family {
IpFamily::V4 => "0.0.0.0:0",
IpFamily::V6 => "[::]:0",
};
let socket = UdpSocket::bind(bind_addr)
.await
.map_err(|e| ProxyError::Proxy(format!("STUN bind failed: {e}")))?;
let target_addr = resolve_stun_addr(stun_addr, family).await?;
if let Some(addr) = target_addr {
socket
.connect(addr)
.await
.map_err(|e| ProxyError::Proxy(format!("STUN connect failed: {e}")))?;
} else {
return Ok(None);
}
let mut req = [0u8; 20];
req[0..2].copy_from_slice(&0x0001u16.to_be_bytes()); // Binding Request
req[2..4].copy_from_slice(&0u16.to_be_bytes()); // length
req[4..8].copy_from_slice(&0x2112A442u32.to_be_bytes()); // magic cookie
rand::rng().fill_bytes(&mut req[8..20]); // transaction ID
socket
.send(&req)
.await
.map_err(|e| ProxyError::Proxy(format!("STUN send failed: {e}")))?;
let mut buf = [0u8; 256];
let n = socket
.recv(&mut buf)
.await
.map_err(|e| ProxyError::Proxy(format!("STUN recv failed: {e}")))?;
if n < 20 {
return Ok(None);
}
let magic = 0x2112A442u32.to_be_bytes();
let txid = &req[8..20];
let mut idx = 20;
while idx + 4 <= n {
let atype = u16::from_be_bytes(buf[idx..idx + 2].try_into().unwrap());
let alen = u16::from_be_bytes(buf[idx + 2..idx + 4].try_into().unwrap()) as usize;
idx += 4;
if idx + alen > n {
break;
}
match atype {
0x0020 /* XOR-MAPPED-ADDRESS */ | 0x0001 /* MAPPED-ADDRESS */ => {
if alen < 8 {
break;
}
let family_byte = buf[idx + 1];
let port_bytes = [buf[idx + 2], buf[idx + 3]];
let len_check = match family_byte {
0x01 => 4,
0x02 => 16,
_ => 0,
};
if len_check == 0 || alen < 4 + len_check {
break;
}
let raw_ip = &buf[idx + 4..idx + 4 + len_check];
let mut port = u16::from_be_bytes(port_bytes);
let reflected_ip = if atype == 0x0020 {
port ^= ((magic[0] as u16) << 8) | magic[1] as u16;
match family_byte {
0x01 => {
let ip = [
raw_ip[0] ^ magic[0],
raw_ip[1] ^ magic[1],
raw_ip[2] ^ magic[2],
raw_ip[3] ^ magic[3],
];
IpAddr::V4(Ipv4Addr::new(ip[0], ip[1], ip[2], ip[3]))
}
0x02 => {
let mut ip = [0u8; 16];
let xor_key = [magic.as_slice(), txid].concat();
for (i, b) in raw_ip.iter().enumerate().take(16) {
ip[i] = *b ^ xor_key[i];
}
IpAddr::V6(Ipv6Addr::from(ip))
}
_ => {
idx += (alen + 3) & !3;
continue;
}
}
} else {
match family_byte {
0x01 => IpAddr::V4(Ipv4Addr::new(raw_ip[0], raw_ip[1], raw_ip[2], raw_ip[3])),
0x02 => IpAddr::V6(Ipv6Addr::from(<[u8; 16]>::try_from(raw_ip).unwrap())),
_ => {
idx += (alen + 3) & !3;
continue;
}
}
};
let reflected_addr = SocketAddr::new(reflected_ip, port);
let local_addr = socket
.local_addr()
.map_err(|e| ProxyError::Proxy(format!("STUN local_addr failed: {e}")))?;
return Ok(Some(StunProbeResult {
local_addr,
reflected_addr,
family,
}));
}
_ => {}
}
idx += (alen + 3) & !3;
}
Ok(None)
}
async fn resolve_stun_addr(stun_addr: &str, family: IpFamily) -> Result<Option<SocketAddr>> {
if let Ok(addr) = stun_addr.parse::<SocketAddr>() {
return Ok(match (addr.is_ipv4(), family) {
(true, IpFamily::V4) | (false, IpFamily::V6) => Some(addr),
_ => None,
});
}
let addrs = lookup_host(stun_addr)
.await
.map_err(|e| ProxyError::Proxy(format!("STUN resolve failed: {e}")))?;
let target = addrs
.filter(|a| match (a.is_ipv4(), family) {
(true, IpFamily::V4) => true,
(false, IpFamily::V6) => true,
_ => false,
})
.next();
Ok(target)
}

View File

@ -80,7 +80,8 @@ where
} }
fn get_dc_addr_static(dc_idx: i16, config: &ProxyConfig) -> Result<SocketAddr> { fn get_dc_addr_static(dc_idx: i16, config: &ProxyConfig) -> Result<SocketAddr> {
let datacenters = if config.general.prefer_ipv6 { let prefer_v6 = config.network.prefer == 6 && config.network.ipv6.unwrap_or(true);
let datacenters = if prefer_v6 {
&*TG_DATACENTERS_V6 &*TG_DATACENTERS_V6
} else { } else {
&*TG_DATACENTERS_V4 &*TG_DATACENTERS_V4
@ -90,7 +91,6 @@ fn get_dc_addr_static(dc_idx: i16, config: &ProxyConfig) -> Result<SocketAddr> {
let dc_key = dc_idx.to_string(); let dc_key = dc_idx.to_string();
if let Some(addrs) = config.dc_overrides.get(&dc_key) { if let Some(addrs) = config.dc_overrides.get(&dc_key) {
let prefer_v6 = config.general.prefer_ipv6;
let mut parsed = Vec::new(); let mut parsed = Vec::new();
for addr_str in addrs { for addr_str in addrs {
match addr_str.parse::<SocketAddr>() { match addr_str.parse::<SocketAddr>() {

View File

@ -16,6 +16,7 @@ use tracing::{debug, info, warn};
use crate::crypto::{SecureRandom, build_middleproxy_prekey, derive_middleproxy_keys, sha256}; use crate::crypto::{SecureRandom, build_middleproxy_prekey, derive_middleproxy_keys, sha256};
use crate::error::{ProxyError, Result}; use crate::error::{ProxyError, Result};
use crate::network::IpFamily;
use crate::protocol::constants::{ use crate::protocol::constants::{
ME_CONNECT_TIMEOUT_SECS, ME_HANDSHAKE_TIMEOUT_SECS, RPC_CRYPTO_AES_U32, RPC_HANDSHAKE_ERROR_U32, ME_CONNECT_TIMEOUT_SECS, ME_HANDSHAKE_TIMEOUT_SECS, RPC_CRYPTO_AES_U32, RPC_HANDSHAKE_ERROR_U32,
RPC_HANDSHAKE_U32, RPC_PING_U32, RPC_PONG_U32, RPC_NONCE_U32, RPC_HANDSHAKE_U32, RPC_PING_U32, RPC_PONG_U32, RPC_NONCE_U32,
@ -101,8 +102,13 @@ impl MePool {
let peer_addr = stream.peer_addr().map_err(ProxyError::Io)?; let peer_addr = stream.peer_addr().map_err(ProxyError::Io)?;
let _ = self.maybe_detect_nat_ip(local_addr.ip()).await; let _ = self.maybe_detect_nat_ip(local_addr.ip()).await;
let family = if local_addr.ip().is_ipv4() {
IpFamily::V4
} else {
IpFamily::V6
};
let reflected = if self.nat_probe { let reflected = if self.nat_probe {
self.maybe_reflect_public_addr().await self.maybe_reflect_public_addr(family).await
} else { } else {
None None
}; };

View File

@ -1,4 +1,4 @@
use std::collections::HashMap; use std::collections::{HashMap, HashSet};
use std::net::SocketAddr; use std::net::SocketAddr;
use std::sync::Arc; use std::sync::Arc;
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
@ -7,17 +7,40 @@ use tracing::{debug, info, warn};
use rand::seq::SliceRandom; use rand::seq::SliceRandom;
use crate::crypto::SecureRandom; use crate::crypto::SecureRandom;
use crate::network::IpFamily;
use super::MePool; use super::MePool;
pub async fn me_health_monitor(pool: Arc<MePool>, rng: Arc<SecureRandom>, _min_connections: usize) { pub async fn me_health_monitor(pool: Arc<MePool>, rng: Arc<SecureRandom>, _min_connections: usize) {
let mut backoff: HashMap<i32, u64> = HashMap::new(); let mut backoff: HashMap<(i32, IpFamily), u64> = HashMap::new();
let mut last_attempt: HashMap<i32, Instant> = HashMap::new(); let mut last_attempt: HashMap<(i32, IpFamily), Instant> = HashMap::new();
loop { loop {
tokio::time::sleep(Duration::from_secs(30)).await; tokio::time::sleep(Duration::from_secs(30)).await;
// Per-DC coverage check check_family(IpFamily::V4, &pool, &rng, &mut backoff, &mut last_attempt).await;
let map = pool.proxy_map_v4.read().await.clone(); check_family(IpFamily::V6, &pool, &rng, &mut backoff, &mut last_attempt).await;
let writer_addrs: std::collections::HashSet<SocketAddr> = pool }
}
async fn check_family(
family: IpFamily,
pool: &Arc<MePool>,
rng: &Arc<SecureRandom>,
backoff: &mut HashMap<(i32, IpFamily), u64>,
last_attempt: &mut HashMap<(i32, IpFamily), Instant>,
) {
let enabled = match family {
IpFamily::V4 => pool.decision.ipv4_me,
IpFamily::V6 => pool.decision.ipv6_me,
};
if !enabled {
return;
}
let map = match family {
IpFamily::V4 => pool.proxy_map_v4.read().await.clone(),
IpFamily::V6 => pool.proxy_map_v6.read().await.clone(),
};
let writer_addrs: HashSet<SocketAddr> = pool
.writers .writers
.read() .read()
.await .await
@ -31,83 +54,37 @@ pub async fn me_health_monitor(pool: Arc<MePool>, rng: Arc<SecureRandom>, _min_c
.map(|(ip, port)| SocketAddr::new(*ip, *port)) .map(|(ip, port)| SocketAddr::new(*ip, *port))
.collect(); .collect();
let has_coverage = dc_addrs.iter().any(|a| writer_addrs.contains(a)); let has_coverage = dc_addrs.iter().any(|a| writer_addrs.contains(a));
if !has_coverage { if has_coverage {
let delay = *backoff.get(dc).unwrap_or(&30); continue;
}
let key = (*dc, family);
let delay = *backoff.get(&key).unwrap_or(&30);
let now = Instant::now(); let now = Instant::now();
if let Some(last) = last_attempt.get(dc) { if let Some(last) = last_attempt.get(&key) {
if now.duration_since(*last).as_secs() < delay { if now.duration_since(*last).as_secs() < delay {
continue; continue;
} }
} }
warn!(dc = %dc, delay, "DC has no ME coverage, reconnecting..."); warn!(dc = %dc, delay, ?family, "DC has no ME coverage, reconnecting...");
let mut shuffled = dc_addrs.clone(); let mut shuffled = dc_addrs.clone();
shuffled.shuffle(&mut rand::rng()); shuffled.shuffle(&mut rand::rng());
let mut reconnected = false; let mut reconnected = false;
for addr in shuffled { for addr in shuffled {
match pool.connect_one(addr, &rng).await { match pool.connect_one(addr, rng.as_ref()).await {
Ok(()) => { Ok(()) => {
info!(%addr, dc = %dc, "ME reconnected for DC coverage"); info!(%addr, dc = %dc, ?family, "ME reconnected for DC coverage");
backoff.insert(*dc, 30); backoff.insert(key, 30);
last_attempt.insert(*dc, now); last_attempt.insert(key, now);
reconnected = true; reconnected = true;
break; break;
} }
Err(e) => debug!(%addr, dc = %dc, error = %e, "ME reconnect failed"), Err(e) => debug!(%addr, dc = %dc, error = %e, ?family, "ME reconnect failed"),
} }
} }
if !reconnected { if !reconnected {
let next = (*backoff.get(dc).unwrap_or(&30)).saturating_mul(2).min(300); let next = (*backoff.get(&key).unwrap_or(&30)).saturating_mul(2).min(300);
backoff.insert(*dc, next); backoff.insert(key, next);
last_attempt.insert(*dc, now); last_attempt.insert(key, now);
}
}
}
// IPv6 coverage check (if available)
let map_v6 = pool.proxy_map_v6.read().await.clone();
let writer_addrs_v6: std::collections::HashSet<SocketAddr> = pool
.writers
.read()
.await
.iter()
.map(|w| w.addr)
.collect();
for (dc, addrs) in map_v6.iter() {
let dc_addrs: Vec<SocketAddr> = addrs
.iter()
.map(|(ip, port)| SocketAddr::new(*ip, *port))
.collect();
let has_coverage = dc_addrs.iter().any(|a| writer_addrs_v6.contains(a));
if !has_coverage {
let delay = *backoff.get(dc).unwrap_or(&30);
let now = Instant::now();
if let Some(last) = last_attempt.get(dc) {
if now.duration_since(*last).as_secs() < delay {
continue;
}
}
warn!(dc = %dc, delay, "IPv6 DC has no ME coverage, reconnecting...");
let mut shuffled = dc_addrs.clone();
shuffled.shuffle(&mut rand::rng());
let mut reconnected = false;
for addr in shuffled {
match pool.connect_one(addr, &rng).await {
Ok(()) => {
info!(%addr, dc = %dc, "ME reconnected for IPv6 DC coverage");
backoff.insert(*dc, 30);
last_attempt.insert(*dc, now);
reconnected = true;
break;
}
Err(e) => debug!(%addr, dc = %dc, error = %e, "ME reconnect failed (IPv6)"),
}
}
if !reconnected {
let next = (*backoff.get(dc).unwrap_or(&30)).saturating_mul(2).min(300);
backoff.insert(*dc, next);
last_attempt.insert(*dc, now);
}
}
} }
} }
} }

View File

@ -19,7 +19,7 @@ use bytes::Bytes;
pub use health::me_health_monitor; pub use health::me_health_monitor;
pub use ping::{run_me_ping, format_sample_line, MePingReport, MePingSample, MePingFamily}; pub use ping::{run_me_ping, format_sample_line, MePingReport, MePingSample, MePingFamily};
pub use pool::MePool; pub use pool::MePool;
pub use pool_nat::{stun_probe, detect_public_ip, StunProbeResult}; pub use pool_nat::{stun_probe, detect_public_ip};
pub use registry::ConnRegistry; pub use registry::ConnRegistry;
pub use secret::fetch_proxy_secret; pub use secret::fetch_proxy_secret;
pub use config_updater::{fetch_proxy_config, me_config_updater}; pub use config_updater::{fetch_proxy_config, me_config_updater};

View File

@ -1,3 +1,4 @@
use std::collections::HashMap;
use std::net::{IpAddr, SocketAddr}; use std::net::{IpAddr, SocketAddr};
use std::sync::Arc; use std::sync::Arc;
@ -92,8 +93,16 @@ mod tests {
pub async fn run_me_ping(pool: &Arc<MePool>, rng: &SecureRandom) -> Vec<MePingReport> { pub async fn run_me_ping(pool: &Arc<MePool>, rng: &SecureRandom) -> Vec<MePingReport> {
let mut reports = Vec::new(); let mut reports = Vec::new();
let v4_map = pool.proxy_map_v4.read().await.clone(); let v4_map = if pool.decision.ipv4_me {
let v6_map = pool.proxy_map_v6.read().await.clone(); pool.proxy_map_v4.read().await.clone()
} else {
HashMap::new()
};
let v6_map = if pool.decision.ipv6_me {
pool.proxy_map_v6.read().await.clone()
} else {
HashMap::new()
};
let mut grouped: Vec<(MePingFamily, i32, Vec<(IpAddr, u16)>)> = Vec::new(); let mut grouped: Vec<(MePingFamily, i32, Vec<(IpAddr, u16)>)> = Vec::new();
for (dc, addrs) in v4_map { for (dc, addrs) in v4_map {

View File

@ -12,6 +12,8 @@ use std::time::Duration;
use crate::crypto::SecureRandom; use crate::crypto::SecureRandom;
use crate::error::{ProxyError, Result}; use crate::error::{ProxyError, Result};
use crate::network::probe::NetworkDecision;
use crate::network::IpFamily;
use crate::protocol::constants::*; use crate::protocol::constants::*;
use super::ConnRegistry; use super::ConnRegistry;
@ -36,6 +38,8 @@ pub struct MePool {
pub(super) registry: Arc<ConnRegistry>, pub(super) registry: Arc<ConnRegistry>,
pub(super) writers: Arc<RwLock<Vec<MeWriter>>>, pub(super) writers: Arc<RwLock<Vec<MeWriter>>>,
pub(super) rr: AtomicU64, pub(super) rr: AtomicU64,
pub(super) decision: NetworkDecision,
pub(super) rng: Arc<SecureRandom>,
pub(super) proxy_tag: Option<Vec<u8>>, pub(super) proxy_tag: Option<Vec<u8>>,
pub(super) proxy_secret: Arc<RwLock<Vec<u8>>>, pub(super) proxy_secret: Arc<RwLock<Vec<u8>>>,
pub(super) nat_ip_cfg: Option<IpAddr>, pub(super) nat_ip_cfg: Option<IpAddr>,
@ -48,10 +52,16 @@ pub struct MePool {
pub(super) next_writer_id: AtomicU64, pub(super) next_writer_id: AtomicU64,
pub(super) ping_tracker: Arc<Mutex<HashMap<i64, (std::time::Instant, u64)>>>, pub(super) ping_tracker: Arc<Mutex<HashMap<i64, (std::time::Instant, u64)>>>,
pub(super) rtt_stats: Arc<Mutex<HashMap<u64, (f64, f64)>>>, pub(super) rtt_stats: Arc<Mutex<HashMap<u64, (f64, f64)>>>,
pub(super) nat_reflection_cache: Arc<Mutex<Option<(std::time::Instant, std::net::SocketAddr)>>>, pub(super) nat_reflection_cache: Arc<Mutex<NatReflectionCache>>,
pool_size: usize, pool_size: usize,
} }
#[derive(Debug, Default)]
pub struct NatReflectionCache {
pub v4: Option<(std::time::Instant, std::net::SocketAddr)>,
pub v6: Option<(std::time::Instant, std::net::SocketAddr)>,
}
impl MePool { impl MePool {
pub fn new( pub fn new(
proxy_tag: Option<Vec<u8>>, proxy_tag: Option<Vec<u8>>,
@ -62,11 +72,15 @@ impl MePool {
proxy_map_v4: HashMap<i32, Vec<(IpAddr, u16)>>, proxy_map_v4: HashMap<i32, Vec<(IpAddr, u16)>>,
proxy_map_v6: HashMap<i32, Vec<(IpAddr, u16)>>, proxy_map_v6: HashMap<i32, Vec<(IpAddr, u16)>>,
default_dc: Option<i32>, default_dc: Option<i32>,
decision: NetworkDecision,
rng: Arc<SecureRandom>,
) -> Arc<Self> { ) -> Arc<Self> {
Arc::new(Self { Arc::new(Self {
registry: Arc::new(ConnRegistry::new()), registry: Arc::new(ConnRegistry::new()),
writers: Arc::new(RwLock::new(Vec::new())), writers: Arc::new(RwLock::new(Vec::new())),
rr: AtomicU64::new(0), rr: AtomicU64::new(0),
decision,
rng,
proxy_tag, proxy_tag,
proxy_secret: Arc::new(RwLock::new(proxy_secret)), proxy_secret: Arc::new(RwLock::new(proxy_secret)),
nat_ip_cfg: nat_ip, nat_ip_cfg: nat_ip,
@ -80,7 +94,7 @@ impl MePool {
next_writer_id: AtomicU64::new(1), next_writer_id: AtomicU64::new(1),
ping_tracker: Arc::new(Mutex::new(HashMap::new())), ping_tracker: Arc::new(Mutex::new(HashMap::new())),
rtt_stats: Arc::new(Mutex::new(HashMap::new())), rtt_stats: Arc::new(Mutex::new(HashMap::new())),
nat_reflection_cache: Arc::new(Mutex::new(None)), nat_reflection_cache: Arc::new(Mutex::new(NatReflectionCache::default())),
}) })
} }
@ -103,15 +117,12 @@ impl MePool {
pub async fn reconcile_connections(self: &Arc<Self>, rng: &SecureRandom) { pub async fn reconcile_connections(self: &Arc<Self>, rng: &SecureRandom) {
use std::collections::HashSet; use std::collections::HashSet;
let map = self.proxy_map_v4.read().await.clone();
let dc_addrs: Vec<(i32, Vec<(IpAddr, u16)>)> = map
.iter()
.map(|(dc, addrs)| (*dc, addrs.clone()))
.collect();
let writers = self.writers.read().await; let writers = self.writers.read().await;
let current: HashSet<SocketAddr> = writers.iter().map(|w| w.addr).collect(); let current: HashSet<SocketAddr> = writers.iter().map(|w| w.addr).collect();
drop(writers); drop(writers);
for family in self.family_order() {
let map = self.proxy_map_for_family(family).await;
for (_dc, addrs) in map.iter() { for (_dc, addrs) in map.iter() {
let dc_addrs: Vec<SocketAddr> = addrs let dc_addrs: Vec<SocketAddr> = addrs
.iter() .iter()
@ -127,6 +138,10 @@ impl MePool {
} }
} }
} }
if !self.decision.effective_multipath && !current.is_empty() {
break;
}
}
} }
pub async fn update_proxy_maps( pub async fn update_proxy_maps(
@ -181,21 +196,51 @@ impl MePool {
} }
} }
pub(super) fn family_order(&self) -> Vec<IpFamily> {
let mut order = Vec::new();
if self.decision.prefer_ipv6() {
if self.decision.ipv6_me {
order.push(IpFamily::V6);
}
if self.decision.ipv4_me {
order.push(IpFamily::V4);
}
} else {
if self.decision.ipv4_me {
order.push(IpFamily::V4);
}
if self.decision.ipv6_me {
order.push(IpFamily::V6);
}
}
order
}
async fn proxy_map_for_family(&self, family: IpFamily) -> HashMap<i32, Vec<(IpAddr, u16)>> {
match family {
IpFamily::V4 => self.proxy_map_v4.read().await.clone(),
IpFamily::V6 => self.proxy_map_v6.read().await.clone(),
}
}
pub async fn init(self: &Arc<Self>, pool_size: usize, rng: &Arc<SecureRandom>) -> Result<()> { pub async fn init(self: &Arc<Self>, pool_size: usize, rng: &Arc<SecureRandom>) -> Result<()> {
let map = self.proxy_map_v4.read().await.clone(); let family_order = self.family_order();
let dc_addrs: Vec<(i32, Vec<(IpAddr, u16)>)> = map
.iter()
.map(|(dc, addrs)| (*dc, addrs.clone()))
.collect();
let ks = self.key_selector().await; let ks = self.key_selector().await;
info!( info!(
me_servers = map.len(), me_servers = self.proxy_map_v4.read().await.len(),
pool_size, pool_size,
key_selector = format_args!("0x{ks:08x}"), key_selector = format_args!("0x{ks:08x}"),
secret_len = self.proxy_secret.read().await.len(), secret_len = self.proxy_secret.read().await.len(),
"Initializing ME pool" "Initializing ME pool"
); );
for family in family_order {
let map = self.proxy_map_for_family(family).await;
let dc_addrs: Vec<(i32, Vec<(IpAddr, u16)>)> = map
.iter()
.map(|(dc, addrs)| (*dc, addrs.clone()))
.collect();
// Ensure at least one connection per DC; run DCs in parallel. // Ensure at least one connection per DC; run DCs in parallel.
let mut join = tokio::task::JoinSet::new(); let mut join = tokio::task::JoinSet::new();
for (dc, addrs) in dc_addrs.iter().cloned() { for (dc, addrs) in dc_addrs.iter().cloned() {
@ -226,6 +271,11 @@ impl MePool {
} }
} }
if !self.decision.effective_multipath && self.connection_count() > 0 {
break;
}
}
if self.writers.read().await.is_empty() { if self.writers.read().await.is_empty() {
return Err(ProxyError::Proxy("No ME connections".into())); return Err(ProxyError::Proxy("No ME connections".into()));
} }
@ -309,14 +359,15 @@ impl MePool {
} }
_ = tokio::time::sleep(Duration::from_secs(wait)) => {} _ = tokio::time::sleep(Duration::from_secs(wait)) => {}
} }
let sent_id = ping_id;
let mut p = Vec::with_capacity(12); let mut p = Vec::with_capacity(12);
p.extend_from_slice(&RPC_PING_U32.to_le_bytes()); p.extend_from_slice(&RPC_PING_U32.to_le_bytes());
p.extend_from_slice(&ping_id.to_le_bytes()); p.extend_from_slice(&sent_id.to_le_bytes());
ping_id = ping_id.wrapping_add(1);
{ {
let mut tracker = ping_tracker_ping.lock().await; let mut tracker = ping_tracker_ping.lock().await;
tracker.insert(ping_id, (std::time::Instant::now(), writer_id)); tracker.insert(sent_id, (std::time::Instant::now(), writer_id));
} }
ping_id = ping_id.wrapping_add(1);
if let Err(e) = rpc_w_ping.lock().await.send(&p).await { if let Err(e) = rpc_w_ping.lock().await.send(&p).await {
debug!(error = %e, "Active ME ping failed, removing dead writer"); debug!(error = %e, "Active ME ping failed, removing dead writer");
cancel_ping.cancel(); cancel_ping.cancel();

View File

@ -4,19 +4,14 @@ use std::time::Duration;
use tracing::{info, warn}; use tracing::{info, warn};
use crate::error::{ProxyError, Result}; use crate::error::{ProxyError, Result};
use crate::network::probe::is_bogon;
use crate::network::stun::{stun_probe_dual, IpFamily, StunProbeResult};
use super::MePool; use super::MePool;
use std::time::Instant; use std::time::Instant;
pub async fn stun_probe(stun_addr: Option<String>) -> Result<crate::network::stun::DualStunResult> {
#[derive(Debug, Clone, Copy)]
pub struct StunProbeResult {
pub local_addr: std::net::SocketAddr,
pub reflected_addr: std::net::SocketAddr,
}
pub async fn stun_probe(stun_addr: Option<String>) -> Result<Option<StunProbeResult>> {
let stun_addr = stun_addr.unwrap_or_else(|| "stun.l.google.com:19302".to_string()); let stun_addr = stun_addr.unwrap_or_else(|| "stun.l.google.com:19302".to_string());
fetch_stun_binding(&stun_addr).await stun_probe_dual(&stun_addr).await
} }
pub async fn detect_public_ip() -> Option<IpAddr> { pub async fn detect_public_ip() -> Option<IpAddr> {
@ -35,7 +30,7 @@ impl MePool {
match (ip, nat_ip) { match (ip, nat_ip) {
(IpAddr::V4(src), IpAddr::V4(dst)) (IpAddr::V4(src), IpAddr::V4(dst))
if is_privateish(IpAddr::V4(src)) if is_bogon(IpAddr::V4(src))
|| src.is_loopback() || src.is_loopback()
|| src.is_unspecified() => || src.is_unspecified() =>
{ {
@ -55,7 +50,7 @@ impl MePool {
) -> std::net::SocketAddr { ) -> std::net::SocketAddr {
let ip = if let Some(r) = reflected { let ip = if let Some(r) = reflected {
// Use reflected IP (not port) only when local address is non-public. // Use reflected IP (not port) only when local address is non-public.
if is_privateish(addr.ip()) || addr.ip().is_loopback() || addr.ip().is_unspecified() { if is_bogon(addr.ip()) || addr.ip().is_loopback() || addr.ip().is_unspecified() {
r.ip() r.ip()
} else { } else {
self.translate_ip_for_nat(addr.ip()) self.translate_ip_for_nat(addr.ip())
@ -73,7 +68,7 @@ impl MePool {
return self.nat_ip_cfg; return self.nat_ip_cfg;
} }
if !(is_privateish(local_ip) || local_ip.is_loopback() || local_ip.is_unspecified()) { if !(is_bogon(local_ip) || local_ip.is_loopback() || local_ip.is_unspecified()) {
return None; return None;
} }
@ -98,12 +93,19 @@ impl MePool {
} }
} }
pub(super) async fn maybe_reflect_public_addr(&self) -> Option<std::net::SocketAddr> { pub(super) async fn maybe_reflect_public_addr(
&self,
family: IpFamily,
) -> Option<std::net::SocketAddr> {
const STUN_CACHE_TTL: Duration = Duration::from_secs(600); const STUN_CACHE_TTL: Duration = Duration::from_secs(600);
if let Ok(mut cache) = self.nat_reflection_cache.try_lock() { if let Ok(mut cache) = self.nat_reflection_cache.try_lock() {
if let Some((ts, addr)) = *cache { let slot = match family {
IpFamily::V4 => &mut cache.v4,
IpFamily::V6 => &mut cache.v6,
};
if let Some((ts, addr)) = slot {
if ts.elapsed() < STUN_CACHE_TTL { if ts.elapsed() < STUN_CACHE_TTL {
return Some(addr); return Some(*addr);
} }
} }
} }
@ -112,12 +114,20 @@ impl MePool {
.nat_stun .nat_stun
.clone() .clone()
.unwrap_or_else(|| "stun.l.google.com:19302".to_string()); .unwrap_or_else(|| "stun.l.google.com:19302".to_string());
match fetch_stun_binding(&stun_addr).await { match stun_probe_dual(&stun_addr).await {
Ok(sa) => { Ok(res) => {
if let Some(result) = sa { let picked: Option<StunProbeResult> = match family {
info!(local = %result.local_addr, reflected = %result.reflected_addr, "NAT probe: reflected address"); IpFamily::V4 => res.v4,
IpFamily::V6 => res.v6,
};
if let Some(result) = picked {
info!(local = %result.local_addr, reflected = %result.reflected_addr, family = ?family, "NAT probe: reflected address");
if let Ok(mut cache) = self.nat_reflection_cache.try_lock() { if let Ok(mut cache) = self.nat_reflection_cache.try_lock() {
*cache = Some((Instant::now(), result.reflected_addr)); let slot = match family {
IpFamily::V4 => &mut cache.v4,
IpFamily::V6 => &mut cache.v6,
};
*slot = Some((Instant::now(), result.reflected_addr));
} }
Some(result.reflected_addr) Some(result.reflected_addr)
} else { } else {
@ -158,98 +168,3 @@ async fn fetch_public_ipv4_once(url: &str) -> Result<Option<Ipv4Addr>> {
let ip = text.trim().parse().ok(); let ip = text.trim().parse().ok();
Ok(ip) Ok(ip)
} }
async fn fetch_stun_binding(stun_addr: &str) -> Result<Option<StunProbeResult>> {
use rand::RngCore;
use tokio::net::UdpSocket;
let socket = UdpSocket::bind("0.0.0.0:0")
.await
.map_err(|e| ProxyError::Proxy(format!("STUN bind failed: {e}")))?;
socket
.connect(stun_addr)
.await
.map_err(|e| ProxyError::Proxy(format!("STUN connect failed: {e}")))?;
// Build minimal Binding Request.
let mut req = vec![0u8; 20];
req[0..2].copy_from_slice(&0x0001u16.to_be_bytes()); // Binding Request
req[2..4].copy_from_slice(&0u16.to_be_bytes()); // length
req[4..8].copy_from_slice(&0x2112A442u32.to_be_bytes()); // magic cookie
rand::rng().fill_bytes(&mut req[8..20]);
socket
.send(&req)
.await
.map_err(|e| ProxyError::Proxy(format!("STUN send failed: {e}")))?;
let mut buf = [0u8; 128];
let n = socket
.recv(&mut buf)
.await
.map_err(|e| ProxyError::Proxy(format!("STUN recv failed: {e}")))?;
if n < 20 {
return Ok(None);
}
// Parse attributes.
let mut idx = 20;
while idx + 4 <= n {
let atype = u16::from_be_bytes(buf[idx..idx + 2].try_into().unwrap());
let alen = u16::from_be_bytes(buf[idx + 2..idx + 4].try_into().unwrap()) as usize;
idx += 4;
if idx + alen > n {
break;
}
match atype {
0x0020 /* XOR-MAPPED-ADDRESS */ | 0x0001 /* MAPPED-ADDRESS */ => {
if alen < 8 {
break;
}
let family = buf[idx + 1];
if family != 0x01 {
// only IPv4 supported here
break;
}
let port_bytes = [buf[idx + 2], buf[idx + 3]];
let ip_bytes = [buf[idx + 4], buf[idx + 5], buf[idx + 6], buf[idx + 7]];
let (port, ip) = if atype == 0x0020 {
let magic = 0x2112A442u32.to_be_bytes();
let port = u16::from_be_bytes(port_bytes) ^ ((magic[0] as u16) << 8 | magic[1] as u16);
let ip = [
ip_bytes[0] ^ magic[0],
ip_bytes[1] ^ magic[1],
ip_bytes[2] ^ magic[2],
ip_bytes[3] ^ magic[3],
];
(port, ip)
} else {
(u16::from_be_bytes(port_bytes), ip_bytes)
};
let reflected = std::net::SocketAddr::new(
IpAddr::V4(Ipv4Addr::new(ip[0], ip[1], ip[2], ip[3])),
port,
);
let local_addr = socket.local_addr().map_err(|e| {
ProxyError::Proxy(format!("STUN local_addr failed: {e}"))
})?;
return Ok(Some(StunProbeResult {
local_addr,
reflected_addr: reflected,
}));
}
_ => {}
}
idx += (alen + 3) & !3; // 4-byte alignment
}
Ok(None)
}
fn is_privateish(ip: IpAddr) -> bool {
match ip {
IpAddr::V4(v4) => v4.is_private() || v4.is_link_local(),
IpAddr::V6(v6) => v6.is_unique_local(),
}
}

View File

@ -152,6 +152,9 @@ pub(crate) async fn reader_loop(
entry.1 = entry.1 * 0.8 + rtt * 0.2; entry.1 = entry.1 * 0.8 + rtt * 0.2;
if rtt < entry.0 { if rtt < entry.0 {
entry.0 = rtt; entry.0 = rtt;
} else {
// allow slow baseline drift upward to avoid stale minimum
entry.0 = entry.0 * 0.99 + rtt * 0.01;
} }
let degraded_now = entry.1 > entry.0 * 2.0; let degraded_now = entry.1 > entry.0 * 2.0;
degraded.store(degraded_now, Ordering::Relaxed); degraded.store(degraded_now, Ordering::Relaxed);

View File

@ -1,4 +1,5 @@
use std::sync::Arc; use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration; use std::time::Duration;
use tracing::{info, warn}; use tracing::{info, warn};
@ -15,7 +16,12 @@ pub async fn me_rotation_task(pool: Arc<MePool>, rng: Arc<SecureRandom>, interva
let candidate = { let candidate = {
let ws = pool.writers.read().await; let ws = pool.writers.read().await;
ws.get(0).cloned() if ws.is_empty() {
None
} else {
let idx = (pool.rr.load(std::sync::atomic::Ordering::Relaxed) as usize) % ws.len();
ws.get(idx).cloned()
}
}; };
let Some(w) = candidate else { let Some(w) = candidate else {
@ -34,4 +40,3 @@ pub async fn me_rotation_task(pool: Arc<MePool>, rng: Arc<SecureRandom>, interva
} }
} }
} }

View File

@ -3,15 +3,14 @@ use std::sync::Arc;
use std::sync::atomic::Ordering; use std::sync::atomic::Ordering;
use std::time::Duration; use std::time::Duration;
use tokio::sync::Mutex;
use tracing::{debug, warn}; use tracing::{debug, warn};
use crate::error::{ProxyError, Result}; use crate::error::{ProxyError, Result};
use crate::network::IpFamily;
use crate::protocol::constants::RPC_CLOSE_EXT_U32; use crate::protocol::constants::RPC_CLOSE_EXT_U32;
use super::MePool; use super::MePool;
use super::wire::build_proxy_req_payload; use super::wire::build_proxy_req_payload;
use crate::crypto::SecureRandom;
use rand::seq::SliceRandom; use rand::seq::SliceRandom;
use super::registry::ConnMeta; use super::registry::ConnMeta;
@ -84,7 +83,7 @@ impl MePool {
drop(map); drop(map);
for (ip, port) in shuffled { for (ip, port) in shuffled {
let addr = SocketAddr::new(ip, port); let addr = SocketAddr::new(ip, port);
if self.connect_one(addr, &SecureRandom::new()).await.is_ok() { if self.connect_one(addr, self.rng.as_ref()).await.is_ok() {
break; break;
} }
} }
@ -173,34 +172,46 @@ impl MePool {
writers: &[super::pool::MeWriter], writers: &[super::pool::MeWriter],
target_dc: i16, target_dc: i16,
) -> Vec<usize> { ) -> Vec<usize> {
let mut preferred = Vec::<SocketAddr>::new();
let key = target_dc as i32; let key = target_dc as i32;
let map = self.proxy_map_v4.read().await; let mut preferred = Vec::<SocketAddr>::new();
if let Some(v) = map.get(&key) { for family in self.family_order() {
let map_guard = match family {
IpFamily::V4 => self.proxy_map_v4.read().await,
IpFamily::V6 => self.proxy_map_v6.read().await,
};
if let Some(v) = map_guard.get(&key) {
preferred.extend(v.iter().map(|(ip, port)| SocketAddr::new(*ip, *port))); preferred.extend(v.iter().map(|(ip, port)| SocketAddr::new(*ip, *port)));
} }
if preferred.is_empty() { if preferred.is_empty() {
let abs = key.abs(); let abs = key.abs();
if let Some(v) = map.get(&abs) { if let Some(v) = map_guard.get(&abs) {
preferred.extend(v.iter().map(|(ip, port)| SocketAddr::new(*ip, *port))); preferred.extend(v.iter().map(|(ip, port)| SocketAddr::new(*ip, *port)));
} }
} }
if preferred.is_empty() { if preferred.is_empty() {
let abs = key.abs(); let abs = key.abs();
if let Some(v) = map.get(&-abs) { if let Some(v) = map_guard.get(&-abs) {
preferred.extend(v.iter().map(|(ip, port)| SocketAddr::new(*ip, *port))); preferred.extend(v.iter().map(|(ip, port)| SocketAddr::new(*ip, *port)));
} }
} }
if preferred.is_empty() { if preferred.is_empty() {
let def = self.default_dc.load(Ordering::Relaxed); let def = self.default_dc.load(Ordering::Relaxed);
if def != 0 { if def != 0 {
if let Some(v) = map.get(&def) { if let Some(v) = map_guard.get(&def) {
preferred.extend(v.iter().map(|(ip, port)| SocketAddr::new(*ip, *port))); preferred.extend(v.iter().map(|(ip, port)| SocketAddr::new(*ip, *port)));
} }
} }
} }
drop(map_guard);
if !preferred.is_empty() && !self.decision.effective_multipath {
break;
}
}
if preferred.is_empty() { if preferred.is_empty() {
return (0..writers.len()).collect(); return (0..writers.len()).collect();
} }

View File

@ -355,6 +355,8 @@ impl UpstreamManager {
&self, &self,
prefer_ipv6: bool, prefer_ipv6: bool,
dc_overrides: &HashMap<String, Vec<String>>, dc_overrides: &HashMap<String, Vec<String>>,
ipv4_enabled: bool,
ipv6_enabled: bool,
) -> Vec<StartupPingResult> { ) -> Vec<StartupPingResult> {
let upstreams: Vec<(usize, UpstreamConfig)> = { let upstreams: Vec<(usize, UpstreamConfig)> = {
let guard = self.upstreams.read().await; let guard = self.upstreams.read().await;
@ -374,10 +376,8 @@ impl UpstreamManager {
UpstreamType::Socks5 { address, .. } => format!("socks5://{}", address), UpstreamType::Socks5 { address, .. } => format!("socks5://{}", address),
}; };
let mut v6_results = Vec::new(); let mut v6_results = Vec::with_capacity(NUM_DCS);
let mut v4_results = Vec::new(); if ipv6_enabled {
// === Ping IPv6 first ===
for dc_zero_idx in 0..NUM_DCS { for dc_zero_idx in 0..NUM_DCS {
let dc_v6 = TG_DATACENTERS_V6[dc_zero_idx]; let dc_v6 = TG_DATACENTERS_V6[dc_zero_idx];
let addr_v6 = SocketAddr::new(dc_v6, TG_DATACENTER_PORT); let addr_v6 = SocketAddr::new(dc_v6, TG_DATACENTER_PORT);
@ -415,8 +415,20 @@ impl UpstreamManager {
}; };
v6_results.push(ping_result); v6_results.push(ping_result);
} }
} else {
for dc_zero_idx in 0..NUM_DCS {
let dc_v6 = TG_DATACENTERS_V6[dc_zero_idx];
v6_results.push(DcPingResult {
dc_idx: dc_zero_idx + 1,
dc_addr: SocketAddr::new(dc_v6, TG_DATACENTER_PORT),
rtt_ms: None,
error: Some("ipv6 disabled".to_string()),
});
}
}
// === Then ping IPv4 === let mut v4_results = Vec::with_capacity(NUM_DCS);
if ipv4_enabled {
for dc_zero_idx in 0..NUM_DCS { for dc_zero_idx in 0..NUM_DCS {
let dc_v4 = TG_DATACENTERS_V4[dc_zero_idx]; let dc_v4 = TG_DATACENTERS_V4[dc_zero_idx];
let addr_v4 = SocketAddr::new(dc_v4, TG_DATACENTER_PORT); let addr_v4 = SocketAddr::new(dc_v4, TG_DATACENTER_PORT);
@ -454,6 +466,17 @@ impl UpstreamManager {
}; };
v4_results.push(ping_result); v4_results.push(ping_result);
} }
} else {
for dc_zero_idx in 0..NUM_DCS {
let dc_v4 = TG_DATACENTERS_V4[dc_zero_idx];
v4_results.push(DcPingResult {
dc_idx: dc_zero_idx + 1,
dc_addr: SocketAddr::new(dc_v4, TG_DATACENTER_PORT),
rtt_ms: None,
error: Some("ipv4 disabled".to_string()),
});
}
}
// === Ping DC overrides (v4/v6) === // === Ping DC overrides (v4/v6) ===
for (dc_key, addrs) in dc_overrides { for (dc_key, addrs) in dc_overrides {
@ -470,6 +493,9 @@ impl UpstreamManager {
match addr_str.parse::<SocketAddr>() { match addr_str.parse::<SocketAddr>() {
Ok(addr) => { Ok(addr) => {
let is_v6 = addr.is_ipv6(); let is_v6 = addr.is_ipv6();
if (is_v6 && !ipv6_enabled) || (!is_v6 && !ipv4_enabled) {
continue;
}
let result = tokio::time::timeout( let result = tokio::time::timeout(
Duration::from_secs(DC_PING_TIMEOUT_SECS), Duration::from_secs(DC_PING_TIMEOUT_SECS),
self.ping_single_dc(&upstream_config, addr) self.ping_single_dc(&upstream_config, addr)
@ -551,7 +577,7 @@ impl UpstreamManager {
/// Background health check: rotates through DCs, 30s interval. /// Background health check: rotates through DCs, 30s interval.
/// Uses preferred IP version based on config. /// Uses preferred IP version based on config.
pub async fn run_health_checks(&self, prefer_ipv6: bool) { pub async fn run_health_checks(&self, prefer_ipv6: bool, ipv4_enabled: bool, ipv6_enabled: bool) {
let mut dc_rotation = 0usize; let mut dc_rotation = 0usize;
loop { loop {
@ -560,16 +586,24 @@ impl UpstreamManager {
let dc_zero_idx = dc_rotation % NUM_DCS; let dc_zero_idx = dc_rotation % NUM_DCS;
dc_rotation += 1; dc_rotation += 1;
let dc_addr = if prefer_ipv6 { let primary_v6 = SocketAddr::new(TG_DATACENTERS_V6[dc_zero_idx], TG_DATACENTER_PORT);
SocketAddr::new(TG_DATACENTERS_V6[dc_zero_idx], TG_DATACENTER_PORT) let primary_v4 = SocketAddr::new(TG_DATACENTERS_V4[dc_zero_idx], TG_DATACENTER_PORT);
let dc_addr = if prefer_ipv6 && ipv6_enabled {
primary_v6
} else if ipv4_enabled {
primary_v4
} else if ipv6_enabled {
primary_v6
} else { } else {
SocketAddr::new(TG_DATACENTERS_V4[dc_zero_idx], TG_DATACENTER_PORT) continue;
}; };
let fallback_addr = if prefer_ipv6 { let fallback_addr = if dc_addr.is_ipv6() && ipv4_enabled {
SocketAddr::new(TG_DATACENTERS_V4[dc_zero_idx], TG_DATACENTER_PORT) Some(primary_v4)
} else if dc_addr.is_ipv4() && ipv6_enabled {
Some(primary_v6)
} else { } else {
SocketAddr::new(TG_DATACENTERS_V6[dc_zero_idx], TG_DATACENTER_PORT) None
}; };
let count = self.upstreams.read().await.len(); let count = self.upstreams.read().await.len();
@ -608,6 +642,7 @@ impl UpstreamManager {
// Try fallback // Try fallback
debug!(dc = dc_zero_idx + 1, "Health check failed, trying fallback"); debug!(dc = dc_zero_idx + 1, "Health check failed, trying fallback");
if let Some(fallback_addr) = fallback_addr {
let start2 = Instant::now(); let start2 = Instant::now();
let result2 = tokio::time::timeout( let result2 = tokio::time::timeout(
Duration::from_secs(10), Duration::from_secs(10),
@ -652,6 +687,17 @@ impl UpstreamManager {
} }
} }
u.last_check = std::time::Instant::now(); u.last_check = std::time::Instant::now();
continue;
}
let mut guard = self.upstreams.write().await;
let u = &mut guard[i];
u.fails += 1;
if u.fails > 3 {
u.healthy = false;
warn!("Upstream unhealthy (no fallback family)");
}
u.last_check = std::time::Instant::now();
} }
} }
} }