astelm
/
telemt
mirror of https://github.com/telemt/telemt.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: resolve clippy warnings 

Reduce clippy warnings from54 to16 by fixing mechanical issues:

- collapsible_if: collapse nested if-let chains with let-chains
- clone_on_copy: remove unnecessary .clone() on Copy types
- manual_clamp: replace .max().min() with .clamp()
- unnecessary_cast: remove redundant type casts
- collapsible_else_if: flatten else-if chains
- contains_vs_iter_any: replace .iter().any() with .contains()
- unnecessary_closure: replace .or_else(|| x) with .or(x)
- useless_conversion: remove redundant .into() calls
- is_none_or: replace .map_or(true, ...) with .is_none_or(...)
- while_let_loop: convert loop with if-let-break to while-let

Remaining16 warnings are design-level issues (too_many_arguments,
await_holding_lock, type_complexity, new_ret_no_self) that require
architectural changes to fix.
This commit is contained in:
Vladislav Yaroslavlev 2026-02-24 05:57:53 +03:00
parent d6214c6bbf
commit 09f56dede2
No known key found for this signature in database
GPG Key ID: C67C785A19F5503F
38 changed files with 336 additions and 358 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/config/defaults.rs
View File

@ -171,15 +171,15 @@ pub(crate) fn default_cache_public_ip_path() -> String {
} }
pub(crate) fn default_proxy_secret_reload_secs() -> u64 { pub(crate) fn default_proxy_secret_reload_secs() -> u64 {
1 * 60 * 60 60 * 60
} }
pub(crate) fn default_proxy_config_reload_secs() -> u64 { pub(crate) fn default_proxy_config_reload_secs() -> u64 {
1 * 60 * 60 60 * 60
} }
pub(crate) fn default_update_every_secs() -> u64 { pub(crate) fn default_update_every_secs() -> u64 {
1 * 30 * 60 30 * 60
} }
pub(crate) fn default_me_reinit_drain_timeout_secs() -> u64 { pub(crate) fn default_me_reinit_drain_timeout_secs() -> u64 {

26
src/config/load.rs
View File

@ -278,23 +278,25 @@ impl ProxyConfig {
reuse_allow: false, reuse_allow: false,
}); });
} }
if let Some(ipv6_str) = &config.server.listen_addr_ipv6 { if let Some(ipv6_str) = &config.server.listen_addr_ipv6
if let Ok(ipv6) = ipv6_str.parse::<IpAddr>() { && let Ok(ipv6) = ipv6_str.parse::<IpAddr>()
config.server.listeners.push(ListenerConfig { {
ip: ipv6, config.server.listeners.push(ListenerConfig {
announce: None, ip: ipv6,
announce_ip: None, announce: None,
proxy_protocol: None, announce_ip: None,
reuse_allow: false, proxy_protocol: None,
}); reuse_allow: false,
} });
} }
} }
// Migration: announce_ip → announce for each listener. // Migration: announce_ip → announce for each listener.
for listener in &mut config.server.listeners { for listener in &mut config.server.listeners {
if listener.announce.is_none() && listener.announce_ip.is_some() { if listener.announce.is_none()
listener.announce = Some(listener.announce_ip.unwrap().to_string()); && let Some(ip) = listener.announce_ip.take()
{
listener.announce = Some(ip.to_string());
} }
} }

9
src/config/types.rs
View File

@ -677,9 +677,10 @@ pub struct ListenerConfig {
/// - `show_link = "*"` — show links for all users /// - `show_link = "*"` — show links for all users
/// - `show_link = ["a", "b"]` — show links for specific users /// - `show_link = ["a", "b"]` — show links for specific users
/// - omitted — show no links (default) /// - omitted — show no links (default)
#[derive(Debug, Clone)] #[derive(Debug, Clone, Default)]
pub enum ShowLink { pub enum ShowLink {
/// Don't show any links (default when omitted). /// Don't show any links (default when omitted).
#[default]
None, None,
/// Show links for all configured users. /// Show links for all configured users.
All, All,
@ -687,12 +688,6 @@ pub enum ShowLink {
Specific(Vec<String>), Specific(Vec<String>),
} }
impl Default for ShowLink {
fn default() -> Self {
ShowLink::None
}
}
impl ShowLink { impl ShowLink {
/// Returns true if no links should be shown. /// Returns true if no links should be shown.
pub fn is_empty(&self) -> bool { pub fn is_empty(&self) -> bool {

16
src/crypto/aes.rs
View File

@ -23,13 +23,13 @@ type Aes256Ctr = Ctr128BE<Aes256>;
// ============= AES-256-CTR ============= // ============= AES-256-CTR =============
/// AES-256-CTR encryptor/decryptor /// AES-256-CTR encryptor/decryptor
/// ///
/// CTR mode is symmetric — encryption and decryption are the same operation. /// CTR mode is symmetric — encryption and decryption are the same operation.
/// ///
/// **Zeroize note:** The inner `Aes256Ctr` cipher state (expanded key schedule /// **Zeroize note:** The inner `Aes256Ctr` cipher state (expanded key schedule
/// + counter) is opaque and cannot be zeroized. If you need to protect key /// + counter) is opaque and cannot be zeroized. If you need to protect key
/// material, zeroize the `[u8; 32]` key and `u128` IV at the call site /// material, zeroize the `[u8; 32]` key and `u128` IV at the call site
/// before dropping them. /// before dropping them.
pub struct AesCtr { pub struct AesCtr {
cipher: Aes256Ctr, cipher: Aes256Ctr,
} }
@ -149,7 +149,7 @@ impl AesCbc {
/// ///
/// CBC Encryption: C[i] = AES_Encrypt(P[i] XOR C[i-1]), where C[-1] = IV /// CBC Encryption: C[i] = AES_Encrypt(P[i] XOR C[i-1]), where C[-1] = IV
pub fn encrypt(&self, data: &[u8]) -> Result<Vec<u8>> { pub fn encrypt(&self, data: &[u8]) -> Result<Vec<u8>> {
if data.len() % Self::BLOCK_SIZE != 0 { if !data.len().is_multiple_of(Self::BLOCK_SIZE) {
return Err(ProxyError::Crypto( return Err(ProxyError::Crypto(
format!("CBC data must be aligned to 16 bytes, got {}", data.len()) format!("CBC data must be aligned to 16 bytes, got {}", data.len())
)); ));
@ -180,7 +180,7 @@ impl AesCbc {
/// ///
/// CBC Decryption: P[i] = AES_Decrypt(C[i]) XOR C[i-1], where C[-1] = IV /// CBC Decryption: P[i] = AES_Decrypt(C[i]) XOR C[i-1], where C[-1] = IV
pub fn decrypt(&self, data: &[u8]) -> Result<Vec<u8>> { pub fn decrypt(&self, data: &[u8]) -> Result<Vec<u8>> {
if data.len() % Self::BLOCK_SIZE != 0 { if !data.len().is_multiple_of(Self::BLOCK_SIZE) {
return Err(ProxyError::Crypto( return Err(ProxyError::Crypto(
format!("CBC data must be aligned to 16 bytes, got {}", data.len()) format!("CBC data must be aligned to 16 bytes, got {}", data.len())
)); ));
@ -209,7 +209,7 @@ impl AesCbc {
/// Encrypt data in-place /// Encrypt data in-place
pub fn encrypt_in_place(&self, data: &mut [u8]) -> Result<()> { pub fn encrypt_in_place(&self, data: &mut [u8]) -> Result<()> {
if data.len() % Self::BLOCK_SIZE != 0 { if !data.len().is_multiple_of(Self::BLOCK_SIZE) {
return Err(ProxyError::Crypto( return Err(ProxyError::Crypto(
format!("CBC data must be aligned to 16 bytes, got {}", data.len()) format!("CBC data must be aligned to 16 bytes, got {}", data.len())
)); ));
@ -242,7 +242,7 @@ impl AesCbc {
/// Decrypt data in-place /// Decrypt data in-place
pub fn decrypt_in_place(&self, data: &mut [u8]) -> Result<()> { pub fn decrypt_in_place(&self, data: &mut [u8]) -> Result<()> {
if data.len() % Self::BLOCK_SIZE != 0 { if !data.len().is_multiple_of(Self::BLOCK_SIZE) {
return Err(ProxyError::Crypto( return Err(ProxyError::Crypto(
format!("CBC data must be aligned to 16 bytes, got {}", data.len()) format!("CBC data must be aligned to 16 bytes, got {}", data.len())
)); ));

2
src/crypto/hash.rs
View File

@ -64,6 +64,7 @@ pub fn crc32c(data: &[u8]) -> u32 {
/// ///
/// Returned buffer layout (IPv4): /// Returned buffer layout (IPv4):
/// nonce_srv | nonce_clt | clt_ts | srv_ip | clt_port | purpose | clt_ip | srv_port | secret | nonce_srv | [clt_v6 | srv_v6] | nonce_clt /// nonce_srv | nonce_clt | clt_ts | srv_ip | clt_port | purpose | clt_ip | srv_port | secret | nonce_srv | [clt_v6 | srv_v6] | nonce_clt
#[allow(clippy::too_many_arguments)]
pub fn build_middleproxy_prekey( pub fn build_middleproxy_prekey(
nonce_srv: &[u8; 16], nonce_srv: &[u8; 16],
nonce_clt: &[u8; 16], nonce_clt: &[u8; 16],
@ -108,6 +109,7 @@ pub fn build_middleproxy_prekey(
/// Uses MD5 + SHA-1 as mandated by the Telegram Middle Proxy protocol. /// Uses MD5 + SHA-1 as mandated by the Telegram Middle Proxy protocol.
/// These algorithms are NOT replaceable here — changing them would break /// These algorithms are NOT replaceable here — changing them would break
/// interoperability with Telegram's middle proxy infrastructure. /// interoperability with Telegram's middle proxy infrastructure.
#[allow(clippy::too_many_arguments)]
pub fn derive_middleproxy_keys( pub fn derive_middleproxy_keys(
nonce_srv: &[u8; 16], nonce_srv: &[u8; 16],
nonce_clt: &[u8; 16], nonce_clt: &[u8; 16],

2
src/crypto/random.rs
View File

@ -95,7 +95,7 @@ impl SecureRandom {
return 0; return 0;
} }
let bytes_needed = (k + 7) / 8; let bytes_needed = k.div_ceil(8);
let bytes = self.bytes(bytes_needed.min(8)); let bytes = self.bytes(bytes_needed.min(8));
let mut result = 0u64; let mut result = 0u64;

11
src/error.rs
View File

@ -91,7 +91,7 @@ impl From<StreamError> for std::io::Error {
std::io::Error::new(std::io::ErrorKind::UnexpectedEof, err) std::io::Error::new(std::io::ErrorKind::UnexpectedEof, err)
} }
StreamError::Poisoned { .. } => { StreamError::Poisoned { .. } => {
std::io::Error::new(std::io::ErrorKind::Other, err) std::io::Error::other(err)
} }
StreamError::BufferOverflow { .. } => { StreamError::BufferOverflow { .. } => {
std::io::Error::new(std::io::ErrorKind::OutOfMemory, err) std::io::Error::new(std::io::ErrorKind::OutOfMemory, err)
@ -100,7 +100,7 @@ impl From<StreamError> for std::io::Error {
std::io::Error::new(std::io::ErrorKind::InvalidData, err) std::io::Error::new(std::io::ErrorKind::InvalidData, err)
} }
StreamError::PartialRead { .. } | StreamError::PartialWrite { .. } => { StreamError::PartialRead { .. } | StreamError::PartialWrite { .. } => {
std::io::Error::new(std::io::ErrorKind::Other, err) std::io::Error::other(err)
} }
} }
} }
@ -135,12 +135,7 @@ impl Recoverable for StreamError {
} }
fn can_continue(&self) -> bool { fn can_continue(&self) -> bool {
match self { !matches!(self, Self::Poisoned { .. } | Self::UnexpectedEof | Self::BufferOverflow { .. })
Self::Poisoned { .. } => false,
Self::UnexpectedEof => false,
Self::BufferOverflow { .. } => false,
_ => true,
}
} }
} }

16
src/main.rs
View File

@ -301,7 +301,7 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).await { match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).await {
Ok(proxy_secret) => { Ok(proxy_secret) => {
info!( info!(
secret_len = proxy_secret.len() as usize, // ← ЯВНЫЙ ТИП usize secret_len = proxy_secret.len(),
key_sig = format_args!( key_sig = format_args!(
"0x{:08x}", "0x{:08x}",
if proxy_secret.len() >= 4 { if proxy_secret.len() >= 4 {
@ -597,14 +597,12 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
} else { } else {
info!(" IPv4 in use / IPv6 is fallback"); info!(" IPv4 in use / IPv6 is fallback");
} }
} else { } else if v6_works && !v4_works {
if v6_works && !v4_works { info!(" IPv6 only / IPv4 unavailable)");
info!(" IPv6 only / IPv4 unavailable)"); } else if v4_works && !v6_works {
} else if v4_works && !v6_works { info!(" IPv4 only / IPv6 unavailable)");
info!(" IPv4 only / IPv6 unavailable)"); } else if !v6_works && !v4_works {
} else if !v6_works && !v4_works { info!(" No DC connectivity");
info!(" No DC connectivity");
}
} }
info!(" via {}", upstream_result.upstream_name); info!(" via {}", upstream_result.upstream_name);

29
src/network/probe.rs
View File

@ -95,23 +95,21 @@ pub async fn run_probe(config: &NetworkConfig, stun_addr: Option<String>, nat_pr
} }
pub fn decide_network_capabilities(config: &NetworkConfig, probe: &NetworkProbe) -> NetworkDecision { pub fn decide_network_capabilities(config: &NetworkConfig, probe: &NetworkProbe) -> NetworkDecision {
let mut decision = NetworkDecision::default(); let ipv4_dc = config.ipv4 && probe.detected_ipv4.is_some();
let ipv6_dc = config.ipv6.unwrap_or(probe.detected_ipv6.is_some()) && probe.detected_ipv6.is_some();
decision.ipv4_dc = config.ipv4 && probe.detected_ipv4.is_some(); let ipv4_me = config.ipv4
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.detected_ipv4.is_some()
&& (!probe.ipv4_is_bogon || probe.reflected_ipv4.is_some()); && (!probe.ipv4_is_bogon || probe.reflected_ipv4.is_some());
let ipv6_enabled = config.ipv6.unwrap_or(probe.detected_ipv6.is_some()); let ipv6_enabled = config.ipv6.unwrap_or(probe.detected_ipv6.is_some());
decision.ipv6_me = ipv6_enabled let ipv6_me = ipv6_enabled
&& probe.detected_ipv6.is_some() && probe.detected_ipv6.is_some()
&& (!probe.ipv6_is_bogon || probe.reflected_ipv6.is_some()); && (!probe.ipv6_is_bogon || probe.reflected_ipv6.is_some());
decision.effective_prefer = match config.prefer { let effective_prefer = match config.prefer {
6 if decision.ipv6_me || decision.ipv6_dc => 6, 6 if ipv6_me || ipv6_dc => 6,
4 if decision.ipv4_me || decision.ipv4_dc => 4, 4 if ipv4_me || ipv4_dc => 4,
6 => { 6 => {
warn!("prefer=6 requested but IPv6 unavailable; falling back to IPv4"); warn!("prefer=6 requested but IPv6 unavailable; falling back to IPv4");
4 4
@ -119,10 +117,17 @@ pub fn decide_network_capabilities(config: &NetworkConfig, probe: &NetworkProbe)
_ => 4, _ => 4,
}; };
let me_families = decision.ipv4_me as u8 + decision.ipv6_me as u8; let me_families = ipv4_me as u8 + ipv6_me as u8;
decision.effective_multipath = config.multipath && me_families >= 2; let effective_multipath = config.multipath && me_families >= 2;
decision NetworkDecision {
ipv4_dc,
ipv6_dc,
ipv4_me,
ipv6_me,
effective_prefer,
effective_multipath,
}
} }
fn detect_local_ip_v4() -> Option<Ipv4Addr> { fn detect_local_ip_v4() -> Option<Ipv4Addr> {

9
src/network/stun.rs
View File

@ -198,16 +198,11 @@ async fn resolve_stun_addr(stun_addr: &str, family: IpFamily) -> Result<Option<S
}); });
} }
let addrs = lookup_host(stun_addr) let mut addrs = lookup_host(stun_addr)
.await .await
.map_err(|e| ProxyError::Proxy(format!("STUN resolve failed: {e}")))?; .map_err(|e| ProxyError::Proxy(format!("STUN resolve failed: {e}")))?;
let target = addrs let target = addrs
.filter(|a| match (a.is_ipv4(), family) { .find(|a| matches!((a.is_ipv4(), family), (true, IpFamily::V4) | (false, IpFamily::V6)));
(true, IpFamily::V4) => true,
(false, IpFamily::V6) => true,
_ => false,
})
.next();
Ok(target) Ok(target)
} }

5
src/protocol/constants.rs
View File

@ -160,7 +160,7 @@ pub const MAX_TLS_CHUNK_SIZE: usize = 16384 + 256;
/// Secure Intermediate payload is expected to be 4-byte aligned. /// Secure Intermediate payload is expected to be 4-byte aligned.
pub fn is_valid_secure_payload_len(data_len: usize) -> bool { pub fn is_valid_secure_payload_len(data_len: usize) -> bool {
data_len % 4 == 0 data_len.is_multiple_of(4)
} }
/// Compute Secure Intermediate payload length from wire length. /// Compute Secure Intermediate payload length from wire length.
@ -179,7 +179,7 @@ pub fn secure_padding_len(data_len: usize, rng: &SecureRandom) -> usize {
is_valid_secure_payload_len(data_len), is_valid_secure_payload_len(data_len),
"Secure payload must be 4-byte aligned, got {data_len}" "Secure payload must be 4-byte aligned, got {data_len}"
); );
(rng.range(3) + 1) as usize rng.range(3) + 1
} }
// ============= Timeouts ============= // ============= Timeouts =============
@ -231,7 +231,6 @@ pub static RESERVED_NONCE_CONTINUES: &[[u8; 4]] = &[
// ============= RPC Constants (for Middle Proxy) ============= // ============= RPC Constants (for Middle Proxy) =============
/// RPC Proxy Request /// RPC Proxy Request
/// RPC Flags (from Erlang mtp_rpc.erl) /// RPC Flags (from Erlang mtp_rpc.erl)
pub const RPC_FLAG_NOT_ENCRYPTED: u32 = 0x2; pub const RPC_FLAG_NOT_ENCRYPTED: u32 = 0x2;
pub const RPC_FLAG_HAS_AD_TAG: u32 = 0x8; pub const RPC_FLAG_HAS_AD_TAG: u32 = 0x8;

2
src/protocol/frame.rs
View File

@ -85,7 +85,7 @@ impl FrameMode {
pub fn validate_message_length(len: usize) -> bool { pub fn validate_message_length(len: usize) -> bool {
use super::constants::{MIN_MSG_LEN, MAX_MSG_LEN, PADDING_FILLER}; use super::constants::{MIN_MSG_LEN, MAX_MSG_LEN, PADDING_FILLER};
len >= MIN_MSG_LEN && len <= MAX_MSG_LEN && len % PADDING_FILLER.len() == 0 (MIN_MSG_LEN..=MAX_MSG_LEN).contains(&len) && len.is_multiple_of(PADDING_FILLER.len())
} }
#[cfg(test)] #[cfg(test)]

14
src/protocol/tls.rs
View File

@ -335,7 +335,7 @@ pub fn validate_tls_handshake(
// This is a quirk in some clients that use uptime instead of real time // This is a quirk in some clients that use uptime instead of real time
let is_boot_time = timestamp < 60 * 60 * 24 * 1000; // < ~2.7 years in seconds let is_boot_time = timestamp < 60 * 60 * 24 * 1000; // < ~2.7 years in seconds
if !is_boot_time && (time_diff < TIME_SKEW_MIN || time_diff > TIME_SKEW_MAX) { if !is_boot_time && !(TIME_SKEW_MIN..=TIME_SKEW_MAX).contains(&time_diff) {
continue; continue;
} }
} }
@ -393,7 +393,7 @@ pub fn build_server_hello(
) -> Vec<u8> { ) -> Vec<u8> {
const MIN_APP_DATA: usize = 64; const MIN_APP_DATA: usize = 64;
const MAX_APP_DATA: usize = 16640; // RFC 8446 §5.2 upper bound const MAX_APP_DATA: usize = 16640; // RFC 8446 §5.2 upper bound
let fake_cert_len = fake_cert_len.max(MIN_APP_DATA).min(MAX_APP_DATA); let fake_cert_len = fake_cert_len.clamp(MIN_APP_DATA, MAX_APP_DATA);
let x25519_key = gen_fake_x25519_key(rng); let x25519_key = gen_fake_x25519_key(rng);
// Build ServerHello // Build ServerHello
@ -525,10 +525,10 @@ pub fn extract_sni_from_client_hello(handshake: &[u8]) -> Option<String> {
if sn_pos + name_len > sn_end { if sn_pos + name_len > sn_end {
break; break;
} }
if name_type == 0 && name_len > 0 { if name_type == 0 && name_len > 0
if let Ok(host) = std::str::from_utf8(&handshake[sn_pos..sn_pos + name_len]) { && let Ok(host) = std::str::from_utf8(&handshake[sn_pos..sn_pos + name_len])
return Some(host.to_string()); {
} return Some(host.to_string());
} }
sn_pos += name_len; sn_pos += name_len;
} }
@ -571,7 +571,7 @@ pub fn extract_alpn_from_client_hello(handshake: &[u8]) -> Vec<Vec<u8>> {
let list_len = u16::from_be_bytes([handshake[pos], handshake[pos+1]]) as usize; let list_len = u16::from_be_bytes([handshake[pos], handshake[pos+1]]) as usize;
let mut lp = pos + 2; let mut lp = pos + 2;
let list_end = (pos + 2).saturating_add(list_len).min(pos + elen); let list_end = (pos + 2).saturating_add(list_len).min(pos + elen);
while lp + 1 <= list_end { while lp < list_end {
let plen = handshake[lp] as usize; let plen = handshake[lp] as usize;
lp += 1; lp += 1;
if lp + plen > list_end { break; } if lp + plen > list_end { break; }

38
src/proxy/client.rs
View File

@ -594,18 +594,18 @@ impl RunningClientHandler {
peer_addr: SocketAddr, peer_addr: SocketAddr,
ip_tracker: &UserIpTracker, ip_tracker: &UserIpTracker,
) -> Result<()> { ) -> Result<()> {
if let Some(expiration) = config.access.user_expirations.get(user) { if let Some(expiration) = config.access.user_expirations.get(user)
if chrono::Utc::now() > *expiration { && chrono::Utc::now() > *expiration
return Err(ProxyError::UserExpired { {
user: user.to_string(), return Err(ProxyError::UserExpired {
}); user: user.to_string(),
} });
} }
// IP limit check // IP limit check
if let Err(reason) = ip_tracker.check_and_add(user, peer_addr.ip()).await { if let Err(reason) = ip_tracker.check_and_add(user, peer_addr.ip()).await {
warn!( warn!(
user = %user, user = %user,
ip = %peer_addr.ip(), ip = %peer_addr.ip(),
reason = %reason, reason = %reason,
"IP limit exceeded" "IP limit exceeded"
@ -615,20 +615,20 @@ impl RunningClientHandler {
}); });
} }
if let Some(limit) = config.access.user_max_tcp_conns.get(user) { if let Some(limit) = config.access.user_max_tcp_conns.get(user)
if stats.get_user_curr_connects(user) >= *limit as u64 { && stats.get_user_curr_connects(user) >= *limit as u64
return Err(ProxyError::ConnectionLimitExceeded { {
user: user.to_string(), return Err(ProxyError::ConnectionLimitExceeded {
}); user: user.to_string(),
} });
} }
if let Some(quota) = config.access.user_data_quota.get(user) { if let Some(quota) = config.access.user_data_quota.get(user)
if stats.get_user_total_octets(user) >= *quota { && stats.get_user_total_octets(user) >= *quota
return Err(ProxyError::DataQuotaExceeded { {
user: user.to_string(), return Err(ProxyError::DataQuotaExceeded {
}); user: user.to_string(),
} });
} }
Ok(()) Ok(())

8
src/proxy/direct_relay.rs
View File

@ -118,10 +118,10 @@ fn get_dc_addr_static(dc_idx: i16, config: &ProxyConfig) -> Result<SocketAddr> {
// Unknown DC requested by client without override: log and fall back. // Unknown DC requested by client without override: log and fall back.
if !config.dc_overrides.contains_key(&dc_key) { if !config.dc_overrides.contains_key(&dc_key) {
warn!(dc_idx = dc_idx, "Requested non-standard DC with no override; falling back to default cluster"); warn!(dc_idx = dc_idx, "Requested non-standard DC with no override; falling back to default cluster");
if let Some(path) = &config.general.unknown_dc_log_path { if let Some(path) = &config.general.unknown_dc_log_path
if let Ok(mut file) = OpenOptions::new().create(true).append(true).open(path) { && let Ok(mut file) = OpenOptions::new().create(true).append(true).open(path)
let _ = writeln!(file, "dc_idx={dc_idx}"); {
} let _ = writeln!(file, "dc_idx={dc_idx}");
} }
} }

10
src/proxy/masking.rs
View File

@ -19,12 +19,12 @@ const MASK_BUFFER_SIZE: usize = 8192;
/// Detect client type based on initial data /// Detect client type based on initial data
fn detect_client_type(data: &[u8]) -> &'static str { fn detect_client_type(data: &[u8]) -> &'static str {
// Check for HTTP request // Check for HTTP request
if data.len() > 4 { if data.len() > 4
if data.starts_with(b"GET ") || data.starts_with(b"POST") || && (data.starts_with(b"GET ") || data.starts_with(b"POST") ||
data.starts_with(b"HEAD") || data.starts_with(b"PUT ") || data.starts_with(b"HEAD") || data.starts_with(b"PUT ") ||
data.starts_with(b"DELETE") || data.starts_with(b"OPTIONS") { data.starts_with(b"DELETE") || data.starts_with(b"OPTIONS"))
return "HTTP"; {
} return "HTTP";
} }
// Check for TLS ClientHello (0x16 = handshake, 0x03 0x01-0x03 = TLS version) // Check for TLS ClientHello (0x16 = handshake, 0x03 0x01-0x03 = TLS version)

18
src/proxy/middle_relay.rs
View File

@ -393,13 +393,13 @@ where
.unwrap_or_else(|e| Err(ProxyError::Proxy(format!("ME writer join error: {e}")))); .unwrap_or_else(|e| Err(ProxyError::Proxy(format!("ME writer join error: {e}"))));
// When client closes, but ME channel stopped as unregistered - it isnt error // When client closes, but ME channel stopped as unregistered - it isnt error
if client_closed { if client_closed
if matches!( && matches!(
writer_result, writer_result,
Err(ProxyError::Proxy(ref msg)) if msg == "ME connection lost" Err(ProxyError::Proxy(ref msg)) if msg == "ME connection lost"
) { )
writer_result = Ok(()); {
} writer_result = Ok(());
} }
let result = match (main_result, c2me_result, writer_result) { let result = match (main_result, c2me_result, writer_result) {
@ -549,7 +549,7 @@ where
match proto_tag { match proto_tag {
ProtoTag::Abridged => { ProtoTag::Abridged => {
if data.len() % 4 != 0 { if !data.len().is_multiple_of(4) {
return Err(ProxyError::Proxy(format!( return Err(ProxyError::Proxy(format!(
"Abridged payload must be 4-byte aligned, got {}", "Abridged payload must be 4-byte aligned, got {}",
data.len() data.len()
@ -567,7 +567,7 @@ where
frame_buf.push(first); frame_buf.push(first);
frame_buf.extend_from_slice(data); frame_buf.extend_from_slice(data);
client_writer client_writer
.write_all(&frame_buf) .write_all(frame_buf)
.await .await
.map_err(ProxyError::Io)?; .map_err(ProxyError::Io)?;
} else if len_words < (1 << 24) { } else if len_words < (1 << 24) {
@ -581,7 +581,7 @@ where
frame_buf.extend_from_slice(&[first, lw[0], lw[1], lw[2]]); frame_buf.extend_from_slice(&[first, lw[0], lw[1], lw[2]]);
frame_buf.extend_from_slice(data); frame_buf.extend_from_slice(data);
client_writer client_writer
.write_all(&frame_buf) .write_all(frame_buf)
.await .await
.map_err(ProxyError::Io)?; .map_err(ProxyError::Io)?;
} else { } else {
@ -618,7 +618,7 @@ where
rng.fill(&mut frame_buf[start..]); rng.fill(&mut frame_buf[start..]);
} }
client_writer client_writer
.write_all(&frame_buf) .write_all(frame_buf)
.await .await
.map_err(ProxyError::Io)?; .map_err(ProxyError::Io)?;
} }

8
src/stats/mod.rs
View File

@ -326,10 +326,10 @@ impl ReplayShard {
// Use key.as_ref() to get &[u8] — avoids Borrow<Q> ambiguity // Use key.as_ref() to get &[u8] — avoids Borrow<Q> ambiguity
// between Borrow<[u8]> and Borrow<Box<[u8]>> // between Borrow<[u8]> and Borrow<Box<[u8]>>
if let Some(entry) = self.cache.peek(key.as_ref()) { if let Some(entry) = self.cache.peek(key.as_ref())
if entry.seq == queue_seq { && entry.seq == queue_seq
self.cache.pop(key.as_ref()); {
} self.cache.pop(key.as_ref());
} }
} }
} }

13
src/stream/crypto_stream.rs
View File

@ -47,7 +47,7 @@
//! - when upstream is Pending but pending still has room: accept `to_accept` bytes and //! - when upstream is Pending but pending still has room: accept `to_accept` bytes and
//! encrypt+append ciphertext directly into pending (in-place encryption of appended range) //! encrypt+append ciphertext directly into pending (in-place encryption of appended range)
//! Encrypted stream wrappers using AES-CTR //! Encrypted stream wrappers using AES-CTR
//! //!
//! This module provides stateful async stream wrappers that handle //! This module provides stateful async stream wrappers that handle
//! encryption/decryption with proper partial read/write handling. //! encryption/decryption with proper partial read/write handling.
@ -153,9 +153,9 @@ impl<R> CryptoReader<R> {
fn take_poison_error(&mut self) -> io::Error { fn take_poison_error(&mut self) -> io::Error {
match &mut self.state { match &mut self.state {
CryptoReaderState::Poisoned { error } => error.take().unwrap_or_else(|| { CryptoReaderState::Poisoned { error } => error.take().unwrap_or_else(|| {
io::Error::new(ErrorKind::Other, "stream previously poisoned") io::Error::other("stream previously poisoned")
}), }),
_ => io::Error::new(ErrorKind::Other, "stream not poisoned"), _ => io::Error::other("stream not poisoned"),
} }
} }
} }
@ -168,6 +168,7 @@ impl<R: AsyncRead + Unpin> AsyncRead for CryptoReader<R> {
) -> Poll<Result<()>> { ) -> Poll<Result<()>> {
let this = self.get_mut(); let this = self.get_mut();
#[allow(clippy::never_loop)]
loop { loop {
match &mut this.state { match &mut this.state {
CryptoReaderState::Poisoned { .. } => { CryptoReaderState::Poisoned { .. } => {
@ -485,14 +486,14 @@ impl<W> CryptoWriter<W> {
fn take_poison_error(&mut self) -> io::Error { fn take_poison_error(&mut self) -> io::Error {
match &mut self.state { match &mut self.state {
CryptoWriterState::Poisoned { error } => error.take().unwrap_or_else(|| { CryptoWriterState::Poisoned { error } => error.take().unwrap_or_else(|| {
io::Error::new(ErrorKind::Other, "stream previously poisoned") io::Error::other("stream previously poisoned")
}), }),
_ => io::Error::new(ErrorKind::Other, "stream not poisoned"), _ => io::Error::other("stream not poisoned"),
} }
} }
/// Ensure we are in Flushing state and return mutable pending buffer. /// Ensure we are in Flushing state and return mutable pending buffer.
fn ensure_pending<'a>(state: &'a mut CryptoWriterState, max_pending: usize) -> &'a mut PendingCiphertext { fn ensure_pending(state: &mut CryptoWriterState, max_pending: usize) -> &mut PendingCiphertext {
if matches!(state, CryptoWriterState::Idle) { if matches!(state, CryptoWriterState::Idle) {
*state = CryptoWriterState::Flushing { *state = CryptoWriterState::Flushing {
pending: PendingCiphertext::new(max_pending), pending: PendingCiphertext::new(max_pending),

2
src/stream/frame_codec.rs
View File

@ -139,7 +139,7 @@ fn encode_abridged(frame: &Frame, dst: &mut BytesMut) -> io::Result<()> {
let data = &frame.data; let data = &frame.data;
// Validate alignment // Validate alignment
if data.len() % 4 != 0 { if !data.len().is_multiple_of(4) {
return Err(Error::new( return Err(Error::new(
ErrorKind::InvalidInput, ErrorKind::InvalidInput,
format!("abridged frame must be 4-byte aligned, got {} bytes", data.len()) format!("abridged frame must be 4-byte aligned, got {} bytes", data.len())

4
src/stream/frame_stream.rs
View File

@ -78,7 +78,7 @@ impl<W> AbridgedFrameWriter<W> {
impl<W: AsyncWrite + Unpin> AbridgedFrameWriter<W> { impl<W: AsyncWrite + Unpin> AbridgedFrameWriter<W> {
/// Write a frame /// Write a frame
pub async fn write_frame(&mut self, data: &[u8], meta: &FrameMeta) -> Result<()> { pub async fn write_frame(&mut self, data: &[u8], meta: &FrameMeta) -> Result<()> {
if data.len() % 4 != 0 { if !data.len().is_multiple_of(4) {
return Err(Error::new( return Err(Error::new(
ErrorKind::InvalidInput, ErrorKind::InvalidInput,
format!("Abridged frame must be aligned to 4 bytes, got {}", data.len()), format!("Abridged frame must be aligned to 4 bytes, got {}", data.len()),
@ -331,7 +331,7 @@ impl<R: AsyncRead + Unpin> MtprotoFrameReader<R> {
} }
// Validate length // Validate length
if len < MIN_MSG_LEN || len > MAX_MSG_LEN || len % PADDING_FILLER.len() != 0 { if !(MIN_MSG_LEN..=MAX_MSG_LEN).contains(&len) || !len.is_multiple_of(PADDING_FILLER.len()) {
return Err(Error::new( return Err(Error::new(
ErrorKind::InvalidData, ErrorKind::InvalidData,
format!("Invalid message length: {}", len), format!("Invalid message length: {}", len),

16
src/stream/tls_stream.rs
View File

@ -135,7 +135,7 @@ impl TlsRecordHeader {
} }
/// Build header bytes /// Build header bytes
fn to_bytes(&self) -> [u8; 5] { fn to_bytes(self) -> [u8; 5] {
[ [
self.record_type, self.record_type,
self.version[0], self.version[0],
@ -260,9 +260,9 @@ impl<R> FakeTlsReader<R> {
fn take_poison_error(&mut self) -> io::Error { fn take_poison_error(&mut self) -> io::Error {
match &mut self.state { match &mut self.state {
TlsReaderState::Poisoned { error } => error.take().unwrap_or_else(|| { TlsReaderState::Poisoned { error } => error.take().unwrap_or_else(|| {
io::Error::new(ErrorKind::Other, "stream previously poisoned") io::Error::other("stream previously poisoned")
}), }),
_ => io::Error::new(ErrorKind::Other, "stream not poisoned"), _ => io::Error::other("stream not poisoned"),
} }
} }
} }
@ -297,7 +297,7 @@ impl<R: AsyncRead + Unpin> AsyncRead for FakeTlsReader<R> {
TlsReaderState::Poisoned { error } => { TlsReaderState::Poisoned { error } => {
this.state = TlsReaderState::Poisoned { error: None }; this.state = TlsReaderState::Poisoned { error: None };
let err = error.unwrap_or_else(|| { let err = error.unwrap_or_else(|| {
io::Error::new(ErrorKind::Other, "stream previously poisoned") io::Error::other("stream previously poisoned")
}); });
return Poll::Ready(Err(err)); return Poll::Ready(Err(err));
} }
@ -616,9 +616,9 @@ impl<W> FakeTlsWriter<W> {
fn take_poison_error(&mut self) -> io::Error { fn take_poison_error(&mut self) -> io::Error {
match &mut self.state { match &mut self.state {
TlsWriterState::Poisoned { error } => error.take().unwrap_or_else(|| { TlsWriterState::Poisoned { error } => error.take().unwrap_or_else(|| {
io::Error::new(ErrorKind::Other, "stream previously poisoned") io::Error::other("stream previously poisoned")
}), }),
_ => io::Error::new(ErrorKind::Other, "stream not poisoned"), _ => io::Error::other("stream not poisoned"),
} }
} }
@ -682,7 +682,7 @@ impl<W: AsyncWrite + Unpin> AsyncWrite for FakeTlsWriter<W> {
TlsWriterState::Poisoned { error } => { TlsWriterState::Poisoned { error } => {
this.state = TlsWriterState::Poisoned { error: None }; this.state = TlsWriterState::Poisoned { error: None };
let err = error.unwrap_or_else(|| { let err = error.unwrap_or_else(|| {
Error::new(ErrorKind::Other, "stream previously poisoned") Error::other("stream previously poisoned")
}); });
return Poll::Ready(Err(err)); return Poll::Ready(Err(err));
} }
@ -771,7 +771,7 @@ impl<W: AsyncWrite + Unpin> AsyncWrite for FakeTlsWriter<W> {
TlsWriterState::Poisoned { error } => { TlsWriterState::Poisoned { error } => {
this.state = TlsWriterState::Poisoned { error: None }; this.state = TlsWriterState::Poisoned { error: None };
let err = error.unwrap_or_else(|| { let err = error.unwrap_or_else(|| {
Error::new(ErrorKind::Other, "stream previously poisoned") Error::other("stream previously poisoned")
}); });
return Poll::Ready(Err(err)); return Poll::Ready(Err(err));
} }

50
src/tls_front/cache.rs
View File

@ -115,32 +115,32 @@ impl TlsFrontCache {
if !name.ends_with(".json") { if !name.ends_with(".json") {
continue; continue;
} }
if let Ok(data) = tokio::fs::read(entry.path()).await { if let Ok(data) = tokio::fs::read(entry.path()).await
if let Ok(mut cached) = serde_json::from_slice::<CachedTlsData>(&data) { && let Ok(mut cached) = serde_json::from_slice::<CachedTlsData>(&data)
if cached.domain.is_empty() {
|| cached.domain.len() > 255 if cached.domain.is_empty()
|| !cached.domain.chars().all(|c| c.is_ascii_alphanumeric() || c == '.' || c == '-') || cached.domain.len() > 255
{ || !cached.domain.chars().all(|c| c.is_ascii_alphanumeric() || c == '.' || c == '-')
warn!(file = %name, "Skipping TLS cache entry with invalid domain"); {
continue; warn!(file = %name, "Skipping TLS cache entry with invalid domain");
} continue;
// fetched_at is skipped during deserialization; approximate with file mtime if available.
if let Ok(meta) = entry.metadata().await {
if let Ok(modified) = meta.modified() {
cached.fetched_at = modified;
}
}
// Drop entries older than 72h
if let Ok(age) = cached.fetched_at.elapsed() {
if age > Duration::from_secs(72 * 3600) {
warn!(domain = %cached.domain, "Skipping stale TLS cache entry (>72h)");
continue;
}
}
let domain = cached.domain.clone();
self.set(&domain, cached).await;
loaded += 1;
} }
// fetched_at is skipped during deserialization; approximate with file mtime if available.
if let Ok(meta) = entry.metadata().await
&& let Ok(modified) = meta.modified()
{
cached.fetched_at = modified;
}
// Drop entries older than 72h
if let Ok(age) = cached.fetched_at.elapsed()
&& age > Duration::from_secs(72 * 3600)
{
warn!(domain = %cached.domain, "Skipping stale TLS cache entry (>72h)");
continue;
}
let domain = cached.domain.clone();
self.set(&domain, cached).await;
loaded += 1;
} }
} }
} }

20
src/tls_front/emulator.rs
View File

@ -12,7 +12,7 @@ fn jitter_and_clamp_sizes(sizes: &[usize], rng: &SecureRandom) -> Vec<usize> {
sizes sizes
.iter() .iter()
.map(|&size| { .map(|&size| {
let base = size.max(MIN_APP_DATA).min(MAX_APP_DATA); let base = size.clamp(MIN_APP_DATA, MAX_APP_DATA);
let jitter_range = ((base as f64) * 0.03).round() as i64; let jitter_range = ((base as f64) * 0.03).round() as i64;
if jitter_range == 0 { if jitter_range == 0 {
return base; return base;
@ -50,7 +50,7 @@ fn ensure_payload_capacity(mut sizes: Vec<usize>, payload_len: usize) -> Vec<usi
while body_total < payload_len { while body_total < payload_len {
let remaining = payload_len - body_total; let remaining = payload_len - body_total;
let chunk = (remaining + 17).min(MAX_APP_DATA).max(MIN_APP_DATA); let chunk = (remaining + 17).clamp(MIN_APP_DATA, MAX_APP_DATA);
sizes.push(chunk); sizes.push(chunk);
body_total += chunk.saturating_sub(17); body_total += chunk.saturating_sub(17);
} }
@ -189,7 +189,7 @@ pub fn build_emulated_server_hello(
.as_ref() .as_ref()
.map(|payload| payload.certificate_message.as_slice()) .map(|payload| payload.certificate_message.as_slice())
.filter(|payload| !payload.is_empty()) .filter(|payload| !payload.is_empty())
.or_else(|| compact_payload.as_deref()) .or(compact_payload.as_deref())
} else { } else {
compact_payload.as_deref() compact_payload.as_deref()
}; };
@ -223,15 +223,13 @@ pub fn build_emulated_server_hello(
} else { } else {
rec.extend_from_slice(&rng.bytes(size)); rec.extend_from_slice(&rng.bytes(size));
} }
} else if size > 17 {
let body_len = size - 17;
rec.extend_from_slice(&rng.bytes(body_len));
rec.push(0x16); // inner content type marker (handshake)
rec.extend_from_slice(&rng.bytes(16)); // AEAD-like tag
} else { } else {
if size > 17 { rec.extend_from_slice(&rng.bytes(size));
let body_len = size - 17;
rec.extend_from_slice(&rng.bytes(body_len));
rec.push(0x16); // inner content type marker (handshake)
rec.extend_from_slice(&rng.bytes(16)); // AEAD-like tag
} else {
rec.extend_from_slice(&rng.bytes(size));
}
} }
app_data.extend_from_slice(&rec); app_data.extend_from_slice(&rec);
} }

2
src/tls_front/fetcher.rs
View File

@ -384,7 +384,7 @@ async fn fetch_via_raw_tls(
for _ in 0..4 { for _ in 0..4 {
match timeout(connect_timeout, read_tls_record(&mut stream)).await { match timeout(connect_timeout, read_tls_record(&mut stream)).await {
Ok(Ok(rec)) => records.push(rec), Ok(Ok(rec)) => records.push(rec),
Ok(Err(e)) => return Err(e.into()), Ok(Err(e)) => return Err(e),
Err(_) => break, Err(_) => break,
} }
if records.len() >= 3 && records.iter().any(|(t, _)| *t == TLS_RECORD_APPLICATION) { if records.len() >= 3 && records.iter().any(|(t, _)| *t == TLS_RECORD_APPLICATION) {

5
src/transport/middle_proxy/codec.rs
View File

@ -165,11 +165,10 @@ fn process_pid16() -> u16 {
} }
fn process_utime() -> u32 { fn process_utime() -> u32 {
let utime = std::time::SystemTime::now() std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs() as u32; .as_secs() as u32
utime
} }
pub(crate) fn cbc_encrypt_padded( pub(crate) fn cbc_encrypt_padded(

44
src/transport/middle_proxy/config_updater.rs
View File

@ -40,14 +40,16 @@ pub struct ProxyConfigData {
} }
fn parse_host_port(s: &str) -> Option<(IpAddr, u16)> { fn parse_host_port(s: &str) -> Option<(IpAddr, u16)> {
if let Some(bracket_end) = s.rfind(']') { if let Some(bracket_end) = s.rfind(']')
if s.starts_with('[') && bracket_end + 1 < s.len() && s.as_bytes().get(bracket_end + 1) == Some(&b':') { && s.starts_with('[')
let host = &s[1..bracket_end]; && bracket_end + 1 < s.len()
let port_str = &s[bracket_end + 2..]; && s.as_bytes().get(bracket_end + 1) == Some(&b':')
let ip = host.parse::<IpAddr>().ok()?; {
let port = port_str.parse::<u16>().ok()?; let host = &s[1..bracket_end];
return Some((ip, port)); let port_str = &s[bracket_end + 2..];
} let ip = host.parse::<IpAddr>().ok()?;
let port = port_str.parse::<u16>().ok()?;
return Some((ip, port));
} }
let idx = s.rfind(':')?; let idx = s.rfind(':')?;
@ -84,20 +86,18 @@ pub async fn fetch_proxy_config(url: &str) -> Result<ProxyConfigData> {
.map_err(|e| crate::error::ProxyError::Proxy(format!("fetch_proxy_config GET failed: {e}")))? .map_err(|e| crate::error::ProxyError::Proxy(format!("fetch_proxy_config GET failed: {e}")))?
; ;
if let Some(date) = resp.headers().get(reqwest::header::DATE) { if let Some(date) = resp.headers().get(reqwest::header::DATE)
if let Ok(date_str) = date.to_str() { && let Ok(date_str) = date.to_str()
if let Ok(server_time) = httpdate::parse_http_date(date_str) { && let Ok(server_time) = httpdate::parse_http_date(date_str)
if let Ok(skew) = SystemTime::now().duration_since(server_time).or_else(|e| { && let Ok(skew) = SystemTime::now().duration_since(server_time).or_else(|e| {
server_time.duration_since(SystemTime::now()).map_err(|_| e) server_time.duration_since(SystemTime::now()).map_err(|_| e)
}) { })
let skew_secs = skew.as_secs(); {
if skew_secs > 60 { let skew_secs = skew.as_secs();
warn!(skew_secs, "Time skew >60s detected from fetch_proxy_config Date header"); if skew_secs > 60 {
} else if skew_secs > 30 { warn!(skew_secs, "Time skew >60s detected from fetch_proxy_config Date header");
warn!(skew_secs, "Time skew >30s detected from fetch_proxy_config Date header"); } else if skew_secs > 30 {
} warn!(skew_secs, "Time skew >30s detected from fetch_proxy_config Date header");
}
}
} }
} }

24
src/transport/middle_proxy/handshake.rs
View File

@ -47,21 +47,21 @@ impl MePool {
pub(crate) async fn connect_tcp(&self, addr: SocketAddr) -> Result<(TcpStream, f64)> { pub(crate) async fn connect_tcp(&self, addr: SocketAddr) -> Result<(TcpStream, f64)> {
let start = Instant::now(); let start = Instant::now();
let connect_fut = async { let connect_fut = async {
if addr.is_ipv6() { if addr.is_ipv6()
if let Some(v6) = self.detected_ipv6 { && let Some(v6) = self.detected_ipv6
match TcpSocket::new_v6() { {
Ok(sock) => { match TcpSocket::new_v6() {
if let Err(e) = sock.bind(SocketAddr::new(IpAddr::V6(v6), 0)) { Ok(sock) => {
debug!(error = %e, bind_ip = %v6, "ME IPv6 bind failed, falling back to default bind"); if let Err(e) = sock.bind(SocketAddr::new(IpAddr::V6(v6), 0)) {
} else { debug!(error = %e, bind_ip = %v6, "ME IPv6 bind failed, falling back to default bind");
match sock.connect(addr).await { } else {
Ok(stream) => return Ok(stream), match sock.connect(addr).await {
Err(e) => debug!(error = %e, target = %addr, "ME IPv6 bound connect failed, retrying default connect"), Ok(stream) => return Ok(stream),
} Err(e) => debug!(error = %e, target = %addr, "ME IPv6 bound connect failed, retrying default connect"),
} }
} }
Err(e) => debug!(error = %e, "ME IPv6 socket creation failed, falling back to default connect"),
} }
Err(e) => debug!(error = %e, "ME IPv6 socket creation failed, falling back to default connect"),
} }
} }
TcpStream::connect(addr).await TcpStream::connect(addr).await

8
src/transport/middle_proxy/health.rs
View File

@ -92,10 +92,10 @@ async fn check_family(
let key = (dc, family); let key = (dc, family);
let now = Instant::now(); let now = Instant::now();
if let Some(ts) = next_attempt.get(&key) { if let Some(ts) = next_attempt.get(&key)
if now < *ts { && now < *ts
continue; {
} continue;
} }
let max_concurrent = pool.me_reconnect_max_concurrent_per_dc.max(1) as usize; let max_concurrent = pool.me_reconnect_max_concurrent_per_dc.max(1) as usize;

62
src/transport/middle_proxy/pool.rs
View File

@ -498,10 +498,10 @@ impl MePool {
let mut guard = self.proxy_map_v4.write().await; let mut guard = self.proxy_map_v4.write().await;
let keys: Vec<i32> = guard.keys().cloned().collect(); let keys: Vec<i32> = guard.keys().cloned().collect();
for k in keys.iter().cloned().filter(|k| *k > 0) { for k in keys.iter().cloned().filter(|k| *k > 0) {
if !guard.contains_key(&-k) { if !guard.contains_key(&-k)
if let Some(addrs) = guard.get(&k).cloned() { && let Some(addrs) = guard.get(&k).cloned()
guard.insert(-k, addrs); {
} guard.insert(-k, addrs);
} }
} }
} }
@ -509,10 +509,10 @@ impl MePool {
let mut guard = self.proxy_map_v6.write().await; let mut guard = self.proxy_map_v6.write().await;
let keys: Vec<i32> = guard.keys().cloned().collect(); let keys: Vec<i32> = guard.keys().cloned().collect();
for k in keys.iter().cloned().filter(|k| *k > 0) { for k in keys.iter().cloned().filter(|k| *k > 0) {
if !guard.contains_key(&-k) { if !guard.contains_key(&-k)
if let Some(addrs) = guard.get(&k).cloned() { && let Some(addrs) = guard.get(&k).cloned()
guard.insert(-k, addrs); {
} guard.insert(-k, addrs);
} }
} }
} }
@ -760,13 +760,12 @@ impl MePool {
cancel_reader_token.clone(), cancel_reader_token.clone(),
) )
.await; .await;
if let Some(pool) = pool.upgrade() { if let Some(pool) = pool.upgrade()
if cleanup_for_reader && cleanup_for_reader
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Relaxed) .compare_exchange(false, true, Ordering::AcqRel, Ordering::Relaxed)
.is_ok() .is_ok()
{ {
pool.remove_writer_and_close_clients(writer_id).await; pool.remove_writer_and_close_clients(writer_id).await;
}
} }
if let Err(e) = res { if let Err(e) = res {
warn!(error = %e, "ME reader ended"); warn!(error = %e, "ME reader ended");
@ -834,13 +833,12 @@ impl MePool {
stats_ping.increment_me_keepalive_failed(); stats_ping.increment_me_keepalive_failed();
debug!("ME ping failed, removing dead writer"); debug!("ME ping failed, removing dead writer");
cancel_ping.cancel(); cancel_ping.cancel();
if let Some(pool) = pool_ping.upgrade() { if let Some(pool) = pool_ping.upgrade()
if cleanup_for_ping && cleanup_for_ping
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Relaxed) .compare_exchange(false, true, Ordering::AcqRel, Ordering::Relaxed)
.is_ok() .is_ok()
{ {
pool.remove_writer_and_close_clients(writer_id).await; pool.remove_writer_and_close_clients(writer_id).await;
}
} }
break; break;
} }
@ -943,24 +941,20 @@ impl MePool {
let pool = Arc::downgrade(self); let pool = Arc::downgrade(self);
tokio::spawn(async move { tokio::spawn(async move {
let deadline = timeout.map(|t| Instant::now() + t); let deadline = timeout.map(|t| Instant::now() + t);
loop { while let Some(p) = pool.upgrade() {
if let Some(p) = pool.upgrade() { if let Some(deadline_at) = deadline
if let Some(deadline_at) = deadline { && Instant::now() >= deadline_at
if Instant::now() >= deadline_at { {
warn!(writer_id, "Drain timeout, force-closing"); warn!(writer_id, "Drain timeout, force-closing");
p.stats.increment_pool_force_close_total(); p.stats.increment_pool_force_close_total();
let _ = p.remove_writer_and_close_clients(writer_id).await; let _ = p.remove_writer_and_close_clients(writer_id).await;
break;
}
}
if p.registry.is_writer_empty(writer_id).await {
let _ = p.remove_writer_only(writer_id).await;
break;
}
tokio::time::sleep(Duration::from_secs(1)).await;
} else {
break; break;
} }
if p.registry.is_writer_empty(writer_id).await {
let _ = p.remove_writer_only(writer_id).await;
break;
}
tokio::time::sleep(Duration::from_secs(1)).await;
} }
}); });
} }

32
src/transport/middle_proxy/pool_nat.rs
View File

@ -25,7 +25,7 @@ impl MePool {
pub(super) fn translate_ip_for_nat(&self, ip: IpAddr) -> IpAddr { pub(super) fn translate_ip_for_nat(&self, ip: IpAddr) -> IpAddr {
let nat_ip = self let nat_ip = self
.nat_ip_cfg .nat_ip_cfg
.or_else(|| self.nat_ip_detected.try_read().ok().and_then(|g| (*g).clone())); .or_else(|| self.nat_ip_detected.try_read().ok().and_then(|g| *g));
let Some(nat_ip) = nat_ip else { let Some(nat_ip) = nat_ip else {
return ip; return ip;
@ -75,7 +75,7 @@ impl MePool {
return None; return None;
} }
if let Some(ip) = self.nat_ip_detected.read().await.clone() { if let Some(ip) = *self.nat_ip_detected.read().await {
return Some(ip); return Some(ip);
} }
@ -102,17 +102,17 @@ impl MePool {
) -> Option<std::net::SocketAddr> { ) -> Option<std::net::SocketAddr> {
const STUN_CACHE_TTL: Duration = Duration::from_secs(600); const STUN_CACHE_TTL: Duration = Duration::from_secs(600);
// Backoff window // Backoff window
if let Some(until) = *self.stun_backoff_until.read().await { if let Some(until) = *self.stun_backoff_until.read().await
if Instant::now() < until { && Instant::now() < until
if let Ok(cache) = self.nat_reflection_cache.try_lock() { {
let slot = match family { if let Ok(cache) = self.nat_reflection_cache.try_lock() {
IpFamily::V4 => cache.v4, let slot = match family {
IpFamily::V6 => cache.v6, IpFamily::V4 => cache.v4,
}; IpFamily::V6 => cache.v6,
return slot.map(|(_, addr)| addr); };
} return slot.map(|(_, addr)| addr);
return None;
} }
return None;
} }
if let Ok(mut cache) = self.nat_reflection_cache.try_lock() { if let Ok(mut cache) = self.nat_reflection_cache.try_lock() {
@ -120,10 +120,10 @@ impl MePool {
IpFamily::V4 => &mut cache.v4, IpFamily::V4 => &mut cache.v4,
IpFamily::V6 => &mut cache.v6, IpFamily::V6 => &mut cache.v6,
}; };
if let Some((ts, addr)) = slot { if let Some((ts, addr)) = slot
if ts.elapsed() < STUN_CACHE_TTL { && ts.elapsed() < STUN_CACHE_TTL
return Some(*addr); {
} return Some(*addr);
} }
} }

26
src/transport/middle_proxy/secret.rs
View File

@ -63,20 +63,18 @@ pub async fn download_proxy_secret() -> Result<Vec<u8>> {
))); )));
} }
if let Some(date) = resp.headers().get(reqwest::header::DATE) { if let Some(date) = resp.headers().get(reqwest::header::DATE)
if let Ok(date_str) = date.to_str() { && let Ok(date_str) = date.to_str()
if let Ok(server_time) = httpdate::parse_http_date(date_str) { && let Ok(server_time) = httpdate::parse_http_date(date_str)
if let Ok(skew) = SystemTime::now().duration_since(server_time).or_else(|e| { && let Ok(skew) = SystemTime::now().duration_since(server_time).or_else(|e| {
server_time.duration_since(SystemTime::now()).map_err(|_| e) server_time.duration_since(SystemTime::now()).map_err(|_| e)
}) { })
let skew_secs = skew.as_secs(); {
if skew_secs > 60 { let skew_secs = skew.as_secs();
warn!(skew_secs, "Time skew >60s detected from proxy-secret Date header"); if skew_secs > 60 {
} else if skew_secs > 30 { warn!(skew_secs, "Time skew >60s detected from proxy-secret Date header");
warn!(skew_secs, "Time skew >30s detected from proxy-secret Date header"); } else if skew_secs > 30 {
} warn!(skew_secs, "Time skew >30s detected from proxy-secret Date header");
}
}
} }
} }

10
src/transport/middle_proxy/send.rs
View File

@ -242,10 +242,10 @@ impl MePool {
} }
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_guard.get(&def) { && 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)));
} }
} }
@ -267,7 +267,7 @@ impl MePool {
if !self.writer_accepts_new_binding(w) { if !self.writer_accepts_new_binding(w) {
continue; continue;
} }
if preferred.iter().any(|p| *p == w.addr) { if preferred.contains(&w.addr) {
out.push(idx); out.push(idx);
} }
} }

20
src/transport/socket.rs
View File

@ -136,17 +136,17 @@ pub fn resolve_interface_ip(name: &str, want_ipv6: bool) -> Option<IpAddr> {
if let Ok(addrs) = getifaddrs() { if let Ok(addrs) = getifaddrs() {
for iface in addrs { for iface in addrs {
if iface.interface_name == name { if iface.interface_name == name
if let Some(address) = iface.address { && let Some(address) = iface.address
if let Some(v4) = address.as_sockaddr_in() { {
if !want_ipv6 { if let Some(v4) = address.as_sockaddr_in() {
return Some(IpAddr::V4(v4.ip())); if !want_ipv6 {
} return Some(IpAddr::V4(v4.ip()));
} else if let Some(v6) = address.as_sockaddr_in6() {
if want_ipv6 {
return Some(IpAddr::V6(v6.ip().clone()));
}
} }
} else if let Some(v6) = address.as_sockaddr_in6()
&& want_ipv6
{
return Some(IpAddr::V6(v6.ip()));
} }
} }
} }

24
src/transport/socks.rs
View File

@ -27,11 +27,11 @@ pub async fn connect_socks4(
buf.extend_from_slice(user); buf.extend_from_slice(user);
buf.push(0); // NULL buf.push(0); // NULL
stream.write_all(&buf).await.map_err(|e| ProxyError::Io(e))?; stream.write_all(&buf).await.map_err(ProxyError::Io)?;
// Response: VN (1) | CD (1) | DSTPORT (2) | DSTIP (4) // Response: VN (1) | CD (1) | DSTPORT (2) | DSTIP (4)
let mut resp = [0u8; 8]; let mut resp = [0u8; 8];
stream.read_exact(&mut resp).await.map_err(|e| ProxyError::Io(e))?; stream.read_exact(&mut resp).await.map_err(ProxyError::Io)?;
if resp[1] != 90 { if resp[1] != 90 {
return Err(ProxyError::Proxy(format!("SOCKS4 request rejected: code {}", resp[1]))); return Err(ProxyError::Proxy(format!("SOCKS4 request rejected: code {}", resp[1])));
@ -56,10 +56,10 @@ pub async fn connect_socks5(
let mut buf = vec![5u8, methods.len() as u8]; let mut buf = vec![5u8, methods.len() as u8];
buf.extend_from_slice(&methods); buf.extend_from_slice(&methods);
stream.write_all(&buf).await.map_err(|e| ProxyError::Io(e))?; stream.write_all(&buf).await.map_err(ProxyError::Io)?;
let mut resp = [0u8; 2]; let mut resp = [0u8; 2];
stream.read_exact(&mut resp).await.map_err(|e| ProxyError::Io(e))?; stream.read_exact(&mut resp).await.map_err(ProxyError::Io)?;
if resp[0] != 5 { if resp[0] != 5 {
return Err(ProxyError::Proxy("Invalid SOCKS5 version".to_string())); return Err(ProxyError::Proxy("Invalid SOCKS5 version".to_string()));
@ -80,10 +80,10 @@ pub async fn connect_socks5(
auth_buf.push(p_bytes.len() as u8); auth_buf.push(p_bytes.len() as u8);
auth_buf.extend_from_slice(p_bytes); auth_buf.extend_from_slice(p_bytes);
stream.write_all(&auth_buf).await.map_err(|e| ProxyError::Io(e))?; stream.write_all(&auth_buf).await.map_err(ProxyError::Io)?;
let mut auth_resp = [0u8; 2]; let mut auth_resp = [0u8; 2];
stream.read_exact(&mut auth_resp).await.map_err(|e| ProxyError::Io(e))?; stream.read_exact(&mut auth_resp).await.map_err(ProxyError::Io)?;
if auth_resp[1] != 0 { if auth_resp[1] != 0 {
return Err(ProxyError::Proxy("SOCKS5 authentication failed".to_string())); return Err(ProxyError::Proxy("SOCKS5 authentication failed".to_string()));
@ -112,11 +112,11 @@ pub async fn connect_socks5(
req.extend_from_slice(&target.port().to_be_bytes()); req.extend_from_slice(&target.port().to_be_bytes());
stream.write_all(&req).await.map_err(|e| ProxyError::Io(e))?; stream.write_all(&req).await.map_err(ProxyError::Io)?;
// Response // Response
let mut head = [0u8; 4]; let mut head = [0u8; 4];
stream.read_exact(&mut head).await.map_err(|e| ProxyError::Io(e))?; stream.read_exact(&mut head).await.map_err(ProxyError::Io)?;
if head[1] != 0 { if head[1] != 0 {
return Err(ProxyError::Proxy(format!("SOCKS5 request failed: code {}", head[1]))); return Err(ProxyError::Proxy(format!("SOCKS5 request failed: code {}", head[1])));
@ -126,17 +126,17 @@ pub async fn connect_socks5(
match head[3] { match head[3] {
1 => { // IPv4 1 => { // IPv4
let mut addr = [0u8; 4 + 2]; let mut addr = [0u8; 4 + 2];
stream.read_exact(&mut addr).await.map_err(|e| ProxyError::Io(e))?; stream.read_exact(&mut addr).await.map_err(ProxyError::Io)?;
}, },
3 => { // Domain 3 => { // Domain
let mut len = [0u8; 1]; let mut len = [0u8; 1];
stream.read_exact(&mut len).await.map_err(|e| ProxyError::Io(e))?; stream.read_exact(&mut len).await.map_err(ProxyError::Io)?;
let mut addr = vec![0u8; len[0] as usize + 2]; let mut addr = vec![0u8; len[0] as usize + 2];
stream.read_exact(&mut addr).await.map_err(|e| ProxyError::Io(e))?; stream.read_exact(&mut addr).await.map_err(ProxyError::Io)?;
}, },
4 => { // IPv6 4 => { // IPv6
let mut addr = [0u8; 16 + 2]; let mut addr = [0u8; 16 + 2];
stream.read_exact(&mut addr).await.map_err(|e| ProxyError::Io(e))?; stream.read_exact(&mut addr).await.map_err(ProxyError::Io)?;
}, },
_ => return Err(ProxyError::Proxy("Invalid address type in SOCKS5 response".to_string())), _ => return Err(ProxyError::Proxy("Invalid address type in SOCKS5 response".to_string())),
} }

25
src/transport/upstream.rs
View File

@ -57,9 +57,10 @@ impl LatencyEma {
// ============= Per-DC IP Preference Tracking ============= // ============= Per-DC IP Preference Tracking =============
/// Tracks which IP version works for each DC /// Tracks which IP version works for each DC
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum IpPreference { pub enum IpPreference {
/// Not yet tested /// Not yet tested
#[default]
Unknown, Unknown,
/// IPv6 works /// IPv6 works
PreferV6, PreferV6,
@ -71,12 +72,6 @@ pub enum IpPreference {
Unavailable, Unavailable,
} }
impl Default for IpPreference {
fn default() -> Self {
Self::Unknown
}
}
// ============= Upstream State ============= // ============= Upstream State =============
#[derive(Debug)] #[derive(Debug)]
@ -112,7 +107,7 @@ impl UpstreamState {
if abs_dc == 0 { if abs_dc == 0 {
return None; return None;
} }
if abs_dc >= 1 && abs_dc <= NUM_DCS { if (1..=NUM_DCS).contains(&abs_dc) {
Some(abs_dc - 1) Some(abs_dc - 1)
} else { } else {
// Unknown DC → default cluster (DC 2, index 1) // Unknown DC → default cluster (DC 2, index 1)
@ -122,10 +117,10 @@ impl UpstreamState {
/// Get latency for a specific DC, falling back to average across all known DCs /// Get latency for a specific DC, falling back to average across all known DCs
fn effective_latency(&self, dc_idx: Option<i16>) -> Option<f64> { fn effective_latency(&self, dc_idx: Option<i16>) -> Option<f64> {
if let Some(di) = dc_idx.and_then(Self::dc_array_idx) { if let Some(di) = dc_idx.and_then(Self::dc_array_idx)
if let Some(ms) = self.dc_latency[di].get() { && let Some(ms) = self.dc_latency[di].get()
return Some(ms); {
} return Some(ms);
} }
let (sum, count) = self.dc_latency.iter() let (sum, count) = self.dc_latency.iter()
@ -582,7 +577,7 @@ impl UpstreamManager {
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, Some(bind_rr.clone()), addr_v6) self.ping_single_dc(upstream_config, Some(bind_rr.clone()), addr_v6)
).await; ).await;
let ping_result = match result { let ping_result = match result {
@ -633,7 +628,7 @@ impl UpstreamManager {
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, Some(bind_rr.clone()), addr_v4) self.ping_single_dc(upstream_config, Some(bind_rr.clone()), addr_v4)
).await; ).await;
let ping_result = match result { let ping_result = match result {
@ -696,7 +691,7 @@ impl UpstreamManager {
} }
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, Some(bind_rr.clone()), addr) self.ping_single_dc(upstream_config, Some(bind_rr.clone()), addr)
).await; ).await;
let ping_result = match result { let ping_result = match result {

62
src/util/ip.rs
View File

@ -67,54 +67,56 @@ pub async fn detect_ip() -> IpInfo {
// Try to get local interface IP first (default gateway interface) // Try to get local interface IP first (default gateway interface)
// We connect to Google DNS to find out which interface is used for routing // We connect to Google DNS to find out which interface is used for routing
if let Some(ip) = get_local_ip("8.8.8.8:80") { if let Some(ip) = get_local_ip("8.8.8.8:80")
if ip.is_ipv4() && !ip.is_loopback() { && ip.is_ipv4()
info.ipv4 = Some(ip); && !ip.is_loopback()
debug!(ip = %ip, "Detected local IPv4 address via routing"); {
} info.ipv4 = Some(ip);
debug!(ip = %ip, "Detected local IPv4 address via routing");
} }
if let Some(ip) = get_local_ipv6("[2001:4860:4860::8888]:80") { if let Some(ip) = get_local_ipv6("[2001:4860:4860::8888]:80")
if ip.is_ipv6() && !ip.is_loopback() { && ip.is_ipv6()
info.ipv6 = Some(ip); && !ip.is_loopback()
debug!(ip = %ip, "Detected local IPv6 address via routing"); {
} info.ipv6 = Some(ip);
debug!(ip = %ip, "Detected local IPv6 address via routing");
} }
// If local detection failed or returned private IP (and we want public), // If local detection failed or returned private IP (and we want public),
// or just as a fallback/verification, we might want to check external services. // or just as a fallback/verification, we might want to check external services.
// However, the requirement is: "if IP for listening is not set... it should be IP from interface... // However, the requirement is: "if IP for listening is not set... it should be IP from interface...
// if impossible - request external resources". // if impossible - request external resources".
// So if we found a local IP, we might be good. But often servers are behind NAT. // So if we found a local IP, we might be good. But often servers are behind NAT.
// If the local IP is private, we probably want the public IP for the tg:// link. // If the local IP is private, we probably want the public IP for the tg:// link.
// Let's check if the detected IPs are private. // Let's check if the detected IPs are private.
let need_external_v4 = info.ipv4.map_or(true, |ip| is_private_ip(ip)); let need_external_v4 = info.ipv4.is_none_or(is_private_ip);
let need_external_v6 = info.ipv6.map_or(true, |ip| is_private_ip(ip)); let need_external_v6 = info.ipv6.is_none_or(is_private_ip);
if need_external_v4 { if need_external_v4 {
debug!("Local IPv4 is private or missing, checking external services..."); debug!("Local IPv4 is private or missing, checking external services...");
for url in IPV4_URLS { for url in IPV4_URLS {
if let Some(ip) = fetch_ip(url).await { if let Some(ip) = fetch_ip(url).await
if ip.is_ipv4() { && ip.is_ipv4()
info.ipv4 = Some(ip); {
debug!(ip = %ip, "Detected public IPv4 address"); info.ipv4 = Some(ip);
break; debug!(ip = %ip, "Detected public IPv4 address");
} break;
} }
} }
} }
if need_external_v6 { if need_external_v6 {
debug!("Local IPv6 is private or missing, checking external services..."); debug!("Local IPv6 is private or missing, checking external services...");
for url in IPV6_URLS { for url in IPV6_URLS {
if let Some(ip) = fetch_ip(url).await { if let Some(ip) = fetch_ip(url).await
if ip.is_ipv6() { && ip.is_ipv6()
info.ipv6 = Some(ip); {
debug!(ip = %ip, "Detected public IPv6 address"); info.ipv6 = Some(ip);
break; debug!(ip = %ip, "Detected public IPv6 address");
} break;
} }
} }
} }

16
src/util/time.rs
View File

@ -67,15 +67,15 @@ pub async fn check_time_sync() -> Option<TimeSyncResult> {
#[allow(dead_code)] #[allow(dead_code)]
pub async fn time_sync_task(check_interval: Duration) -> ! { pub async fn time_sync_task(check_interval: Duration) -> ! {
loop { loop {
if let Some(result) = check_time_sync().await { if let Some(result) = check_time_sync().await
if result.is_skewed { && result.is_skewed
error!( {
"System clock is off by {} seconds. Please sync your clock.", error!(
result.skew_secs "System clock is off by {} seconds. Please sync your clock.",
); result.skew_secs
} );
} }
tokio::time::sleep(check_interval).await; tokio::time::sleep(check_interval).await;
} }
} }