Merge branch 'main-stage' into flow

Cargo.toml

@@ -53,6 +53,8 @@ anyhow = "1.0"
 
 # HTTP
 reqwest = { version = "0.12", features = ["rustls-tls"], default-features = false }
+notify = { version = "6", features = ["macos_fsevent"] }
+ipnetwork = "0.20"
 hyper = { version = "1", features = ["server", "http1"] }
 hyper-util = { version = "0.1", features = ["tokio", "server-auto"] }
 http-body-util = "0.1"

src/config/defaults.rs

@@ -1,5 +1,6 @@
 use std::net::IpAddr;
 use std::collections::HashMap;
+use ipnetwork::IpNetwork;
 use serde::Deserialize;
 
 // Helper defaults kept private to the config module.
@@ -66,8 +67,11 @@ pub(crate) fn default_weight() -> u16 {
     1
 }
 
-pub(crate) fn default_metrics_whitelist() -> Vec<IpAddr> {
-    vec!["127.0.0.1".parse().unwrap(), "::1".parse().unwrap()]
+pub(crate) fn default_metrics_whitelist() -> Vec<IpNetwork> {
+    vec![
+        "127.0.0.1/32".parse().unwrap(),
+        "::1/128".parse().unwrap(),
+    ]
 }
 
 pub(crate) fn default_prefer_4() -> u8 {

src/config/hot_reload.rs

@@ -0,0 +1,433 @@
+//! Hot-reload: watches the config file via inotify (Linux) / FSEvents (macOS)
+//! / ReadDirectoryChangesW (Windows) using the `notify` crate.
+//! SIGHUP is also supported on Unix as an additional manual trigger.
+//!
+//! # What can be reloaded without restart
+//!
+//! | Section   | Field                         | Effect                            |
+//! |-----------|-------------------------------|-----------------------------------|
+//! | `general` | `log_level`                   | Filter updated via `log_level_tx` |
+//! | `general` | `ad_tag`                      | Passed on next connection         |
+//! | `general` | `middle_proxy_pool_size`      | Passed on next connection         |
+//! | `general` | `me_keepalive_*`              | Passed on next connection         |
+//! | `access`  | All user/quota fields         | Effective immediately             |
+//!
+//! Fields that require re-binding sockets (`server.port`, `censorship.*`,
+//! `network.*`, `use_middle_proxy`) are **not** applied; a warning is emitted.
+
+use std::net::IpAddr;
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use notify::{EventKind, RecursiveMode, Watcher, recommended_watcher};
+use tokio::sync::{mpsc, watch};
+use tracing::{error, info, warn};
+
+use crate::config::LogLevel;
+use super::load::ProxyConfig;
+
+// ── Hot fields ────────────────────────────────────────────────────────────────
+
+/// Fields that are safe to swap without restarting listeners.
+#[derive(Debug, Clone, PartialEq)]
+pub struct HotFields {
+    pub log_level:               LogLevel,
+    pub ad_tag:                  Option<String>,
+    pub middle_proxy_pool_size:  usize,
+    pub me_keepalive_enabled:    bool,
+    pub me_keepalive_interval_secs: u64,
+    pub me_keepalive_jitter_secs:   u64,
+    pub me_keepalive_payload_random: bool,
+    pub access:                  crate::config::AccessConfig,
+}
+
+impl HotFields {
+    pub fn from_config(cfg: &ProxyConfig) -> Self {
+        Self {
+            log_level:               cfg.general.log_level.clone(),
+            ad_tag:                  cfg.general.ad_tag.clone(),
+            middle_proxy_pool_size:  cfg.general.middle_proxy_pool_size,
+            me_keepalive_enabled:    cfg.general.me_keepalive_enabled,
+            me_keepalive_interval_secs: cfg.general.me_keepalive_interval_secs,
+            me_keepalive_jitter_secs:   cfg.general.me_keepalive_jitter_secs,
+            me_keepalive_payload_random: cfg.general.me_keepalive_payload_random,
+            access:                  cfg.access.clone(),
+        }
+    }
+}
+
+// ── Helpers ───────────────────────────────────────────────────────────────────
+
+/// Warn if any non-hot fields changed (require restart).
+fn warn_non_hot_changes(old: &ProxyConfig, new: &ProxyConfig) {
+    if old.server.port != new.server.port {
+        warn!(
+            "config reload: server.port changed ({} → {}); restart required",
+            old.server.port, new.server.port
+        );
+    }
+    if old.censorship.tls_domain != new.censorship.tls_domain {
+        warn!(
+            "config reload: censorship.tls_domain changed ('{}' → '{}'); restart required",
+            old.censorship.tls_domain, new.censorship.tls_domain
+        );
+    }
+    if old.network.ipv4 != new.network.ipv4 || old.network.ipv6 != new.network.ipv6 {
+        warn!("config reload: network.ipv4/ipv6 changed; restart required");
+    }
+    if old.general.use_middle_proxy != new.general.use_middle_proxy {
+        warn!("config reload: use_middle_proxy changed; restart required");
+    }
+}
+
+/// Resolve the public host for link generation — mirrors the logic in main.rs.
+///
+/// Priority:
+/// 1. `[general.links] public_host` — explicit override in config
+/// 2. `detected_ip_v4` — from STUN/interface probe at startup
+/// 3. `detected_ip_v6` — fallback
+/// 4. `"UNKNOWN"` — warn the user to set `public_host`
+fn resolve_link_host(
+    cfg: &ProxyConfig,
+    detected_ip_v4: Option<IpAddr>,
+    detected_ip_v6: Option<IpAddr>,
+) -> String {
+    if let Some(ref h) = cfg.general.links.public_host {
+        return h.clone();
+    }
+    detected_ip_v4
+        .or(detected_ip_v6)
+        .map(|ip| ip.to_string())
+        .unwrap_or_else(|| {
+            warn!(
+                "config reload: could not determine public IP for proxy links. \
+                 Set [general.links] public_host in config."
+            );
+            "UNKNOWN".to_string()
+        })
+}
+
+/// Print TG proxy links for a single user — mirrors print_proxy_links() in main.rs.
+fn print_user_links(user: &str, secret: &str, host: &str, port: u16, cfg: &ProxyConfig) {
+    info!(target: "telemt::links", "--- New user: {} ---", user);
+    if cfg.general.modes.classic {
+        info!(
+            target: "telemt::links",
+            "  Classic: tg://proxy?server={}&port={}&secret={}",
+            host, port, secret
+        );
+    }
+    if cfg.general.modes.secure {
+        info!(
+            target: "telemt::links",
+            "  DD:      tg://proxy?server={}&port={}&secret=dd{}",
+            host, port, secret
+        );
+    }
+    if cfg.general.modes.tls {
+        let mut domains = vec![cfg.censorship.tls_domain.clone()];
+        for d in &cfg.censorship.tls_domains {
+            if !domains.contains(d) {
+                domains.push(d.clone());
+            }
+        }
+        for domain in &domains {
+            let domain_hex = hex::encode(domain.as_bytes());
+            info!(
+                target: "telemt::links",
+                "  EE-TLS:  tg://proxy?server={}&port={}&secret=ee{}{}",
+                host, port, secret, domain_hex
+            );
+        }
+    }
+    info!(target: "telemt::links", "--------------------");
+}
+
+/// Log all detected changes and emit TG links for new users.
+fn log_changes(
+    old_hot: &HotFields,
+    new_hot: &HotFields,
+    new_cfg: &ProxyConfig,
+    log_tx: &watch::Sender<LogLevel>,
+    detected_ip_v4: Option<IpAddr>,
+    detected_ip_v6: Option<IpAddr>,
+) {
+    if old_hot.log_level != new_hot.log_level {
+        info!(
+            "config reload: log_level: '{}' → '{}'",
+            old_hot.log_level, new_hot.log_level
+        );
+        log_tx.send(new_hot.log_level.clone()).ok();
+    }
+
+    if old_hot.ad_tag != new_hot.ad_tag {
+        info!(
+            "config reload: ad_tag: {} → {}",
+            old_hot.ad_tag.as_deref().unwrap_or("none"),
+            new_hot.ad_tag.as_deref().unwrap_or("none"),
+        );
+    }
+
+    if old_hot.middle_proxy_pool_size != new_hot.middle_proxy_pool_size {
+        info!(
+            "config reload: middle_proxy_pool_size: {} → {}",
+            old_hot.middle_proxy_pool_size, new_hot.middle_proxy_pool_size,
+        );
+    }
+
+    if old_hot.me_keepalive_enabled        != new_hot.me_keepalive_enabled
+    || old_hot.me_keepalive_interval_secs  != new_hot.me_keepalive_interval_secs
+    || old_hot.me_keepalive_jitter_secs    != new_hot.me_keepalive_jitter_secs
+    || old_hot.me_keepalive_payload_random != new_hot.me_keepalive_payload_random
+    {
+        info!(
+            "config reload: me_keepalive: enabled={} interval={}s jitter={}s random_payload={}",
+            new_hot.me_keepalive_enabled,
+            new_hot.me_keepalive_interval_secs,
+            new_hot.me_keepalive_jitter_secs,
+            new_hot.me_keepalive_payload_random,
+        );
+    }
+
+    if old_hot.access.users != new_hot.access.users {
+        let mut added: Vec<&String> = new_hot.access.users.keys()
+            .filter(|u| !old_hot.access.users.contains_key(*u))
+            .collect();
+        added.sort();
+
+        let mut removed: Vec<&String> = old_hot.access.users.keys()
+            .filter(|u| !new_hot.access.users.contains_key(*u))
+            .collect();
+        removed.sort();
+
+        let mut changed: Vec<&String> = new_hot.access.users.keys()
+            .filter(|u| {
+                old_hot.access.users.get(*u)
+                    .map(|s| s != &new_hot.access.users[*u])
+                    .unwrap_or(false)
+            })
+            .collect();
+        changed.sort();
+
+        if !added.is_empty() {
+            info!(
+                "config reload: users added: [{}]",
+                added.iter().map(|s| s.as_str()).collect::<Vec<_>>().join(", ")
+            );
+            let host = resolve_link_host(new_cfg, detected_ip_v4, detected_ip_v6);
+            let port = new_cfg.general.links.public_port.unwrap_or(new_cfg.server.port);
+            for user in &added {
+                if let Some(secret) = new_hot.access.users.get(*user) {
+                    print_user_links(user, secret, &host, port, new_cfg);
+                }
+            }
+        }
+        if !removed.is_empty() {
+            info!(
+                "config reload: users removed: [{}]",
+                removed.iter().map(|s| s.as_str()).collect::<Vec<_>>().join(", ")
+            );
+        }
+        if !changed.is_empty() {
+            info!(
+                "config reload: users secret changed: [{}]",
+                changed.iter().map(|s| s.as_str()).collect::<Vec<_>>().join(", ")
+            );
+        }
+    }
+
+    if old_hot.access.user_max_tcp_conns != new_hot.access.user_max_tcp_conns {
+        info!(
+            "config reload: user_max_tcp_conns updated ({} entries)",
+            new_hot.access.user_max_tcp_conns.len()
+        );
+    }
+    if old_hot.access.user_expirations != new_hot.access.user_expirations {
+        info!(
+            "config reload: user_expirations updated ({} entries)",
+            new_hot.access.user_expirations.len()
+        );
+    }
+    if old_hot.access.user_data_quota != new_hot.access.user_data_quota {
+        info!(
+            "config reload: user_data_quota updated ({} entries)",
+            new_hot.access.user_data_quota.len()
+        );
+    }
+    if old_hot.access.user_max_unique_ips != new_hot.access.user_max_unique_ips {
+        info!(
+            "config reload: user_max_unique_ips updated ({} entries)",
+            new_hot.access.user_max_unique_ips.len()
+        );
+    }
+}
+
+/// Load config, validate, diff against current, and broadcast if changed.
+fn reload_config(
+    config_path: &PathBuf,
+    config_tx: &watch::Sender<Arc<ProxyConfig>>,
+    log_tx: &watch::Sender<LogLevel>,
+    detected_ip_v4: Option<IpAddr>,
+    detected_ip_v6: Option<IpAddr>,
+) {
+    let new_cfg = match ProxyConfig::load(config_path) {
+        Ok(c) => c,
+        Err(e) => {
+            error!("config reload: failed to parse {:?}: {}", config_path, e);
+            return;
+        }
+    };
+
+    if let Err(e) = new_cfg.validate() {
+        error!("config reload: validation failed: {}; keeping old config", e);
+        return;
+    }
+
+    let old_cfg = config_tx.borrow().clone();
+    let old_hot = HotFields::from_config(&old_cfg);
+    let new_hot = HotFields::from_config(&new_cfg);
+
+    if old_hot == new_hot {
+        return;
+    }
+
+    warn_non_hot_changes(&old_cfg, &new_cfg);
+    log_changes(&old_hot, &new_hot, &new_cfg, log_tx, detected_ip_v4, detected_ip_v6);
+    config_tx.send(Arc::new(new_cfg)).ok();
+}
+
+// ── Public API ────────────────────────────────────────────────────────────────
+
+/// Spawn the hot-reload watcher task.
+///
+/// Uses `notify` (inotify on Linux) to detect file changes instantly.
+/// SIGHUP is also handled on Unix as an additional manual trigger.
+///
+/// `detected_ip_v4` / `detected_ip_v6` are the IPs discovered during the
+/// startup probe — used when generating proxy links for newly added users,
+/// matching the same logic as the startup output.
+pub fn spawn_config_watcher(
+    config_path: PathBuf,
+    initial: Arc<ProxyConfig>,
+    detected_ip_v4: Option<IpAddr>,
+    detected_ip_v6: Option<IpAddr>,
+) -> (watch::Receiver<Arc<ProxyConfig>>, watch::Receiver<LogLevel>) {
+    let initial_level = initial.general.log_level.clone();
+    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.
+    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).
+    let config_path = match config_path.canonicalize() {
+        Ok(p) => p,
+        Err(_) => config_path.to_path_buf(), // file doesn't exist yet, use as-is
+    };
+
+    // 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.
+    let watch_dir = config_path
+        .parent()
+        .unwrap_or_else(|| std::path::Path::new("."))
+        .to_path_buf();
+
+    let config_file = config_path.clone();
+    let tx_clone    = notify_tx.clone();
+
+    let watcher_result = 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;
+        }
+        let relevant = matches!(
+            event.kind,
+            EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_)
+        );
+        if relevant {
+            let _ = tx_clone.try_send(());
+        }
+    });
+
+    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)
+}

src/config/mod.rs

@@ -3,6 +3,7 @@
 pub(crate) mod defaults;
 mod types;
 mod load;
+pub mod hot_reload;
 
 pub use load::ProxyConfig;
 pub use types::*;

src/config/types.rs

@@ -1,4 +1,5 @@
 use chrono::{DateTime, Utc};
+use ipnetwork::IpNetwork;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::net::IpAddr;
@@ -304,7 +305,7 @@ pub struct ServerConfig {
     pub metrics_port: Option<u16>,
 
     #[serde(default = "default_metrics_whitelist")]
-    pub metrics_whitelist: Vec<IpAddr>,
+    pub metrics_whitelist: Vec<IpNetwork>,
 
     #[serde(default)]
     pub listeners: Vec<ListenerConfig>,
@@ -412,7 +413,7 @@ impl Default for AntiCensorshipConfig {
     }
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct AccessConfig {
     #[serde(default)]
     pub users: HashMap<String, String>,

src/main.rs

@@ -28,6 +28,7 @@ mod tls_front;
 mod util;
 
 use crate::config::{LogLevel, ProxyConfig};
+use crate::config::hot_reload::spawn_config_watcher;
 use crate::crypto::SecureRandom;
 use crate::ip_tracker::UserIpTracker;
 use crate::network::probe::{decide_network_capabilities, log_probe_result, run_probe};
@@ -683,6 +684,19 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
         detected_ip_v4, detected_ip_v6
     );
 
+    // ── Hot-reload watcher ────────────────────────────────────────────────
+    // Uses inotify to detect file changes instantly (SIGHUP also works).
+    // detected_ip_v4/v6 are passed so newly added users get correct TG links.
+    let (config_rx, mut log_level_rx): (
+        tokio::sync::watch::Receiver<Arc<ProxyConfig>>,
+        tokio::sync::watch::Receiver<LogLevel>,
+    ) = spawn_config_watcher(
+        std::path::PathBuf::from(&config_path),
+        config.clone(),
+        detected_ip_v4,
+        detected_ip_v6,
+    );
+
     let mut listeners = Vec::new();
 
     for listener_conf in &config.server.listeners {
@@ -789,7 +803,7 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
 
         has_unix_listener = true;
 
-        let config = config.clone();
+        let mut config_rx_unix: tokio::sync::watch::Receiver<Arc<ProxyConfig>> = config_rx.clone();
         let stats = stats.clone();
         let upstream_manager = upstream_manager.clone();
         let replay_checker = replay_checker.clone();
@@ -808,7 +822,7 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
                         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 % 65535) as u16));
 
-                        let config = config.clone();
+                        let config = config_rx_unix.borrow_and_update().clone();
                         let stats = stats.clone();
                         let upstream_manager = upstream_manager.clone();
                         let replay_checker = replay_checker.clone();
@@ -855,6 +869,20 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
         .reload(runtime_filter)
         .expect("Failed to switch log filter");
 
+    // Apply log_level changes from hot-reload to the tracing filter.
+    tokio::spawn(async move {
+        loop {
+            if log_level_rx.changed().await.is_err() {
+                break;
+            }
+            let level = log_level_rx.borrow_and_update().clone();
+            let new_filter = tracing_subscriber::EnvFilter::new(level.to_filter_str());
+            if let Err(e) = filter_handle.reload(new_filter) {
+                tracing::error!("config reload: failed to update log filter: {}", e);
+            }
+        }
+    });
+
     if let Some(port) = config.server.metrics_port {
         let stats = stats.clone();
         let whitelist = config.server.metrics_whitelist.clone();
@@ -863,8 +891,8 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
         });
     }
 
-    for (listener, listener_proxy_protocol) in listeners {
-        let config = config.clone();
+    for listener in listeners {
+        let mut config_rx: tokio::sync::watch::Receiver<Arc<ProxyConfig>> = config_rx.clone();
         let stats = stats.clone();
         let upstream_manager = upstream_manager.clone();
         let replay_checker = replay_checker.clone();
@@ -878,7 +906,7 @@ match crate::transport::middle_proxy::fetch_proxy_secret(proxy_secret_path).awai
             loop {
                 match listener.accept().await {
                     Ok((stream, peer_addr)) => {
-                        let config = config.clone();
+                        let config = config_rx.borrow_and_update().clone();
                         let stats = stats.clone();
                         let upstream_manager = upstream_manager.clone();
                         let replay_checker = replay_checker.clone();

src/metrics.rs

@@ -1,5 +1,5 @@
 use std::convert::Infallible;
-use std::net::{IpAddr, SocketAddr};
+use std::net::SocketAddr;
 use std::sync::Arc;
 
 use http_body_util::Full;
@@ -7,12 +7,13 @@ use hyper::body::Bytes;
 use hyper::server::conn::http1;
 use hyper::service::service_fn;
 use hyper::{Request, Response, StatusCode};
+use ipnetwork::IpNetwork;
 use tokio::net::TcpListener;
 use tracing::{info, warn, debug};
 
 use crate::stats::Stats;
 
-pub async fn serve(port: u16, stats: Arc<Stats>, whitelist: Vec<IpAddr>) {
+pub async fn serve(port: u16, stats: Arc<Stats>, whitelist: Vec<IpNetwork>) {
     let addr = SocketAddr::from(([0, 0, 0, 0], port));
     let listener = match TcpListener::bind(addr).await {
         Ok(l) => l,
@@ -32,7 +33,7 @@ pub async fn serve(port: u16, stats: Arc<Stats>, whitelist: Vec<IpAddr>) {
             }
         };
 
-        if !whitelist.is_empty() && !whitelist.contains(&peer.ip()) {
+        if !whitelist.is_empty() && !whitelist.iter().any(|net| net.contains(peer.ip())) {
             debug!(peer = %peer, "Metrics request denied by whitelist");
             continue;
         }