naughtysoul/telemt
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Revert "Unix socket listener + reverse proxy improvements"

Browse Source
This commit is contained in:
Alexey
2026-02-15 10:09:03 +03:00
committed by GitHub
parent 0599a6ec8c
commit 0663e71c52
9 changed files with 67 additions and 486 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.lock generated
View File

@@ -1723,7 +1723,7 @@ dependencies = [
[[package]] [[package]]
name = "telemt" name = "telemt"
version = "2.0.0" version = "1.2.0"
dependencies = [ dependencies = [
"aes", "aes",
"base64", "base64",

12
README.md
View File

@@ -164,6 +164,10 @@ then Ctrl+X -> Y -> Enter to save
## Configuration ## Configuration
### Minimal Configuration for First Start ### Minimal Configuration for First Start
```toml ```toml
# === UI ===
# Users to show in the startup log (tg:// links)
show_link = ["hello"]
# === General Settings === # === General Settings ===
[general] [general]
prefer_ipv6 = false prefer_ipv6 = false
@@ -181,17 +185,9 @@ tls = true
port = 443 port = 443
listen_addr_ipv4 = "0.0.0.0" listen_addr_ipv4 = "0.0.0.0"
listen_addr_ipv6 = "::" listen_addr_ipv6 = "::"
# listen_unix_sock = "/var/run/telemt.sock" # Unix socket
# listen_unix_sock_perm = "0666" # Socket file permissions
# metrics_port = 9090 # metrics_port = 9090
# metrics_whitelist = ["127.0.0.1", "::1"] # metrics_whitelist = ["127.0.0.1", "::1"]
# Users to show in the startup log (tg:// links)
[general.links]
show = ["hello"]
# public_host = "proxy.example.com" # Host (IP or domain) for tg:// links
# public_port = 443 # Port for tg:// links (default: server.port)
# Listen on multiple interfaces/IPs (overrides listen_addr_*) # Listen on multiple interfaces/IPs (overrides listen_addr_*)
[[server.listeners]] [[server.listeners]]
ip = "0.0.0.0" ip = "0.0.0.0"

12
config.toml
View File

@@ -1,3 +1,7 @@
# === UI ===
# Users to show in the startup log (tg:// links)
show_link = ["hello"]
# === General Settings === # === General Settings ===
[general] [general]
prefer_ipv6 = true prefer_ipv6 = true
@@ -20,17 +24,9 @@ tls = true
port = 443 port = 443
listen_addr_ipv4 = "0.0.0.0" listen_addr_ipv4 = "0.0.0.0"
listen_addr_ipv6 = "::" listen_addr_ipv6 = "::"
# listen_unix_sock = "/var/run/telemt.sock" # Unix socket
# listen_unix_sock_perm = "0666" # Socket file permissions
# metrics_port = 9090 # metrics_port = 9090
# metrics_whitelist = ["127.0.0.1", "::1"] # metrics_whitelist = ["127.0.0.1", "::1"]
# Users to show in the startup log (tg:// links)
[general.links]
show = ["hello"]
# public_host = "proxy.example.com" # Host (IP or domain) for tg:// links
# public_port = 443 # Port for tg:// links (default: server.port)
# Listen on multiple interfaces/IPs (overrides listen_addr_*) # Listen on multiple interfaces/IPs (overrides listen_addr_*)
[[server.listeners]] [[server.listeners]]
ip = "0.0.0.0" ip = "0.0.0.0"

11
src/cli.rs
View File

@@ -186,6 +186,8 @@ fn generate_config(username: &str, secret: &str, port: u16, domain: &str) -> Str
r#"# Telemt MTProxy — auto-generated config r#"# Telemt MTProxy — auto-generated config
# Re-run `telemt --init` to regenerate # Re-run `telemt --init` to regenerate
show_link = ["{username}"]
[general] [general]
prefer_ipv6 = false prefer_ipv6 = false
fast_mode = true fast_mode = true
@@ -197,17 +199,10 @@ classic = false
secure = false secure = false
tls = true tls = true
[general.links]
show = ["{username}"]
# public_host = "proxy.example.com"
# public_port = 443
[server] [server]
port = {port} port = {port}
listen_addr_ipv4 = "0.0.0.0" listen_addr_ipv4 = "0.0.0.0"
listen_addr_ipv6 = "::" listen_addr_ipv6 = "::"
# listen_unix_sock = "/var/run/telemt.sock"
# listen_unix_sock_perm = "0666"
[[server.listeners]] [[server.listeners]]
ip = "0.0.0.0" ip = "0.0.0.0"
@@ -225,8 +220,6 @@ client_ack = 300
tls_domain = "{domain}" tls_domain = "{domain}"
mask = true mask = true
mask_port = 443 mask_port = 443
# mask_host = "{domain}"
# mask_unix_sock = "/var/run/nginx.sock"
fake_cert_len = 2048 fake_cert_len = 2048
[access] [access]

135
src/config/mod.rs
View File

@@ -4,7 +4,7 @@ use crate::error::{ProxyError, Result};
use chrono::{DateTime, Utc}; use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::collections::HashMap; use std::collections::HashMap;
use std::net::IpAddr; use std::net::{IpAddr, SocketAddr};
use std::path::Path; use std::path::Path;
// ============= Helper Defaults ============= // ============= Helper Defaults =============
@@ -39,6 +39,9 @@ fn default_keepalive() -> u64 {
fn default_ack_timeout() -> u64 { fn default_ack_timeout() -> u64 {
300 300
} }
fn default_listen_addr() -> String {
"0.0.0.0".to_string()
}
fn default_fake_cert_len() -> usize { fn default_fake_cert_len() -> usize {
2048 2048
} }
@@ -161,26 +164,6 @@ pub struct GeneralConfig {
#[serde(default)] #[serde(default)]
pub log_level: LogLevel, pub log_level: LogLevel,
#[serde(default)]
pub links: LinksConfig,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct LinksConfig {
/// Users whose tg:// links to show at startup.
#[serde(default)]
pub show: Vec<String>,
/// Public host (IP or domain) for tg:// link generation.
/// Overrides announce_ip / detected IP in links.
#[serde(default)]
pub public_host: Option<String>,
/// Public port for tg:// link generation.
/// Overrides server.port in links.
#[serde(default)]
pub public_port: Option<u16>,
} }
impl Default for GeneralConfig { impl Default for GeneralConfig {
@@ -196,7 +179,6 @@ impl Default for GeneralConfig {
middle_proxy_nat_probe: false, middle_proxy_nat_probe: false,
middle_proxy_nat_stun: None, middle_proxy_nat_stun: None,
log_level: LogLevel::Normal, log_level: LogLevel::Normal,
links: LinksConfig::default(),
} }
} }
} }
@@ -206,8 +188,8 @@ pub struct ServerConfig {
#[serde(default = "default_port")] #[serde(default = "default_port")]
pub port: u16, pub port: u16,
#[serde(default)] #[serde(default = "default_listen_addr")]
pub listen_addr_ipv4: Option<String>, pub listen_addr_ipv4: String,
#[serde(default)] #[serde(default)]
pub listen_addr_ipv6: Option<String>, pub listen_addr_ipv6: Option<String>,
@@ -215,11 +197,6 @@ pub struct ServerConfig {
#[serde(default)] #[serde(default)]
pub listen_unix_sock: Option<String>, pub listen_unix_sock: Option<String>,
/// Unix socket file permissions (octal string, e.g. "0666").
/// Applied after bind. If not set, inherits from process umask.
#[serde(default)]
pub listen_unix_sock_perm: Option<String>,
#[serde(default)] #[serde(default)]
pub metrics_port: Option<u16>, pub metrics_port: Option<u16>,
@@ -234,10 +211,9 @@ impl Default for ServerConfig {
fn default() -> Self { fn default() -> Self {
Self { Self {
port: default_port(), port: default_port(),
listen_addr_ipv4: None, listen_addr_ipv4: default_listen_addr(),
listen_addr_ipv6: Some("::".to_string()), listen_addr_ipv6: Some("::".to_string()),
listen_unix_sock: None, listen_unix_sock: None,
listen_unix_sock_perm: None,
metrics_port: None, metrics_port: None,
metrics_whitelist: default_metrics_whitelist(), metrics_whitelist: default_metrics_whitelist(),
listeners: Vec::new(), listeners: Vec::new(),
@@ -530,26 +506,15 @@ pub struct ProxyConfig {
/// If not set, defaults to 2 (matching Telegram's official `default 2;` in proxy-multi.conf). /// If not set, defaults to 2 (matching Telegram's official `default 2;` in proxy-multi.conf).
#[serde(default)] #[serde(default)]
pub default_dc: Option<u8>, pub default_dc: Option<u8>,
/// Non-fatal warnings collected during config loading.
#[serde(skip)]
pub warnings: Vec<String>,
} }
impl ProxyConfig { impl ProxyConfig {
pub fn load<P: AsRef<Path>>(path: P) -> Result<Self> { pub fn load<P: AsRef<Path>>(path: P) -> Result<Self> {
let content = std::fs::read_to_string(path) let content =
.map_err(|e| ProxyError::Config(e.to_string()))?; std::fs::read_to_string(path).map_err(|e| ProxyError::Config(e.to_string()))?;
// Pre-parse raw TOML to detect defaulted fields let mut config: ProxyConfig =
let raw: toml::Value = toml::from_str(&content) toml::from_str(&content).map_err(|e| ProxyError::Config(e.to_string()))?;
.map_err(|e| ProxyError::Config(e.to_string()))?;
let port_explicit = raw.get("server")
.and_then(|s| s.get("port"))
.is_some();
let mut config: ProxyConfig = toml::from_str(&content)
.map_err(|e| ProxyError::Config(e.to_string()))?;
// Validate secrets // Validate secrets
for (user, secret) in &config.access.users { for (user, secret) in &config.access.users {
@@ -601,51 +566,15 @@ impl ProxyConfig {
use rand::Rng; use rand::Rng;
config.censorship.fake_cert_len = rand::rng().gen_range(1024..4096); config.censorship.fake_cert_len = rand::rng().gen_range(1024..4096);
// Validate listen_unix_sock // Migration: Populate listeners if empty
if let Some(ref sock_path) = config.server.listen_unix_sock {
if sock_path.is_empty() {
return Err(ProxyError::Config(
"listen_unix_sock cannot be empty".to_string()
));
}
#[cfg(unix)]
if sock_path.len() > 107 {
return Err(ProxyError::Config(
format!("listen_unix_sock path too long: {} bytes (max 107)", sock_path.len())
));
}
#[cfg(not(unix))]
return Err(ProxyError::Config(
"listen_unix_sock is only supported on Unix platforms".to_string()
));
}
// Validate listen_unix_sock_perm
if let Some(ref perm_str) = config.server.listen_unix_sock_perm {
if config.server.listen_unix_sock.is_none() {
return Err(ProxyError::Config(
"listen_unix_sock_perm requires listen_unix_sock to be set".to_string()
));
}
u32::from_str_radix(perm_str, 8).map_err(|_| {
ProxyError::Config(format!(
"listen_unix_sock_perm must be an octal string (e.g. \"0666\"), got \"{}\"",
perm_str
))
})?;
}
// Migration: Populate listeners from legacy listen_addr_* fields.
if config.server.listeners.is_empty() { if config.server.listeners.is_empty() {
if let Some(ref ipv4_str) = config.server.listen_addr_ipv4 { if let Ok(ipv4) = config.server.listen_addr_ipv4.parse::<IpAddr>() {
if let Ok(ipv4) = ipv4_str.parse::<IpAddr>() { config.server.listeners.push(ListenerConfig {
config.server.listeners.push(ListenerConfig { ip: ipv4,
ip: ipv4, announce_ip: None,
announce_ip: None, });
});
}
} }
if let Some(ref 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>() { if let Ok(ipv6) = ipv6_str.parse::<IpAddr>() {
config.server.listeners.push(ListenerConfig { config.server.listeners.push(ListenerConfig {
ip: ipv6, ip: ipv6,
@@ -655,20 +584,6 @@ impl ProxyConfig {
} }
} }
// Validate: at least one listen endpoint must be configured.
if config.server.listeners.is_empty() && config.server.listen_unix_sock.is_none() {
return Err(ProxyError::Config(
"No listen address configured. Set [[server.listeners]], listen_addr_ipv4, or listen_unix_sock".to_string()
));
}
// Migration: show_link → general.links.show
if !config.show_link.is_empty() && config.general.links.show.is_empty() {
let migrated = config.show_link.resolve_users(&config.access.users)
.into_iter().cloned().collect::<Vec<_>>();
config.general.links.show = migrated;
}
// Migration: Populate upstreams if empty (Default Direct) // Migration: Populate upstreams if empty (Default Direct)
if config.upstreams.is_empty() { if config.upstreams.is_empty() {
config.upstreams.push(UpstreamConfig { config.upstreams.push(UpstreamConfig {
@@ -678,20 +593,6 @@ impl ProxyConfig {
}); });
} }
// Warnings for defaulted fields
if !config.server.listeners.is_empty() && !port_explicit {
config.warnings.push(format!(
"[server] port is not set; defaulting to {}",
config.server.port
));
}
if config.server.listen_unix_sock.is_some() && config.general.links.public_port.is_none() {
config.warnings.push(format!(
"[general.links] public_port is not set; using [server] port {} for tg:// links",
config.server.port
));
}
Ok(config) Ok(config)
} }

183
src/main.rs
View File

@@ -4,8 +4,6 @@ use std::net::SocketAddr;
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use tokio::net::TcpListener; use tokio::net::TcpListener;
#[cfg(unix)]
use tokio::net::UnixListener;
use tokio::signal; use tokio::signal;
use tokio::sync::Semaphore; use tokio::sync::Semaphore;
use tracing::{debug, error, info, warn}; use tracing::{debug, error, info, warn};
@@ -23,10 +21,7 @@ mod stream;
mod transport; mod transport;
mod util; mod util;
use crate::config::{ProxyConfig, LogLevel}; use crate::config::{LogLevel, ProxyConfig};
use crate::proxy::{ClientHandler, handle_client_stream};
#[cfg(unix)]
use crate::transport::{create_unix_listener, cleanup_unix_socket};
use crate::crypto::SecureRandom; use crate::crypto::SecureRandom;
use crate::ip_tracker::UserIpTracker; use crate::ip_tracker::UserIpTracker;
use crate::proxy::ClientHandler; use crate::proxy::ClientHandler;
@@ -104,31 +99,6 @@ fn parse_cli() -> (String, bool, Option<String>) {
(config_path, silent, log_level) (config_path, silent, log_level)
} }
fn print_proxy_links(host: &str, port: u16, config: &ProxyConfig) {
info!("--- Proxy Links ({}) ---", host);
for user_name in &config.general.links.show {
if let Some(secret) = config.access.users.get(user_name) {
info!("User: {}", user_name);
if config.general.modes.classic {
info!(" Classic: tg://proxy?server={}&port={}&secret={}",
host, port, secret);
}
if config.general.modes.secure {
info!(" DD: tg://proxy?server={}&port={}&secret=dd{}",
host, port, secret);
}
if config.general.modes.tls {
let domain_hex = hex::encode(&config.censorship.tls_domain);
info!(" EE-TLS: tg://proxy?server={}&port={}&secret=ee{}{}",
host, port, secret, domain_hex);
}
} else {
warn!("User '{}' listed in [general.links] show not found in [access.users]", user_name);
}
}
info!("------------------------");
}
#[tokio::main] #[tokio::main]
async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> { async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
let (config_path, cli_silent, cli_log_level) = parse_cli(); let (config_path, cli_silent, cli_log_level) = parse_cli();
@@ -200,10 +170,6 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
warn!("Using default tls_domain. Consider setting a custom domain."); warn!("Using default tls_domain. Consider setting a custom domain.");
} }
for w in &config.warnings {
warn!("{}", w);
}
let prefer_ipv6 = config.general.prefer_ipv6; let prefer_ipv6 = config.general.prefer_ipv6;
let use_middle_proxy = config.general.use_middle_proxy; let use_middle_proxy = config.general.use_middle_proxy;
let config = Arc::new(config); let config = Arc::new(config);
@@ -440,12 +406,35 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
listener_conf.ip listener_conf.ip
}; };
// Per-listener links (only when public_host is NOT set) if !config.show_link.is_empty() {
let links = &config.general.links; info!("--- Proxy Links ({}) ---", public_ip);
if links.public_host.is_none() && !links.show.is_empty() { for user_name in config.show_link.resolve_users(&config.access.users) {
let link_host = public_ip.to_string(); if let Some(secret) = config.access.users.get(user_name) {
let link_port = links.public_port.unwrap_or(config.server.port); info!("User: {}", user_name);
print_proxy_links(&link_host, link_port, &config); if config.general.modes.classic {
info!(
" Classic: tg://proxy?server={}&port={}&secret={}",
public_ip, config.server.port, secret
);
}
if config.general.modes.secure {
info!(
" DD: tg://proxy?server={}&port={}&secret=dd{}",
public_ip, config.server.port, secret
);
}
if config.general.modes.tls {
let domain_hex = hex::encode(&config.censorship.tls_domain);
info!(
" EE-TLS: tg://proxy?server={}&port={}&secret=ee{}{}",
public_ip, config.server.port, secret, domain_hex
);
}
} else {
warn!("User '{}' in show_link not found", user_name);
}
}
info!("------------------------");
} }
listeners.push(listener); listeners.push(listener);
@@ -456,109 +445,9 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
} }
} }
// Unix socket listener
#[cfg(unix)]
let unix_sock_path = if let Some(ref unix_path) = config.server.listen_unix_sock {
match create_unix_listener(unix_path) {
Ok(std_listener) => {
// Set socket file permissions if configured
if let Some(ref perm_str) = config.server.listen_unix_sock_perm {
if let Ok(mode) = u32::from_str_radix(perm_str, 8) {
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(
unix_path,
std::fs::Permissions::from_mode(mode),
)?;
}
}
let unix_listener = UnixListener::from_std(std_listener)?;
info!("Listening on unix:{}", unix_path);
let config = config.clone();
let stats = stats.clone();
let upstream_manager = upstream_manager.clone();
let replay_checker = replay_checker.clone();
let buffer_pool = buffer_pool.clone();
let rng = rng.clone();
let me_pool = me_pool.clone();
let unix_conn_counter = std::sync::Arc::new(
std::sync::atomic::AtomicU64::new(1)
);
tokio::spawn(async move {
loop {
match unix_listener.accept().await {
Ok((stream, _unix_addr)) => {
let conn_id = unix_conn_counter.fetch_add(
1, std::sync::atomic::Ordering::Relaxed
);
let fake_peer = SocketAddr::from(([127, 0, 0, 1], conn_id as u16));
let config = config.clone();
let stats = stats.clone();
let upstream_manager = upstream_manager.clone();
let replay_checker = replay_checker.clone();
let buffer_pool = buffer_pool.clone();
let rng = rng.clone();
let me_pool = me_pool.clone();
tokio::spawn(async move {
if let Err(e) = handle_client_stream(
stream, fake_peer, config, stats,
upstream_manager, replay_checker, buffer_pool, rng,
me_pool,
).await {
debug!(error = %e, "Unix socket connection error");
}
});
}
Err(e) => {
error!("Unix socket accept error: {}", e);
tokio::time::sleep(Duration::from_millis(100)).await;
}
}
}
});
Some(unix_path.clone())
}
Err(e) => {
error!("Failed to bind to unix:{}: {}", unix_path, e);
None
}
}
} else {
None
};
// Links with explicit public_host (independent of TCP listeners)
let links = &config.general.links;
if let Some(ref public_host) = links.public_host {
if !links.show.is_empty() {
let link_port = links.public_port.unwrap_or(config.server.port);
print_proxy_links(public_host, link_port, &config);
}
}
// Warn if links were configured but couldn't be shown
// (no TCP listeners succeeded and no public_host set)
let links = &config.general.links;
if listeners.is_empty() && links.public_host.is_none() && !links.show.is_empty() {
warn!("Proxy links not shown: no TCP listeners bound. Set [general.links] public_host or fix listener errors above.");
}
if listeners.is_empty() { if listeners.is_empty() {
#[cfg(unix)] error!("No listeners. Exiting.");
if unix_sock_path.is_none() { std::process::exit(1);
error!("No listeners. Exiting.");
std::process::exit(1);
}
#[cfg(not(unix))]
{
error!("No listeners. Exiting.");
std::process::exit(1);
}
} }
// Switch to user-configured log level after startup // Switch to user-configured log level after startup
@@ -624,13 +513,7 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
} }
match signal::ctrl_c().await { match signal::ctrl_c().await {
Ok(()) => { Ok(()) => info!("Shutting down..."),
info!("Shutting down...");
#[cfg(unix)]
if let Some(ref path) = unix_sock_path {
cleanup_unix_socket(path);
}
}
Err(e) => error!("Signal error: {}", e), Err(e) => error!("Signal error: {}", e),
} }

147
src/proxy/client.rs
View File

@@ -24,149 +24,6 @@ use crate::proxy::handshake::{HandshakeSuccess, handle_mtproto_handshake, handle
use crate::proxy::masking::handle_bad_client; use crate::proxy::masking::handle_bad_client;
use crate::proxy::middle_relay::handle_via_middle_proxy; use crate::proxy::middle_relay::handle_via_middle_proxy;
/// Handle a client connection from any stream type (TCP, Unix socket)
///
/// This is the generic entry point for client handling. Unlike `ClientHandler::new().run()`,
/// it skips TCP-specific socket configuration (TCP_NODELAY, keepalive, TCP_USER_TIMEOUT)
/// which is appropriate for non-TCP streams like Unix sockets.
pub async fn handle_client_stream<S>(
mut stream: S,
peer: SocketAddr,
config: Arc<ProxyConfig>,
stats: Arc<Stats>,
upstream_manager: Arc<UpstreamManager>,
replay_checker: Arc<ReplayChecker>,
buffer_pool: Arc<BufferPool>,
rng: Arc<SecureRandom>,
me_pool: Option<Arc<MePool>>,
) -> Result<()>
where
S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
{
stats.increment_connects_all();
debug!(peer = %peer, "New connection (generic stream)");
let handshake_timeout = Duration::from_secs(config.timeouts.client_handshake);
let stats_for_timeout = stats.clone();
// For non-TCP streams, use a synthetic local address
let local_addr: SocketAddr = format!("0.0.0.0:{}", config.server.port)
.parse()
.unwrap_or_else(|_| "0.0.0.0:443".parse().unwrap());
let result = timeout(handshake_timeout, async {
let mut first_bytes = [0u8; 5];
stream.read_exact(&mut first_bytes).await?;
let is_tls = tls::is_tls_handshake(&first_bytes[..3]);
debug!(peer = %peer, is_tls = is_tls, "Handshake type detected");
if is_tls {
let tls_len = u16::from_be_bytes([first_bytes[3], first_bytes[4]]) as usize;
if tls_len < 512 {
debug!(peer = %peer, tls_len = tls_len, "TLS handshake too short");
stats.increment_connects_bad();
let (reader, writer) = tokio::io::split(stream);
handle_bad_client(reader, writer, &first_bytes, &config).await;
return Ok(());
}
let mut handshake = vec![0u8; 5 + tls_len];
handshake[..5].copy_from_slice(&first_bytes);
stream.read_exact(&mut handshake[5..]).await?;
let (read_half, write_half) = tokio::io::split(stream);
let (mut tls_reader, tls_writer, _tls_user) = match handle_tls_handshake(
&handshake, read_half, write_half, peer,
&config, &replay_checker, &rng,
).await {
HandshakeResult::Success(result) => result,
HandshakeResult::BadClient { reader, writer } => {
stats.increment_connects_bad();
handle_bad_client(reader, writer, &handshake, &config).await;
return Ok(());
}
HandshakeResult::Error(e) => return Err(e),
};
debug!(peer = %peer, "Reading MTProto handshake through TLS");
let mtproto_data = tls_reader.read_exact(HANDSHAKE_LEN).await?;
let mtproto_handshake: [u8; HANDSHAKE_LEN] = mtproto_data[..].try_into()
.map_err(|_| ProxyError::InvalidHandshake("Short MTProto handshake".into()))?;
let (crypto_reader, crypto_writer, success) = match handle_mtproto_handshake(
&mtproto_handshake, tls_reader, tls_writer, peer,
&config, &replay_checker, true,
).await {
HandshakeResult::Success(result) => result,
HandshakeResult::BadClient { reader: _, writer: _ } => {
stats.increment_connects_bad();
debug!(peer = %peer, "Valid TLS but invalid MTProto handshake");
return Ok(());
}
HandshakeResult::Error(e) => return Err(e),
};
RunningClientHandler::handle_authenticated_static(
crypto_reader, crypto_writer, success,
upstream_manager, stats, config, buffer_pool, rng, me_pool,
local_addr,
).await
} else {
if !config.general.modes.classic && !config.general.modes.secure {
debug!(peer = %peer, "Non-TLS modes disabled");
stats.increment_connects_bad();
let (reader, writer) = tokio::io::split(stream);
handle_bad_client(reader, writer, &first_bytes, &config).await;
return Ok(());
}
let mut handshake = [0u8; HANDSHAKE_LEN];
handshake[..5].copy_from_slice(&first_bytes);
stream.read_exact(&mut handshake[5..]).await?;
let (read_half, write_half) = tokio::io::split(stream);
let (crypto_reader, crypto_writer, success) = match handle_mtproto_handshake(
&handshake, read_half, write_half, peer,
&config, &replay_checker, false,
).await {
HandshakeResult::Success(result) => result,
HandshakeResult::BadClient { reader, writer } => {
stats.increment_connects_bad();
handle_bad_client(reader, writer, &handshake, &config).await;
return Ok(());
}
HandshakeResult::Error(e) => return Err(e),
};
RunningClientHandler::handle_authenticated_static(
crypto_reader, crypto_writer, success,
upstream_manager, stats, config, buffer_pool, rng, me_pool,
local_addr,
).await
}
}).await;
match result {
Ok(Ok(())) => {
debug!(peer = %peer, "Connection handled successfully");
Ok(())
}
Ok(Err(e)) => {
debug!(peer = %peer, error = %e, "Handshake failed");
Err(e)
}
Err(_) => {
stats_for_timeout.increment_handshake_timeouts();
debug!(peer = %peer, "Handshake timeout");
Err(ProxyError::TgHandshakeTimeout)
}
}
}
pub struct ClientHandler; pub struct ClientHandler;
pub struct RunningClientHandler { pub struct RunningClientHandler {
@@ -418,9 +275,9 @@ impl RunningClientHandler {
/// Main dispatch after successful handshake. /// Main dispatch after successful handshake.
/// Two modes: /// Two modes:
/// - Direct: TCP relay to TG DC (existing behavior) /// - Direct: TCP relay to TG DC (existing behavior)
/// - Middle Proxy: RPC multiplex through ME pool (new — supports CDN DCs) /// - Middle Proxy: RPC multiplex through ME pool (new — supports CDN DCs)
pub(crate) async fn handle_authenticated_static<R, W>( async fn handle_authenticated_static<R, W>(
client_reader: CryptoReader<R>, client_reader: CryptoReader<R>,
client_writer: CryptoWriter<W>, client_writer: CryptoWriter<W>,
success: HandshakeSuccess, success: HandshakeSuccess,

6
src/proxy/mod.rs
View File

@@ -1,13 +1,13 @@
//! Proxy Defs //! Proxy Defs
pub mod client; pub mod client;
pub(crate) mod direct_relay; pub mod direct_relay;
pub mod handshake; pub mod handshake;
pub mod masking; pub mod masking;
pub(crate) mod middle_relay; pub mod middle_relay;
pub mod relay; pub mod relay;
pub use client::{ClientHandler, handle_client_stream}; pub use client::ClientHandler;
pub use handshake::*; pub use handshake::*;
pub use masking::*; pub use masking::*;
pub use relay::*; pub use relay::*;

45
src/transport/socket.rs
View File

@@ -202,51 +202,6 @@ pub fn create_listener(addr: SocketAddr, options: &ListenOptions) -> Result<Sock
Ok(socket) Ok(socket)
} }
/// Create a Unix socket listener with stale socket detection.
///
/// If the socket file already exists, attempts to connect to it:
/// - If connection succeeds → another instance is running → returns AddrInUse error
/// - If connection fails → stale socket → removes it and binds
#[cfg(unix)]
pub fn create_unix_listener(path: &str) -> Result<std::os::unix::net::UnixListener> {
use std::os::unix::net::UnixListener;
use std::path::Path;
let socket_path = Path::new(path);
if socket_path.exists() {
match std::os::unix::net::UnixStream::connect(socket_path) {
Ok(_) => {
return Err(std::io::Error::new(
std::io::ErrorKind::AddrInUse,
format!("Unix socket {} is already in use by another process", path)
));
}
Err(_) => {
debug!("Removing stale Unix socket: {}", path);
std::fs::remove_file(socket_path)?;
}
}
}
let listener = UnixListener::bind(socket_path)?;
listener.set_nonblocking(true)?;
debug!("Created Unix socket listener at {}", path);
Ok(listener)
}
/// Remove Unix socket file on shutdown
#[cfg(unix)]
pub fn cleanup_unix_socket(path: &str) {
if std::path::Path::new(path).exists() {
match std::fs::remove_file(path) {
Ok(_) => debug!("Cleaned up Unix socket: {}", path),
Err(e) => debug!("Failed to remove Unix socket {}: {}", path, e),
}
}
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;