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

Merge remote-tracking branch 'upstream/main'

This commit is contained in:
ivulit 2026-02-21 09:45:31 +03:00
parent ebeb3ddba1 1a525f7d29
commit 379a69a419
No known key found for this signature in database
18 changed files with 651 additions and 309 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
config.toml
View File

@ -124,6 +124,10 @@ hello = "00000000000000000000000000000000"
# [access.user_data_quota]
# hello = 1073741824 # 1 GB
# [access.user_expirations]
# format: username = "[year]-[month]-[day]T[hour]:[minute]:[second]Z" UTC
# hello = "2027-01-01T00:00:00Z"
# === Upstreams & Routing ===
[[upstreams]]
type = "direct"

69
src/config/defaults.rs
View File

@ -110,6 +110,75 @@ pub(crate) fn default_reconnect_backoff_cap_ms() -> u64 {
30_000
}
pub(crate) fn default_crypto_pending_buffer() -> usize {
256 * 1024
}
pub(crate) fn default_max_client_frame() -> usize {
16 * 1024 * 1024
}
pub(crate) fn default_tls_new_session_tickets() -> u8 {
0
}
pub(crate) fn default_server_hello_delay_min_ms() -> u64 {
0
}
pub(crate) fn default_server_hello_delay_max_ms() -> u64 {
0
}
pub(crate) fn default_alpn_enforce() -> bool {
true
}
pub(crate) fn default_stun_servers() -> Vec<String> {
vec![
"stun.l.google.com:19302".to_string(),
"stun1.l.google.com:19302".to_string(),
"stun2.l.google.com:19302".to_string(),
"stun.stunprotocol.org:3478".to_string(),
"stun.voip.eutelia.it:3478".to_string(),
]
}
pub(crate) fn default_http_ip_detect_urls() -> Vec<String> {
vec![
"https://ifconfig.me/ip".to_string(),
"https://api.ipify.org".to_string(),
]
}
pub(crate) fn default_cache_public_ip_path() -> String {
"cache/public_ip.txt".to_string()
}
pub(crate) fn default_proxy_secret_reload_secs() -> u64 {
12 * 60 * 60
}
pub(crate) fn default_proxy_config_reload_secs() -> u64 {
12 * 60 * 60
}
pub(crate) fn default_ntp_check() -> bool {
true
}
pub(crate) fn default_ntp_servers() -> Vec<String> {
vec!["pool.ntp.org".to_string()]
}
pub(crate) fn default_fast_mode_min_tls_record() -> usize {
0
}
pub(crate) fn default_degradation_min_unavailable_dc_groups() -> u8 {
2
}
// Custom deserializer helpers
#[derive(Deserialize)]

176
src/config/hot_reload.rs
View File

@ -316,118 +316,106 @@ pub fn spawn_config_watcher(
let (config_tx, config_rx) = watch::channel(initial);
let (log_tx, log_rx) = watch::channel(initial_level);
// Bridge: sync notify callback → async task via mpsc.
// Bridge: sync notify callbacks → async task via mpsc.
let (notify_tx, mut notify_rx) = mpsc::channel::<()>(4);
// Canonicalize the config path so it matches what notify returns in events
// (notify always gives absolute paths, but config_path may be relative).
// Canonicalize so path matches what notify returns (absolute) in events.
let config_path = match config_path.canonicalize() {
Ok(p) => p,
Err(_) => config_path.to_path_buf(), // file doesn't exist yet, use as-is
Err(_) => config_path.to_path_buf(),
};
// Watch the parent directory rather than the file itself, because many
// editors (vim, nano, systemd-sysusers) write via rename, which would
// cause inotify to lose track of the original inode.
// editors (vim, nano) and systemd write via rename, which would cause
// inotify to lose track of the original inode.
let watch_dir = config_path
.parent()
.unwrap_or_else(|| std::path::Path::new("."))
.to_path_buf();
// ── inotify watcher (instant on local fs) ────────────────────────────
let config_file = config_path.clone();
let tx_clone = notify_tx.clone();
let watcher_result = recommended_watcher(move |res: notify::Result<notify::Event>| {
let tx_inotify = notify_tx.clone();
let inotify_ok = match recommended_watcher(move |res: notify::Result<notify::Event>| {
let Ok(event) = res else { return };
let is_our_file = event.paths.iter().any(|p| p == &config_file);
if !is_our_file {
return;
if !is_our_file { return; }
if matches!(event.kind, EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_)) {
let _ = tx_inotify.try_send(());
}
let relevant = matches!(
event.kind,
EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_)
);
if relevant {
let _ = tx_clone.try_send(());
}) {
Ok(mut w) => match w.watch(&watch_dir, RecursiveMode::NonRecursive) {
Ok(()) => {
info!("config watcher: inotify active on {:?}", config_path);
Box::leak(Box::new(w));
true
}
Err(e) => { warn!("config watcher: inotify watch failed: {}", e); false }
},
Err(e) => { warn!("config watcher: inotify unavailable: {}", e); false }
};
// ── poll watcher (always active, fixes Docker bind mounts / NFS) ─────
// inotify does not receive events for files mounted from the host into
// a container. PollWatcher compares file contents every 3 s and fires
// on any change regardless of the underlying fs.
let config_file2 = config_path.clone();
let tx_poll = notify_tx.clone();
match notify::poll::PollWatcher::new(
move |res: notify::Result<notify::Event>| {
let Ok(event) = res else { return };
let is_our_file = event.paths.iter().any(|p| p == &config_file2);
if !is_our_file { return; }
if matches!(event.kind, EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_)) {
let _ = tx_poll.try_send(());
}
},
notify::Config::default()
.with_poll_interval(std::time::Duration::from_secs(3))
.with_compare_contents(true),
) {
Ok(mut w) => match w.watch(&config_path, RecursiveMode::NonRecursive) {
Ok(()) => {
if inotify_ok {
info!("config watcher: poll watcher also active (Docker/NFS safe)");
} else {
info!("config watcher: poll watcher active on {:?} (3s interval)", config_path);
}
Box::leak(Box::new(w));
}
Err(e) => warn!("config watcher: poll watch failed: {}", e),
},
Err(e) => warn!("config watcher: poll watcher unavailable: {}", e),
}
// ── event loop ───────────────────────────────────────────────────────
tokio::spawn(async move {
#[cfg(unix)]
let mut sighup = {
use tokio::signal::unix::{SignalKind, signal};
signal(SignalKind::hangup()).expect("Failed to register SIGHUP handler")
};
loop {
#[cfg(unix)]
tokio::select! {
msg = notify_rx.recv() => {
if msg.is_none() { break; }
}
_ = sighup.recv() => {
info!("SIGHUP received — reloading {:?}", config_path);
}
}
#[cfg(not(unix))]
if notify_rx.recv().await.is_none() { break; }
// Debounce: drain extra events that arrive within 50 ms.
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
while notify_rx.try_recv().is_ok() {}
reload_config(&config_path, &config_tx, &log_tx, detected_ip_v4, detected_ip_v6);
}
});
match watcher_result {
Ok(mut watcher) => {
match watcher.watch(&watch_dir, RecursiveMode::NonRecursive) {
Ok(()) => info!("config watcher: watching {:?} via inotify", config_path),
Err(e) => warn!(
"config watcher: failed to watch {:?}: {}; use SIGHUP to reload",
watch_dir, e
),
}
tokio::spawn(async move {
let _watcher = watcher; // keep alive
#[cfg(unix)]
let mut sighup = {
use tokio::signal::unix::{SignalKind, signal};
signal(SignalKind::hangup()).expect("Failed to register SIGHUP handler")
};
loop {
#[cfg(unix)]
tokio::select! {
msg = notify_rx.recv() => {
if msg.is_none() { break; }
}
_ = sighup.recv() => {
info!("SIGHUP received — reloading {:?}", config_path);
}
}
#[cfg(not(unix))]
if notify_rx.recv().await.is_none() { break; }
// Debounce: drain extra events fired within 50ms.
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
while notify_rx.try_recv().is_ok() {}
reload_config(
&config_path,
&config_tx,
&log_tx,
detected_ip_v4,
detected_ip_v6,
);
}
});
}
Err(e) => {
warn!(
"config watcher: inotify unavailable ({}); only SIGHUP will trigger reload",
e
);
// Fall back to SIGHUP-only.
tokio::spawn(async move {
#[cfg(unix)]
{
use tokio::signal::unix::{SignalKind, signal};
let mut sighup = signal(SignalKind::hangup())
.expect("Failed to register SIGHUP handler");
loop {
sighup.recv().await;
info!("SIGHUP received — reloading {:?}", config_path);
reload_config(
&config_path,
&config_tx,
&log_tx,
detected_ip_v4,
detected_ip_v6,
);
}
}
#[cfg(not(unix))]
let _ = (config_tx, log_tx, config_path);
});
}
}
(config_rx, log_rx)
}

86
src/config/types.rs
View File

@ -96,6 +96,22 @@ pub struct NetworkConfig {
#[serde(default)]
pub multipath: bool,
/// STUN servers list for public IP discovery.
#[serde(default = "default_stun_servers")]
pub stun_servers: Vec<String>,
/// Enable TCP STUN fallback when UDP is blocked.
#[serde(default)]
pub stun_tcp_fallback: bool,
/// HTTP-based public IP detection endpoints (fallback after STUN).
#[serde(default = "default_http_ip_detect_urls")]
pub http_ip_detect_urls: Vec<String>,
/// Cache file path for detected public IP.
#[serde(default = "default_cache_public_ip_path")]
pub cache_public_ip_path: String,
}
impl Default for NetworkConfig {
@ -105,6 +121,10 @@ impl Default for NetworkConfig {
ipv6: None,
prefer: 4,
multipath: false,
stun_servers: default_stun_servers(),
stun_tcp_fallback: true,
http_ip_detect_urls: default_http_ip_detect_urls(),
cache_public_ip_path: default_cache_public_ip_path(),
}
}
}
@ -172,6 +192,15 @@ pub struct GeneralConfig {
#[serde(default = "default_true")]
pub me_keepalive_payload_random: bool,
/// Max pending ciphertext buffer per client writer (bytes).
/// Controls FakeTLS backpressure vs throughput.
#[serde(default = "default_crypto_pending_buffer")]
pub crypto_pending_buffer: usize,
/// Maximum allowed client MTProto frame size (bytes).
#[serde(default = "default_max_client_frame")]
pub max_client_frame: usize,
/// Enable staggered warmup of extra ME writers.
#[serde(default = "default_true")]
pub me_warmup_stagger_enabled: bool,
@ -218,6 +247,34 @@ pub struct GeneralConfig {
/// [general.links] — proxy link generation overrides.
#[serde(default)]
pub links: LinksConfig,
/// Minimum TLS record size when fast_mode coalescing is enabled (0 = disabled).
#[serde(default = "default_fast_mode_min_tls_record")]
pub fast_mode_min_tls_record: usize,
/// Automatically reload proxy-secret every N seconds.
#[serde(default = "default_proxy_secret_reload_secs")]
pub proxy_secret_auto_reload_secs: u64,
/// Automatically reload proxy-multi.conf every N seconds.
#[serde(default = "default_proxy_config_reload_secs")]
pub proxy_config_auto_reload_secs: u64,
/// Enable NTP drift check at startup.
#[serde(default = "default_ntp_check")]
pub ntp_check: bool,
/// NTP servers for drift check.
#[serde(default = "default_ntp_servers")]
pub ntp_servers: Vec<String>,
/// Enable auto-degradation from ME to Direct-DC.
#[serde(default = "default_true")]
pub auto_degradation_enabled: bool,
/// Minimum unavailable ME DC groups before degrading.
#[serde(default = "default_degradation_min_unavailable_dc_groups")]
pub degradation_min_unavailable_dc_groups: u8,
}
impl Default for GeneralConfig {
@ -251,6 +308,15 @@ impl Default for GeneralConfig {
log_level: LogLevel::Normal,
disable_colors: false,
links: LinksConfig::default(),
crypto_pending_buffer: default_crypto_pending_buffer(),
max_client_frame: default_max_client_frame(),
fast_mode_min_tls_record: default_fast_mode_min_tls_record(),
proxy_secret_auto_reload_secs: default_proxy_secret_reload_secs(),
proxy_config_auto_reload_secs: default_proxy_config_reload_secs(),
ntp_check: default_ntp_check(),
ntp_servers: default_ntp_servers(),
auto_degradation_enabled: true,
degradation_min_unavailable_dc_groups: default_degradation_min_unavailable_dc_groups(),
}
}
}
@ -395,6 +461,22 @@ pub struct AntiCensorshipConfig {
/// Directory to store TLS front cache (on disk).
#[serde(default = "default_tls_front_dir")]
pub tls_front_dir: String,
/// Minimum server_hello delay in milliseconds (anti-fingerprint).
#[serde(default = "default_server_hello_delay_min_ms")]
pub server_hello_delay_min_ms: u64,
/// Maximum server_hello delay in milliseconds.
#[serde(default = "default_server_hello_delay_max_ms")]
pub server_hello_delay_max_ms: u64,
/// Number of NewSessionTicket messages to emit post-handshake.
#[serde(default = "default_tls_new_session_tickets")]
pub tls_new_session_tickets: u8,
/// Enforce ALPN echo of client preference.
#[serde(default = "default_alpn_enforce")]
pub alpn_enforce: bool,
}
impl Default for AntiCensorshipConfig {
@ -409,6 +491,10 @@ impl Default for AntiCensorshipConfig {
fake_cert_len: default_fake_cert_len(),
tls_emulation: false,
tls_front_dir: default_tls_front_dir(),
server_hello_delay_min_ms: default_server_hello_delay_min_ms(),
server_hello_delay_max_ms: default_server_hello_delay_max_ms(),
tls_new_session_tickets: default_tls_new_session_tickets(),
alpn_enforce: default_alpn_enforce(),
}
}
}

3
src/main.rs
View File

@ -213,6 +213,9 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
"Modes: classic={} secure={} tls={}",
config.general.modes.classic, config.general.modes.secure, config.general.modes.tls
);
if config.general.modes.classic {
warn!("Classic mode is vulnerable to DPI detection; enable only for legacy clients");
}
info!("TLS domain: {}", config.censorship.tls_domain);
if let Some(ref sock) = config.censorship.mask_unix_sock {
info!("Mask: {} -> unix:{}", config.censorship.mask, sock);

15
src/protocol/constants.rs
View File

@ -1,6 +1,8 @@
//! Protocol constants and datacenter addresses
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use crate::crypto::SecureRandom;
use std::sync::LazyLock;
// ============= Telegram Datacenters =============
@ -151,7 +153,18 @@ pub const TLS_RECORD_ALERT: u8 = 0x15;
/// Maximum TLS record size
pub const MAX_TLS_RECORD_SIZE: usize = 16384;
/// Maximum TLS chunk size (with overhead)
pub const MAX_TLS_CHUNK_SIZE: usize = 16384 + 24;
/// RFC 8446 §5.2 allows up to 16384 + 256 bytes of ciphertext
pub const MAX_TLS_CHUNK_SIZE: usize = 16384 + 256;
/// Generate padding length for Secure Intermediate protocol.
/// Total (data + padding) must not be divisible by 4 per MTProto spec.
pub fn secure_padding_len(data_len: usize, rng: &SecureRandom) -> usize {
if data_len % 4 == 0 {
(rng.range(3) + 1) as usize // 1-3
} else {
rng.range(4) as usize // 0-3
}
}
// ============= Timeouts =============

188
src/protocol/tls.rs
View File

@ -32,6 +32,7 @@ pub const TIME_SKEW_MAX: i64 = 10 * 60; // 10 minutes after
mod extension_type {
pub const KEY_SHARE: u16 = 0x0033;
pub const SUPPORTED_VERSIONS: u16 = 0x002b;
pub const ALPN: u16 = 0x0010;
}
/// TLS Cipher Suites
@ -62,6 +63,7 @@ pub struct TlsValidation {
// ============= TLS Extension Builder =============
/// Builder for TLS extensions with correct length calculation
#[derive(Clone)]
struct TlsExtensionBuilder {
extensions: Vec<u8>,
}
@ -109,6 +111,27 @@ impl TlsExtensionBuilder {
self
}
/// Add ALPN extension with a single selected protocol.
fn add_alpn(&mut self, proto: &[u8]) -> &mut Self {
// Extension type: ALPN (0x0010)
self.extensions.extend_from_slice(&extension_type::ALPN.to_be_bytes());
// ALPN extension format:
// extension_data length (2 bytes)
// protocols length (2 bytes)
// protocol name length (1 byte)
// protocol name bytes
let proto_len = proto.len() as u8;
let list_len: u16 = 1 + proto_len as u16;
let ext_len: u16 = 2 + list_len;
self.extensions.extend_from_slice(&ext_len.to_be_bytes());
self.extensions.extend_from_slice(&list_len.to_be_bytes());
self.extensions.push(proto_len);
self.extensions.extend_from_slice(proto);
self
}
/// Build final extensions with length prefix
fn build(self) -> Vec<u8> {
let mut result = Vec::with_capacity(2 + self.extensions.len());
@ -144,6 +167,8 @@ struct ServerHelloBuilder {
compression: u8,
/// Extensions
extensions: TlsExtensionBuilder,
/// Selected ALPN protocol (if any)
alpn: Option<Vec<u8>>,
}
impl ServerHelloBuilder {
@ -154,6 +179,7 @@ impl ServerHelloBuilder {
cipher_suite: cipher_suite::TLS_AES_128_GCM_SHA256,
compression: 0x00,
extensions: TlsExtensionBuilder::new(),
alpn: None,
}
}
@ -168,9 +194,18 @@ impl ServerHelloBuilder {
self
}
fn with_alpn(mut self, proto: Option<Vec<u8>>) -> Self {
self.alpn = proto;
self
}
/// Build ServerHello message (without record header)
fn build_message(&self) -> Vec<u8> {
let extensions = self.extensions.extensions.clone();
let mut ext_builder = self.extensions.clone();
if let Some(ref alpn) = self.alpn {
ext_builder.add_alpn(alpn);
}
let extensions = ext_builder.extensions.clone();
let extensions_len = extensions.len() as u16;
// Calculate total length
@ -350,6 +385,8 @@ pub fn build_server_hello(
session_id: &[u8],
fake_cert_len: usize,
rng: &SecureRandom,
alpn: Option<Vec<u8>>,
new_session_tickets: u8,
) -> Vec<u8> {
const MIN_APP_DATA: usize = 64;
const MAX_APP_DATA: usize = 16640; // RFC 8446 §5.2 upper bound
@ -360,6 +397,7 @@ pub fn build_server_hello(
let server_hello = ServerHelloBuilder::new(session_id.to_vec())
.with_x25519_key(&x25519_key)
.with_tls13_version()
.with_alpn(alpn)
.build_record();
// Build Change Cipher Spec record
@ -376,21 +414,35 @@ pub fn build_server_hello(
app_data_record.push(TLS_RECORD_APPLICATION);
app_data_record.extend_from_slice(&TLS_VERSION);
app_data_record.extend_from_slice(&(fake_cert_len as u16).to_be_bytes());
if fake_cert_len > 17 {
app_data_record.extend_from_slice(&fake_cert[..fake_cert_len - 17]);
app_data_record.push(0x16); // inner content type marker
app_data_record.extend_from_slice(&rng.bytes(16)); // AEAD-like tag mimic
} else {
app_data_record.extend_from_slice(&fake_cert);
// Fill ApplicationData with fully random bytes of desired length to avoid
// deterministic DPI fingerprints (fixed inner content type markers).
app_data_record.extend_from_slice(&fake_cert);
// Build optional NewSessionTicket records (TLS 1.3 handshake messages are encrypted;
// here we mimic with opaque ApplicationData records of plausible size).
let mut tickets = Vec::new();
if new_session_tickets > 0 {
for _ in 0..new_session_tickets {
let ticket_len: usize = rng.range(48) + 48; // 48-95 bytes
let mut record = Vec::with_capacity(5 + ticket_len);
record.push(TLS_RECORD_APPLICATION);
record.extend_from_slice(&TLS_VERSION);
record.extend_from_slice(&(ticket_len as u16).to_be_bytes());
record.extend_from_slice(&rng.bytes(ticket_len));
tickets.push(record);
}
}
// Combine all records
let mut response = Vec::with_capacity(
server_hello.len() + change_cipher_spec.len() + app_data_record.len()
server_hello.len() + change_cipher_spec.len() + app_data_record.len() + tickets.iter().map(|r| r.len()).sum::<usize>()
);
response.extend_from_slice(&server_hello);
response.extend_from_slice(&change_cipher_spec);
response.extend_from_slice(&app_data_record);
for t in &tickets {
response.extend_from_slice(t);
}
// Compute HMAC for the response
let mut hmac_input = Vec::with_capacity(TLS_DIGEST_LEN + response.len());
@ -484,85 +536,53 @@ pub fn extract_sni_from_client_hello(handshake: &[u8]) -> Option<String> {
None
}
/// Extract ALPN protocol list from TLS ClientHello.
pub fn extract_alpn_from_client_hello(handshake: &[u8]) -> Option<Vec<String>> {
if handshake.len() < 43 || handshake[0] != TLS_RECORD_HANDSHAKE {
return None;
}
/// Extract ALPN protocol list from ClientHello, return in offered order.
pub fn extract_alpn_from_client_hello(handshake: &[u8]) -> Vec<Vec<u8>> {
let mut pos = 5; // after record header
if handshake.get(pos).copied()? != 0x01 {
return None; // not ClientHello
if handshake.get(pos) != Some(&0x01) {
return Vec::new();
}
// Handshake length bytes
pos += 4; // type + len (3)
// version (2) + random (32)
pos += 2 + 32;
if pos + 1 > handshake.len() {
return None;
}
let session_id_len = *handshake.get(pos)? as usize;
pos += 4; // type + len
pos += 2 + 32; // version + random
if pos >= handshake.len() { return Vec::new(); }
let session_id_len = *handshake.get(pos).unwrap_or(&0) as usize;
pos += 1 + session_id_len;
if pos + 2 > handshake.len() {
return None;
}
let cipher_suites_len = u16::from_be_bytes([handshake[pos], handshake[pos + 1]]) as usize;
pos += 2 + cipher_suites_len;
if pos + 1 > handshake.len() {
return None;
}
let comp_len = *handshake.get(pos)? as usize;
if pos + 2 > handshake.len() { return Vec::new(); }
let cipher_len = u16::from_be_bytes([handshake[pos], handshake[pos+1]]) as usize;
pos += 2 + cipher_len;
if pos >= handshake.len() { return Vec::new(); }
let comp_len = *handshake.get(pos).unwrap_or(&0) as usize;
pos += 1 + comp_len;
if pos + 2 > handshake.len() {
return None;
}
let ext_len = u16::from_be_bytes([handshake[pos], handshake[pos + 1]]) as usize;
if pos + 2 > handshake.len() { return Vec::new(); }
let ext_len = u16::from_be_bytes([handshake[pos], handshake[pos+1]]) as usize;
pos += 2;
let ext_end = pos + ext_len;
if ext_end > handshake.len() {
return None;
}
if ext_end > handshake.len() { return Vec::new(); }
let mut out = Vec::new();
while pos + 4 <= ext_end {
let etype = u16::from_be_bytes([handshake[pos], handshake[pos + 1]]);
let elen = u16::from_be_bytes([handshake[pos + 2], handshake[pos + 3]]) as usize;
let etype = u16::from_be_bytes([handshake[pos], handshake[pos+1]]);
let elen = u16::from_be_bytes([handshake[pos+2], handshake[pos+3]]) as usize;
pos += 4;
if pos + elen > ext_end {
if pos + elen > ext_end { break; }
if etype == extension_type::ALPN && elen >= 3 {
let list_len = u16::from_be_bytes([handshake[pos], handshake[pos+1]]) as usize;
let mut lp = pos + 2;
let list_end = (pos + 2).saturating_add(list_len).min(pos + elen);
while lp + 1 <= list_end {
let plen = handshake[lp] as usize;
lp += 1;
if lp + plen > list_end { break; }
out.push(handshake[lp..lp+plen].to_vec());
lp += plen;
}
break;
}
if etype == 0x0010 && elen >= 3 {
// ALPN
let list_len = u16::from_be_bytes([handshake[pos], handshake[pos + 1]]) as usize;
let mut alpn_pos = pos + 2;
let list_end = std::cmp::min(alpn_pos + list_len, pos + elen);
let mut protocols = Vec::new();
while alpn_pos < list_end {
let proto_len = *handshake.get(alpn_pos)? as usize;
alpn_pos += 1;
if alpn_pos + proto_len > list_end {
break;
}
if let Ok(p) = std::str::from_utf8(&handshake[alpn_pos..alpn_pos + proto_len]) {
protocols.push(p.to_string());
}
alpn_pos += proto_len;
}
return Some(protocols);
}
pos += elen;
}
None
out
}
/// Check if bytes look like a TLS ClientHello
pub fn is_tls_handshake(first_bytes: &[u8]) -> bool {
if first_bytes.len() < 3 {
@ -741,7 +761,7 @@ mod tests {
let session_id = vec![0xAA; 32];
let rng = SecureRandom::new();
let response = build_server_hello(secret, &client_digest, &session_id, 2048, &rng);
let response = build_server_hello(secret, &client_digest, &session_id, 2048, &rng, None, 0);
// Should have at least 3 records
assert!(response.len() > 100);
@ -774,8 +794,8 @@ mod tests {
let session_id = vec![0xAA; 32];
let rng = SecureRandom::new();
let response1 = build_server_hello(secret, &client_digest, &session_id, 1024, &rng);
let response2 = build_server_hello(secret, &client_digest, &session_id, 1024, &rng);
let response1 = build_server_hello(secret, &client_digest, &session_id, 1024, &rng, None, 0);
let response2 = build_server_hello(secret, &client_digest, &session_id, 1024, &rng, None, 0);
// Digest position should have non-zero data
let digest1 = &response1[TLS_DIGEST_POS..TLS_DIGEST_POS + TLS_DIGEST_LEN];
@ -904,8 +924,12 @@ mod tests {
alpn_data.push(2);
alpn_data.extend_from_slice(b"h2");
let ch = build_client_hello_with_exts(vec![(0x0010, alpn_data)], "alpn.test");
let alpn = extract_alpn_from_client_hello(&ch).unwrap();
assert_eq!(alpn, vec!["h2"]);
let alpn = extract_alpn_from_client_hello(&ch);
let alpn_str: Vec<String> = alpn
.iter()
.map(|p| std::str::from_utf8(p).unwrap().to_string())
.collect();
assert_eq!(alpn_str, vec!["h2"]);
}
#[test]
@ -920,7 +944,11 @@ mod tests {
alpn_data.push(2);
alpn_data.extend_from_slice(b"h3");
let ch = build_client_hello_with_exts(vec![(0x0010, alpn_data)], "alpn.test");
let alpn = extract_alpn_from_client_hello(&ch).unwrap();
assert_eq!(alpn, vec!["h2", "spdy", "h3"]);
let alpn = extract_alpn_from_client_hello(&ch);
let alpn_str: Vec<String> = alpn
.iter()
.map(|p| std::str::from_utf8(p).unwrap().to_string())
.collect();
assert_eq!(alpn_str, vec!["h2", "spdy", "h3"]);
}
}

3
src/proxy/direct_relay.rs
View File

@ -178,8 +178,9 @@ async fn do_tg_handshake_static(
let (read_half, write_half) = stream.into_split();
let max_pending = config.general.crypto_pending_buffer;
Ok((
CryptoReader::new(read_half, tg_decryptor),
CryptoWriter::new(write_half, tg_encryptor),
CryptoWriter::new(write_half, tg_encryptor, max_pending),
))
}

39
src/proxy/handshake.rs
View File

@ -7,6 +7,7 @@ use tracing::{debug, warn, trace, info};
use zeroize::Zeroize;
use crate::crypto::{sha256, AesCtr, SecureRandom};
use rand::Rng;
use crate::protocol::constants::*;
use crate::protocol::tls;
use crate::stream::{FakeTlsReader, FakeTlsWriter, CryptoReader, CryptoWriter};
@ -119,6 +120,23 @@ where
None
};
let alpn_list = if config.censorship.alpn_enforce {
tls::extract_alpn_from_client_hello(handshake)
} else {
Vec::new()
};
let selected_alpn = if config.censorship.alpn_enforce {
if alpn_list.iter().any(|p| p == b"h2") {
Some(b"h2".to_vec())
} else if alpn_list.iter().any(|p| p == b"http/1.1") {
Some(b"http/1.1".to_vec())
} else {
None
}
} else {
None
};
let response = if let Some(cached_entry) = cached {
emulator::build_emulated_server_hello(
secret,
@ -126,6 +144,8 @@ where
&validation.session_id,
&cached_entry,
rng,
selected_alpn.clone(),
config.censorship.tls_new_session_tickets,
)
} else {
tls::build_server_hello(
@ -134,9 +154,25 @@ where
&validation.session_id,
config.censorship.fake_cert_len,
rng,
selected_alpn.clone(),
config.censorship.tls_new_session_tickets,
)
};
// Optional anti-fingerprint delay before sending ServerHello.
if config.censorship.server_hello_delay_max_ms > 0 {
let min = config.censorship.server_hello_delay_min_ms;
let max = config.censorship.server_hello_delay_max_ms.max(min);
let delay_ms = if max == min {
max
} else {
rand::rng().random_range(min..=max)
};
if delay_ms > 0 {
tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
}
}
debug!(peer = %peer, response_len = response.len(), "Sending TLS ServerHello");
if let Err(e) = writer.write_all(&response).await {
@ -264,9 +300,10 @@ where
"MTProto handshake successful"
);
let max_pending = config.general.crypto_pending_buffer;
return HandshakeResult::Success((
CryptoReader::new(reader, decryptor),
CryptoWriter::new(writer, encryptor),
CryptoWriter::new(writer, encryptor, max_pending),
success,
));
}

149
src/proxy/middle_relay.rs
View File

@ -2,12 +2,13 @@ use std::net::SocketAddr;
use std::sync::Arc;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tracing::{debug, info, trace};
use tokio::sync::oneshot;
use tracing::{debug, info, trace, warn};
use crate::config::ProxyConfig;
use crate::crypto::SecureRandom;
use crate::error::{ProxyError, Result};
use crate::protocol::constants::*;
use crate::protocol::constants::{*, secure_padding_len};
use crate::proxy::handshake::HandshakeSuccess;
use crate::stats::Stats;
use crate::stream::{BufferPool, CryptoReader, CryptoWriter};
@ -15,11 +16,11 @@ use crate::transport::middle_proxy::{MePool, MeResponse, proto_flags_for_tag};
pub(crate) async fn handle_via_middle_proxy<R, W>(
mut crypto_reader: CryptoReader<R>,
mut crypto_writer: CryptoWriter<W>,
crypto_writer: CryptoWriter<W>,
success: HandshakeSuccess,
me_pool: Arc<MePool>,
stats: Arc<Stats>,
_config: Arc<ProxyConfig>,
config: Arc<ProxyConfig>,
_buffer_pool: Arc<BufferPool>,
local_addr: SocketAddr,
rng: Arc<SecureRandom>,
@ -41,7 +42,7 @@ where
"Routing via Middle-End"
);
let (conn_id, mut me_rx) = me_pool.registry().register().await;
let (conn_id, me_rx) = me_pool.registry().register().await;
stats.increment_user_connects(&user);
stats.increment_user_curr_connects(&user);
@ -56,59 +57,90 @@ where
let translated_local_addr = me_pool.translate_our_addr(local_addr);
let result: Result<()> = loop {
tokio::select! {
client_frame = read_client_payload(&mut crypto_reader, proto_tag) => {
match client_frame {
Ok(Some((payload, quickack))) => {
trace!(conn_id, bytes = payload.len(), "C->ME frame");
stats.add_user_octets_from(&user, payload.len() as u64);
let mut flags = proto_flags;
if quickack {
flags |= RPC_FLAG_QUICKACK;
let frame_limit = config.general.max_client_frame;
let (stop_tx, mut stop_rx) = oneshot::channel::<()>();
let mut me_rx_task = me_rx;
let stats_clone = stats.clone();
let rng_clone = rng.clone();
let user_clone = user.clone();
let me_writer = tokio::spawn(async move {
let mut writer = crypto_writer;
loop {
tokio::select! {
msg = me_rx_task.recv() => {
match msg {
Some(MeResponse::Data { flags, data }) => {
trace!(conn_id, bytes = data.len(), flags, "ME->C data");
stats_clone.add_user_octets_to(&user_clone, data.len() as u64);
write_client_payload(&mut writer, proto_tag, flags, &data, rng_clone.as_ref()).await?;
}
if payload.len() >= 8 && payload[..8].iter().all(|b| *b == 0) {
flags |= RPC_FLAG_NOT_ENCRYPTED;
Some(MeResponse::Ack(confirm)) => {
trace!(conn_id, confirm, "ME->C quickack");
write_client_ack(&mut writer, proto_tag, confirm).await?;
}
Some(MeResponse::Close) => {
debug!(conn_id, "ME sent close");
return Ok(());
}
None => {
debug!(conn_id, "ME channel closed");
return Err(ProxyError::Proxy("ME connection lost".into()));
}
me_pool.send_proxy_req(
conn_id,
success.dc_idx,
peer,
translated_local_addr,
&payload,
flags,
).await?;
}
Ok(None) => {
debug!(conn_id, "Client EOF");
let _ = me_pool.send_close(conn_id).await;
break Ok(());
}
Err(e) => break Err(e),
}
}
me_msg = me_rx.recv() => {
match me_msg {
Some(MeResponse::Data { flags, data }) => {
trace!(conn_id, bytes = data.len(), flags, "ME->C data");
stats.add_user_octets_to(&user, data.len() as u64);
write_client_payload(&mut crypto_writer, proto_tag, flags, &data, rng.as_ref()).await?;
}
Some(MeResponse::Ack(confirm)) => {
trace!(conn_id, confirm, "ME->C quickack");
write_client_ack(&mut crypto_writer, proto_tag, confirm).await?;
}
Some(MeResponse::Close) => {
debug!(conn_id, "ME sent close");
break Ok(());
}
None => {
debug!(conn_id, "ME channel closed");
break Err(ProxyError::Proxy("ME connection lost".into()));
}
_ = &mut stop_rx => {
debug!(conn_id, "ME writer stop signal");
return Ok(());
}
}
}
});
let mut main_result: Result<()> = Ok(());
loop {
match read_client_payload(&mut crypto_reader, proto_tag, frame_limit, &user).await {
Ok(Some((payload, quickack))) => {
trace!(conn_id, bytes = payload.len(), "C->ME frame");
stats.add_user_octets_from(&user, payload.len() as u64);
let mut flags = proto_flags;
if quickack {
flags |= RPC_FLAG_QUICKACK;
}
if payload.len() >= 8 && payload[..8].iter().all(|b| *b == 0) {
flags |= RPC_FLAG_NOT_ENCRYPTED;
}
if let Err(e) = me_pool.send_proxy_req(
conn_id,
success.dc_idx,
peer,
translated_local_addr,
&payload,
flags,
).await {
main_result = Err(e);
break;
}
}
Ok(None) => {
debug!(conn_id, "Client EOF");
let _ = me_pool.send_close(conn_id).await;
break;
}
Err(e) => {
main_result = Err(e);
break;
}
}
}
let _ = stop_tx.send(());
let writer_result = me_writer.await.unwrap_or_else(|e| Err(ProxyError::Proxy(format!("ME writer join error: {e}"))));
let result = match (main_result, writer_result) {
(Ok(()), Ok(())) => Ok(()),
(Err(e), _) => Err(e),
(_, Err(e)) => Err(e),
};
debug!(user = %user, conn_id, "ME relay cleanup");
@ -120,6 +152,8 @@ where
async fn read_client_payload<R>(
client_reader: &mut CryptoReader<R>,
proto_tag: ProtoTag,
max_frame: usize,
user: &str,
) -> Result<Option<(Vec<u8>, bool)>>
where
R: AsyncRead + Unpin + Send + 'static,
@ -162,8 +196,15 @@ where
}
};
if len > 16 * 1024 * 1024 {
return Err(ProxyError::Proxy(format!("Frame too large: {len}")));
if len > max_frame {
warn!(
user = %user,
raw_len = len,
raw_len_hex = format_args!("0x{:08x}", len),
proto = ?proto_tag,
"Frame too large — possible crypto desync or TLS record error"
);
return Err(ProxyError::Proxy(format!("Frame too large: {len} (max {max_frame})")));
}
let mut payload = vec![0u8; len];
@ -237,7 +278,7 @@ where
}
ProtoTag::Intermediate | ProtoTag::Secure => {
let padding_len = if proto_tag == ProtoTag::Secure {
(rng.bytes(1)[0] % 4) as usize
secure_padding_len(data.len(), rng)
} else {
0
};

36
src/stream/crypto_stream.rs
View File

@ -34,7 +34,7 @@
//! └────────────────────────────────────────┘
//!
//! Backpressure
//! - pending ciphertext buffer is bounded (MAX_PENDING_WRITE)
//! - pending ciphertext buffer is bounded (configurable per connection)
//! - pending is full and upstream is pending
//! -> poll_write returns Poll::Pending
//! -> do not accept any plaintext
@ -62,10 +62,9 @@ use super::state::{StreamState, YieldBuffer};
// ============= Constants =============
/// Maximum size for pending ciphertext buffer (bounded backpressure).
/// Reduced to 64KB to prevent bufferbloat on mobile networks.
/// 512KB was causing high latency on 3G/LTE connections.
const MAX_PENDING_WRITE: usize = 64 * 1024;
/// Default size for pending ciphertext buffer (bounded backpressure).
/// Actual limit is supplied at runtime from configuration.
const DEFAULT_MAX_PENDING_WRITE: usize = 64 * 1024;
/// Default read buffer capacity (reader mostly decrypts in-place into caller buffer).
const DEFAULT_READ_CAPACITY: usize = 16 * 1024;
@ -427,15 +426,22 @@ pub struct CryptoWriter<W> {
encryptor: AesCtr,
state: CryptoWriterState,
scratch: BytesMut,
max_pending_write: usize,
}
impl<W> CryptoWriter<W> {
pub fn new(upstream: W, encryptor: AesCtr) -> Self {
pub fn new(upstream: W, encryptor: AesCtr, max_pending_write: usize) -> Self {
let max_pending = if max_pending_write == 0 {
DEFAULT_MAX_PENDING_WRITE
} else {
max_pending_write
};
Self {
upstream,
encryptor,
state: CryptoWriterState::Idle,
scratch: BytesMut::with_capacity(16 * 1024),
max_pending_write: max_pending.max(4 * 1024),
}
}
@ -484,10 +490,10 @@ impl<W> CryptoWriter<W> {
}
/// Ensure we are in Flushing state and return mutable pending buffer.
fn ensure_pending<'a>(state: &'a mut CryptoWriterState) -> &'a mut PendingCiphertext {
fn ensure_pending<'a>(state: &'a mut CryptoWriterState, max_pending: usize) -> &'a mut PendingCiphertext {
if matches!(state, CryptoWriterState::Idle) {
*state = CryptoWriterState::Flushing {
pending: PendingCiphertext::new(MAX_PENDING_WRITE),
pending: PendingCiphertext::new(max_pending),
};
}
@ -498,14 +504,14 @@ impl<W> CryptoWriter<W> {
}
/// Select how many plaintext bytes can be accepted in buffering path
fn select_to_accept_for_buffering(state: &CryptoWriterState, buf_len: usize) -> usize {
fn select_to_accept_for_buffering(state: &CryptoWriterState, buf_len: usize, max_pending: usize) -> usize {
if buf_len == 0 {
return 0;
}
match state {
CryptoWriterState::Flushing { pending } => buf_len.min(pending.remaining_capacity()),
CryptoWriterState::Idle => buf_len.min(MAX_PENDING_WRITE),
CryptoWriterState::Idle => buf_len.min(max_pending),
CryptoWriterState::Poisoned { .. } => 0,
}
}
@ -603,7 +609,7 @@ impl<W: AsyncWrite + Unpin> AsyncWrite for CryptoWriter<W> {
Poll::Pending => {
// Upstream blocked. Apply ideal backpressure
let to_accept =
Self::select_to_accept_for_buffering(&this.state, buf.len());
Self::select_to_accept_for_buffering(&this.state, buf.len(), this.max_pending_write);
if to_accept == 0 {
trace!(
@ -618,7 +624,7 @@ impl<W: AsyncWrite + Unpin> AsyncWrite for CryptoWriter<W> {
// Disjoint borrows
let encryptor = &mut this.encryptor;
let pending = Self::ensure_pending(&mut this.state);
let pending = Self::ensure_pending(&mut this.state, this.max_pending_write);
if let Err(e) = pending.push_encrypted(encryptor, plaintext) {
if e.kind() == ErrorKind::WouldBlock {
@ -635,7 +641,7 @@ impl<W: AsyncWrite + Unpin> AsyncWrite for CryptoWriter<W> {
// 2) Fast path: pending empty -> write-through
debug_assert!(matches!(this.state, CryptoWriterState::Idle));
let to_accept = buf.len().min(MAX_PENDING_WRITE);
let to_accept = buf.len().min(this.max_pending_write);
let plaintext = &buf[..to_accept];
Self::encrypt_into_scratch(&mut this.encryptor, &mut this.scratch, plaintext);
@ -645,7 +651,7 @@ impl<W: AsyncWrite + Unpin> AsyncWrite for CryptoWriter<W> {
// Upstream blocked: buffer FULL ciphertext for accepted bytes.
let ciphertext = std::mem::take(&mut this.scratch);
let pending = Self::ensure_pending(&mut this.state);
let pending = Self::ensure_pending(&mut this.state, this.max_pending_write);
pending.replace_with(ciphertext);
Poll::Ready(Ok(to_accept))
@ -672,7 +678,7 @@ impl<W: AsyncWrite + Unpin> AsyncWrite for CryptoWriter<W> {
let remainder = this.scratch.split_off(n);
this.scratch.clear();
let pending = Self::ensure_pending(&mut this.state);
let pending = Self::ensure_pending(&mut this.state, this.max_pending_write);
pending.replace_with(remainder);
Poll::Ready(Ok(to_accept))

4
src/stream/frame_stream.rs
View File

@ -267,8 +267,8 @@ impl<W: AsyncWrite + Unpin> SecureIntermediateFrameWriter<W> {
return Ok(());
}
// Add random padding (0-3 bytes)
let padding_len = self.rng.range(4);
// Add padding so total length is never divisible by 4 (MTProto Secure)
let padding_len = secure_padding_len(data.len(), &self.rng);
let padding = self.rng.bytes(padding_len);
let total_len = data.len() + padding_len;

15
src/stream/tls_stream.rs
View File

@ -25,7 +25,8 @@
//! - However, the on-the-wire record length can exceed 16384 because TLS 1.3
//! uses AEAD and can include tag/overhead/padding.
//! - Telegram FakeTLS clients (notably iOS) may send Application Data records
//! with length up to 16384 + 24 bytes. We accept that as MAX_TLS_CHUNK_SIZE.
//! with length up to 16384 + 256 bytes (RFC 8446 §5.2). We accept that as
//! MAX_TLS_CHUNK_SIZE.
//!
//! If you reject those (e.g. validate length <= 16384), you will see errors like:
//! "TLS record too large: 16408 bytes"
@ -52,9 +53,8 @@ use super::state::{StreamState, HeaderBuffer, YieldBuffer, WriteBuffer};
const TLS_HEADER_SIZE: usize = 5;
/// Maximum TLS fragment size we emit for Application Data.
/// Real TLS 1.3 ciphertexts often add ~16-24 bytes AEAD overhead, so to mimic
/// on-the-wire record sizes we allow up to 16384 + 24 bytes of plaintext.
const MAX_TLS_PAYLOAD: usize = 16384 + 24;
/// Real TLS 1.3 allows up to 16384 + 256 bytes of ciphertext (incl. tag).
const MAX_TLS_PAYLOAD: usize = 16384 + 256;
/// Maximum pending write buffer for one record remainder.
/// Note: we never queue unlimited amount of data here; state holds at most one record.
@ -91,7 +91,7 @@ impl TlsRecordHeader {
/// - We accept TLS 1.0 header version for ClientHello-like records (0x03 0x01),
/// and TLS 1.2/1.3 style version bytes for the rest (we use TLS_VERSION = 0x03 0x03).
/// - For Application Data, Telegram FakeTLS may send payload length up to
/// MAX_TLS_CHUNK_SIZE (16384 + 24).
/// MAX_TLS_CHUNK_SIZE (16384 + 256).
/// - For other record types we keep stricter bounds to avoid memory abuse.
fn validate(&self) -> Result<()> {
// Version: accept TLS 1.0 header (ClientHello quirk) and TLS_VERSION (0x0303).
@ -105,7 +105,7 @@ impl TlsRecordHeader {
let len = self.length as usize;
// Length checks depend on record type.
// Telegram FakeTLS: ApplicationData length may be 16384 + 24.
// Telegram FakeTLS: ApplicationData length may be 16384 + 256.
match self.record_type {
TLS_RECORD_APPLICATION => {
if len > MAX_TLS_CHUNK_SIZE {
@ -755,9 +755,6 @@ impl<W: AsyncWrite + Unpin> AsyncWrite for FakeTlsWriter<W> {
payload_size: chunk_size,
};
// Wake to retry flushing soon.
cx.waker().wake_by_ref();
Poll::Ready(Ok(chunk_size))
}
}

22
src/tls_front/cache.rs
View File

@ -72,7 +72,27 @@ impl TlsFrontCache {
continue;
}
if let Ok(data) = tokio::fs::read(entry.path()).await {
if let Ok(cached) = serde_json::from_slice::<CachedTlsData>(&data) {
if let Ok(mut cached) = serde_json::from_slice::<CachedTlsData>(&data) {
if cached.domain.is_empty()
|| 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;
}
// 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;

28
src/tls_front/emulator.rs
View File

@ -34,6 +34,8 @@ pub fn build_emulated_server_hello(
session_id: &[u8],
cached: &CachedTlsData,
rng: &SecureRandom,
alpn: Option<Vec<u8>>,
new_session_tickets: u8,
) -> Vec<u8> {
// --- ServerHello ---
let mut extensions = Vec::new();
@ -48,6 +50,15 @@ pub fn build_emulated_server_hello(
extensions.extend_from_slice(&0x002bu16.to_be_bytes());
extensions.extend_from_slice(&(2u16).to_be_bytes());
extensions.extend_from_slice(&0x0304u16.to_be_bytes());
if let Some(alpn_proto) = &alpn {
extensions.extend_from_slice(&0x0010u16.to_be_bytes());
let list_len: u16 = 1 + alpn_proto.len() as u16;
let ext_len: u16 = 2 + list_len;
extensions.extend_from_slice(&ext_len.to_be_bytes());
extensions.extend_from_slice(&list_len.to_be_bytes());
extensions.push(alpn_proto.len() as u8);
extensions.extend_from_slice(alpn_proto);
}
let extensions_len = extensions.len() as u16;
@ -118,10 +129,25 @@ pub fn build_emulated_server_hello(
}
// --- Combine ---
let mut response = Vec::with_capacity(server_hello.len() + change_cipher_spec.len() + app_data.len());
// Optional NewSessionTicket mimic records (opaque ApplicationData for fingerprint).
let mut tickets = Vec::new();
if new_session_tickets > 0 {
for _ in 0..new_session_tickets {
let ticket_len: usize = rng.range(48) + 48;
let mut rec = Vec::with_capacity(5 + ticket_len);
rec.push(TLS_RECORD_APPLICATION);
rec.extend_from_slice(&TLS_VERSION);
rec.extend_from_slice(&(ticket_len as u16).to_be_bytes());
rec.extend_from_slice(&rng.bytes(ticket_len));
tickets.extend_from_slice(&rec);
}
}
let mut response = Vec::with_capacity(server_hello.len() + change_cipher_spec.len() + app_data.len() + tickets.len());
response.extend_from_slice(&server_hello);
response.extend_from_slice(&change_cipher_spec);
response.extend_from_slice(&app_data);
response.extend_from_slice(&tickets);
// --- HMAC ---
let mut hmac_input = Vec::with_capacity(TLS_DIGEST_LEN + response.len());

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

@ -567,12 +567,14 @@ impl MePool {
let cancel_keepalive = cancel_keepalive_token;
tokio::spawn(async move {
// Per-writer jittered start to avoid phase sync.
let initial_jitter_ms = rand::rng().random_range(0..=keepalive_jitter.as_millis().max(1) as u64);
let jitter_cap_ms = keepalive_interval.as_millis() / 2;
let effective_jitter_ms = keepalive_jitter.as_millis().min(jitter_cap_ms).max(1);
let initial_jitter_ms = rand::rng().random_range(0..=effective_jitter_ms as u64);
tokio::time::sleep(Duration::from_millis(initial_jitter_ms)).await;
loop {
tokio::select! {
_ = cancel_keepalive.cancelled() => break,
_ = tokio::time::sleep(keepalive_interval + Duration::from_millis(rand::rng().random_range(0..=keepalive_jitter.as_millis() as u64))) => {}
_ = tokio::time::sleep(keepalive_interval + Duration::from_millis(rand::rng().random_range(0..=effective_jitter_ms as u64))) => {}
}
if tx_keepalive.send(WriterCommand::Keepalive).await.is_err() {
break;

107
src/transport/upstream.rs
View File

@ -383,32 +383,43 @@ impl UpstreamManager {
Ok(stream)
},
UpstreamType::Socks4 { address, interface, user_id } => {
let proxy_addr: SocketAddr = address.parse()
.map_err(|_| ProxyError::Config("Invalid SOCKS4 address".to_string()))?;
// Try to parse as SocketAddr first (IP:port), otherwise treat as hostname:port
let mut stream = if let Ok(proxy_addr) = address.parse::<SocketAddr>() {
// IP:port format - use socket with optional interface binding
let bind_ip = Self::resolve_bind_address(
interface,
&None,
proxy_addr,
bind_rr.as_deref(),
);
let bind_ip = Self::resolve_bind_address(
interface,
&None,
proxy_addr,
bind_rr.as_deref(),
);
let socket = create_outgoing_socket_bound(proxy_addr, bind_ip)?;
let socket = create_outgoing_socket_bound(proxy_addr, bind_ip)?;
socket.set_nonblocking(true)?;
match socket.connect(&proxy_addr.into()) {
Ok(()) => {},
Err(err) if err.raw_os_error() == Some(libc::EINPROGRESS) || err.kind() == std::io::ErrorKind::WouldBlock => {},
Err(err) => return Err(ProxyError::Io(err)),
}
socket.set_nonblocking(true)?;
match socket.connect(&proxy_addr.into()) {
Ok(()) => {},
Err(err) if err.raw_os_error() == Some(libc::EINPROGRESS) || err.kind() == std::io::ErrorKind::WouldBlock => {},
Err(err) => return Err(ProxyError::Io(err)),
}
let std_stream: std::net::TcpStream = socket.into();
let stream = TcpStream::from_std(std_stream)?;
let std_stream: std::net::TcpStream = socket.into();
let mut stream = TcpStream::from_std(std_stream)?;
stream.writable().await?;
if let Some(e) = stream.take_error()? {
return Err(ProxyError::Io(e));
}
stream
} else {
// Hostname:port format - use tokio DNS resolution
// Note: interface binding is not supported for hostnames
if interface.is_some() {
warn!("SOCKS4 interface binding is not supported for hostname addresses, ignoring");
}
TcpStream::connect(address).await
.map_err(ProxyError::Io)?
};
stream.writable().await?;
if let Some(e) = stream.take_error()? {
return Err(ProxyError::Io(e));
}
// replace socks user_id with config.selected_scope, if set
let scope: Option<&str> = Some(config.selected_scope.as_str())
.filter(|s| !s.is_empty());
@ -418,32 +429,42 @@ impl UpstreamManager {
Ok(stream)
},
UpstreamType::Socks5 { address, interface, username, password } => {
let proxy_addr: SocketAddr = address.parse()
.map_err(|_| ProxyError::Config("Invalid SOCKS5 address".to_string()))?;
// Try to parse as SocketAddr first (IP:port), otherwise treat as hostname:port
let mut stream = if let Ok(proxy_addr) = address.parse::<SocketAddr>() {
// IP:port format - use socket with optional interface binding
let bind_ip = Self::resolve_bind_address(
interface,
&None,
proxy_addr,
bind_rr.as_deref(),
);
let bind_ip = Self::resolve_bind_address(
interface,
&None,
proxy_addr,
bind_rr.as_deref(),
);
let socket = create_outgoing_socket_bound(proxy_addr, bind_ip)?;
let socket = create_outgoing_socket_bound(proxy_addr, bind_ip)?;
socket.set_nonblocking(true)?;
match socket.connect(&proxy_addr.into()) {
Ok(()) => {},
Err(err) if err.raw_os_error() == Some(libc::EINPROGRESS) || err.kind() == std::io::ErrorKind::WouldBlock => {},
Err(err) => return Err(ProxyError::Io(err)),
}
socket.set_nonblocking(true)?;
match socket.connect(&proxy_addr.into()) {
Ok(()) => {},
Err(err) if err.raw_os_error() == Some(libc::EINPROGRESS) || err.kind() == std::io::ErrorKind::WouldBlock => {},
Err(err) => return Err(ProxyError::Io(err)),
}
let std_stream: std::net::TcpStream = socket.into();
let stream = TcpStream::from_std(std_stream)?;
let std_stream: std::net::TcpStream = socket.into();
let mut stream = TcpStream::from_std(std_stream)?;
stream.writable().await?;
if let Some(e) = stream.take_error()? {
return Err(ProxyError::Io(e));
}
stream.writable().await?;
if let Some(e) = stream.take_error()? {
return Err(ProxyError::Io(e));
}
stream
} else {
// Hostname:port format - use tokio DNS resolution
// Note: interface binding is not supported for hostnames
if interface.is_some() {
warn!("SOCKS5 interface binding is not supported for hostname addresses, ignoring");
}
TcpStream::connect(address).await
.map_err(ProxyError::Io)?
};
debug!(config = ?config, "Socks5 connection");
// replace socks user:pass with config.selected_scope, if set