alighieri 0.4.0

//! The listener and accept loop.

use std::io::ErrorKind;
use std::net::{IpAddr, SocketAddr};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;

use tokio::net::TcpListener;
use tokio::sync::watch;
use tokio::sync::RwLock;
use tokio::sync::Semaphore;
use tokio::task::{JoinHandle, JoinSet};
use tracing::{debug, error, info, warn};

use crate::abuse::AbuseControls;
use crate::acl::Scope;
use crate::auth::{CommandAuth, UserDb};
use crate::client_stream::ClientStream;
use crate::config::Config;
use crate::connection::{Connection, ConnectionResources};
use crate::dns::DnsResolver;
use crate::errors::Result;
use crate::metrics::{self, Metrics};
use crate::net::Cidr;
use crate::tls;

/// A bound SOCKS5 server ready to accept connections.
pub struct Server {
    state: Arc<RwLock<ServerState>>,
    process_config: Arc<Config>,
    listener: TcpListener,
    max_connections: usize,
    metrics: Arc<Metrics>,
    abuse: Arc<AbuseControls>,
    /// Restart-only registry of active plugins. Empty unless registered via
    /// [`Server::with_plugins`]; not part of the hot-reloaded [`ServerState`].
    #[cfg(feature = "plugins")]
    plugins: Arc<crate::plugin::PluginHost>,
    metrics_addr: Option<SocketAddr>,
    metrics_listener: Mutex<Option<TcpListener>>,
    tls_listener: Option<tls::TlsListener>,
    /// Background ACME renewal task, aborted when the `Server` is dropped (the
    /// same `AbortOnDrop` guard used for the metrics task). Held only for that
    /// drop side effect, so a `Server` bound and dropped without running to
    /// process exit does not leak it.
    #[allow(dead_code)]
    acme_driver: Option<AbortOnDrop<()>>,
    has_run: AtomicBool,
    /// Set to `true` by [`Server::begin_shutdown`] to tell the accept loop to
    /// stop accepting and drain in-flight connections. `send_replace` latches the
    /// value, so a signal that arrives before `run` subscribes is not missed.
    shutdown: watch::Sender<bool>,
}

struct ServerState {
    config: Arc<Config>,
    users: Arc<UserDb>,
    command_auth: Option<Arc<CommandAuth>>,
    dns_resolver: Arc<DnsResolver>,
}

impl Server {
    /// Binds the listener and loads the user database (if configured).
    ///
    /// Emits warnings for configurations that would silently deny all traffic,
    /// since deny-by-default can otherwise be surprising.
    pub async fn bind(config: Config) -> Result<Server> {
        // Reject startup-only footguns (e.g. an unauthenticated metrics endpoint
        // exposed off loopback) before binding anything.
        config.validate_startup()?;
        let users = load_users(&config).await?;
        warn_config_footguns(&config);
        // Build the acceptor and (for ACME) the renewal driver up front so config
        // errors surface before binding, but defer spawning the driver until the
        // listener is bound (below).
        let tls_setup = tls::load_acceptor(config.tls.as_ref())?;

        let (metrics_addr, metrics_listener) = match config.metrics_listen {
            Some(addr) => {
                let listener = TcpListener::bind(addr).await?;
                let listen = listener.local_addr()?;
                // A non-loopback bind is only reachable here because the config
                // set `metrics.allowpublic` (it is refused otherwise), so this is
                // a reminder the operator opted into an unauthenticated endpoint.
                // Canonicalize so an IPv4-mapped loopback matches `validate_startup`.
                let canonical = listen.ip().to_canonical();
                if canonical.is_unspecified() || !canonical.is_loopback() {
                    warn!(
                        listen = %listen,
                        "metrics endpoint is exposed off loopback (metrics.allowpublic is set) and is unauthenticated; protect it with network access controls"
                    );
                }
                (Some(listen), Some(listener))
            }
            None => (None, None),
        };
        let listener = TcpListener::bind(config.internal).await?;
        let listen = listener.local_addr()?;
        if let Some(tls) = &config.tls {
            info!(listen = %listen, "listening with TLS");
            // The port stays fixed for the life of the process (it is not
            // reloadable), so this startup-only check is sufficient. The
            // ACME-vs-proxyprotocol check lives in `warn_config_footguns` so it
            // also fires on reload, where `proxyprotocol` can change.
            if matches!(tls, crate::config::TlsConfig::Acme(_)) && listen.port() != 443 {
                warn!(
                    listen = %listen,
                    "ACME uses TLS-ALPN-01, which Let's Encrypt validates on port 443; ensure this listener is reachable on port 443 (directly or via forwarding) or certificate issuance will fail"
                );
            }
        } else {
            info!(listen = %listen, "listening");
        }
        // Now the listener is bound, spawn the ACME renewal driver (if any): a
        // failed bind above cannot leak it, and validation cannot start before
        // the listener can answer the TLS-ALPN-01 challenge.
        let (tls_listener, acme_driver) = match tls_setup {
            Some(setup) => {
                // Hold the renewal task as an abort-on-drop handle so a `Server`
                // bound and dropped without running to process exit (e.g. in
                // tests) does not leak it.
                let driver = setup.acme_driver.map(|d| AbortOnDrop(tokio::spawn(d)));
                (Some(setup.listener), driver)
            }
            None => (None, None),
        };
        let max_connections = config.max_connections;
        let abuse = AbuseControls::new(config.rate_limits.clone());
        let mut process_config = config.clone();
        process_config.internal = listen;
        process_config.metrics_listen = metrics_addr;
        let process_config = Arc::new(process_config);

        Ok(Server {
            state: Arc::new(RwLock::new(ServerState {
                config: process_config.clone(),
                users: Arc::new(users),
                command_auth: build_command_auth(&config),
                dns_resolver: Arc::new(DnsResolver::new()),
            })),
            process_config,
            listener,
            max_connections,
            metrics: Metrics::new(),
            abuse,
            #[cfg(feature = "plugins")]
            plugins: Arc::new(crate::plugin::PluginHost::default()),
            metrics_addr,
            metrics_listener: Mutex::new(metrics_listener),
            tls_listener,
            acme_driver,
            has_run: AtomicBool::new(false),
            shutdown: watch::channel(false).0,
        })
    }

    /// Registers the plugin set (a build-time registry) for this server. Because
    /// plugins are statically linked, this is how a private build injects its
    /// compiled-in plugins; call it once after [`Server::bind`] and before
    /// [`Server::run`]. The set is restart-only — a config reload does not change
    /// it.
    #[cfg(feature = "plugins")]
    pub fn with_plugins(mut self, plugins: crate::plugin::PluginHost) -> Self {
        self.plugins = Arc::new(plugins);
        self
    }

    /// Returns the actual local address the listener is bound to (useful when
    /// the configured port was `0`).
    pub fn local_addr(&self) -> std::io::Result<SocketAddr> {
        self.listener.local_addr()
    }

    /// Returns the metrics endpoint address when metrics are enabled.
    pub fn metrics_addr(&self) -> std::io::Result<Option<SocketAddr>> {
        Ok(self.metrics_addr)
    }

    /// Signals the accept loop to stop accepting new connections and drain the
    /// in-flight ones (see [`Server::run`]). Idempotent, and safe to call before
    /// `run` starts — the signal latches, so it is observed once `run` begins.
    pub fn begin_shutdown(&self) {
        let _ = self.shutdown.send_replace(true);
    }

    /// Replaces the runtime configuration used for newly accepted
    /// connections. Startup resources such as listener addresses, logging
    /// sinks, and the max-connection semaphore require a process restart.
    pub async fn reload(&self, mut config: Config) -> Result<()> {
        warn_restart_required_changes(&self.process_config, &config);
        let users = load_users(&config).await?;
        preserve_process_config(&mut config, &self.process_config);
        // Warn on the *effective* config: `preserve_process_config` has just
        // restored the restart-only fields (`internal`, `tls`) to their live
        // values, so e.g. a reload requesting a loopback `internal` no longer
        // suppresses the open-proxy warning while the public listener stays live
        // until restart. Reloadable fields (rules, socksmethod, proxyprotocol)
        // already hold their new values here, so those footguns stay accurate.
        warn_config_footguns(&config);
        let command_auth = build_command_auth(&config);
        let rate_limits = config.rate_limits.clone();

        // Acquiring the state write lock is the last `.await`; everything below it
        // is synchronous, so the reload is *cancellation*-atomic — the driver
        // races reload against shutdown, and a cancellation can only land above,
        // before anything is applied (never with one half done). Swap the runtime
        // state under the lock, release it, then retune the abuse controls: that
        // step is O(tracked clients), so keeping it out of the state lock avoids
        // stalling the accept loop's `state.read()`. A concurrent task on another
        // worker may briefly see the new state before the abuse retune lands, but
        // that window is tiny and the rate limits are approximate, so it is benign
        // (and shorter than the old order, which updated abuse first).
        {
            let mut state = self.state.write().await;
            state.config = Arc::new(config);
            state.users = Arc::new(users);
            state.command_auth = command_auth;
            // The DNS resolver (and its TTL cache) is intentionally preserved
            // across reloads: the resolution *policy* lives in the config and is
            // read per lookup, so a reload takes effect on the next resolve
            // without discarding the cache and triggering a re-resolve storm. The
            // deny policy is also applied per call, so newly denied categories
            // take effect immediately.
        }
        self.abuse.update_config(rate_limits);
        info!("configuration reloaded");
        Ok(())
    }

    /// Runs the accept loop until a fatal error occurs.
    ///
    /// Per-connection errors are logged and never abort the loop. A semaphore
    /// caps the number of concurrent connections at `max_connections`.
    pub async fn run(&self) -> Result<()> {
        if self
            .has_run
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            return Err(std::io::Error::new(
                std::io::ErrorKind::AlreadyExists,
                "server accept loop is already running",
            )
            .into());
        }

        let limiter = Arc::new(Semaphore::new(self.max_connections));
        let metrics_listener = self
            .metrics_listener
            .lock()
            .map_err(|_| std::io::Error::other("metrics listener lock poisoned"))?
            .take();
        let _metrics_task = metrics_listener
            .map(|listener| tokio::spawn(metrics::serve_metrics(listener, self.metrics.clone())))
            .map(AbortOnDrop);

        // In-flight connection tasks, tracked so a shutdown can drain them rather
        // than cut them. Finished tasks are reaped after each accept, so the set
        // stays bounded (by `max_connections` plus whatever finished while idle).
        let mut conns: JoinSet<()> = JoinSet::new();
        // Latched shutdown flag. `wait_for` also observes a signal that arrived
        // before this subscribe, so an early `begin_shutdown` is not missed.
        let mut shutdown_rx = self.shutdown.subscribe();
        // Consecutive `accept()` failures, for the backoff below. Reset on success.
        let mut consecutive_accept_errors: u32 = 0;

        loop {
            // Acquire a slot before accepting so we apply backpressure rather
            // than unboundedly spawning tasks under load; a shutdown signal stops
            // the wait instead.
            let permit = tokio::select! {
                biased;
                _ = shutdown_rx.wait_for(|&stop| stop) => break,
                slot = limiter.clone().acquire_owned() => match slot {
                    Ok(p) => p,
                    Err(_) => {
                        error!("connection limiter closed unexpectedly");
                        break;
                    }
                },
            };

            // Accept the next connection, letting a shutdown signal interrupt the
            // wait so an idle server stops promptly rather than after one more.
            // The accept yields `Some(pair)`, or `None` after an error (setting
            // `pending_backoff` when the error warrants a delay). The backoff
            // sleep happens *after* this `select!` so the shutdown-watch guard it
            // produces is not held across that await (which would make the run
            // future non-`Send`).
            let mut pending_backoff: Option<Duration> = None;
            let accepted = tokio::select! {
                biased;
                _ = shutdown_rx.wait_for(|&stop| stop) => {
                    drop(permit);
                    break;
                }
                accepted = self.listener.accept() => match accepted {
                    Ok(pair) => Some(pair),
                    Err(e) => {
                        // A connection the peer aborted or reset before we
                        // accepted it, or an interrupted syscall, is benign churn
                        // (e.g. a port scan): retry at once, and do not count it
                        // toward backoff or the failure metric.
                        if matches!(
                            e.kind(),
                            ErrorKind::ConnectionAborted
                                | ErrorKind::Interrupted
                                | ErrorKind::ConnectionReset
                        ) {
                            debug!(error = %e, "accept failed (transient); retrying");
                        } else {
                            // Other errors tend to persist (e.g. EMFILE/ENFILE
                            // descriptor exhaustion). Count them, then back off
                            // with a capped exponential delay so a degraded
                            // listener does not spin the loop hot and flood the
                            // log. A successful accept resets the streak.
                            self.metrics.accept_failed();
                            consecutive_accept_errors =
                                consecutive_accept_errors.saturating_add(1);
                            let backoff = accept_backoff(consecutive_accept_errors);
                            warn!(
                                error = %e,
                                consecutive = consecutive_accept_errors,
                                backoff_ms = backoff.as_millis() as u64,
                                "accept failed; backing off"
                            );
                            pending_backoff = Some(backoff);
                        }
                        None
                    }
                },
            };
            let (mut stream, peer) = match accepted {
                // A successful accept clears the backoff streak.
                Some(pair) => {
                    consecutive_accept_errors = 0;
                    pair
                }
                None => {
                    // Release the slot; we are not handling a connection.
                    drop(permit);
                    if let Some(backoff) = pending_backoff {
                        // Let shutdown interrupt the backoff so a degraded server
                        // still stops promptly. `shutdown_rx` is free here (the
                        // accept `select!` above has completed), so reuse it
                        // rather than subscribing a fresh receiver each time.
                        tokio::select! {
                            biased;
                            _ = shutdown_rx.wait_for(|&stop| stop) => break,
                            _ = tokio::time::sleep(backoff) => {}
                        }
                    }
                    continue;
                }
            };

            let local = match stream.local_addr() {
                Ok(a) => a,
                Err(e) => {
                    warn!(error = %e, "could not read local address; dropping connection");
                    // No `accepted_connection()` ran for this socket yet (that
                    // happens after admission below), so do not call
                    // `closed_connection()` here — it would drift the active gauge.
                    drop(permit);
                    continue;
                }
            };

            // Snapshot the live config up front: the PROXY-protocol gate needs
            // the trusted-upstream set and the connection task needs the rest.
            let state = self.state.read().await;
            let config = state.config.clone();
            let users = state.users.clone();
            let command_auth = state.command_auth.clone();
            let dns_resolver = state.dns_resolver.clone();
            drop(state);

            // PROXY-protocol admission gate. The cheap source-IP check is done
            // here; the header read itself is deferred to the task so a silent
            // upstream cannot stall the accept loop. When enabled, only trusted
            // upstreams may connect — a direct client must not reach this
            // listener, or it could forge its advertised source address.
            let expect_proxy = if config.proxy_protocol.is_empty() {
                false
            } else if is_trusted_proxy_upstream(&config.proxy_protocol, peer.ip()) {
                true
            } else {
                warn!(peer = %peer, "rejecting connection: proxyprotocol is enabled and the source is not a trusted upstream");
                drop(permit);
                continue;
            };

            // Disable Nagle's algorithm: proxied interactive traffic benefits
            // from low latency more than from coalescing.
            if let Err(e) = stream.set_nodelay(true) {
                debug!(peer = %peer, error = %e, "failed to set TCP_NODELAY");
            }

            let metrics = self.metrics.clone();
            let tls_listener = self.tls_listener.clone();
            let abuse = self.abuse.clone();
            #[cfg(feature = "plugins")]
            let plugins = self.plugins.clone();
            let handshake_timeout = config.handshake_timeout;
            conns.spawn(async move {
                // Resolve the real client address from a trusted upstream's
                // PROXY header before admitting and handling the connection.
                let peer = if expect_proxy {
                    match tokio::time::timeout(
                        handshake_timeout,
                        crate::proxy_protocol::read_header(&mut stream),
                    )
                    .await
                    {
                        Ok(Ok(Some(real))) => real,
                        // LOCAL / UNSPEC (e.g. a health check): keep the peer.
                        Ok(Ok(None)) => peer,
                        Ok(Err(e)) => {
                            debug!(peer = %peer, error = %e, "invalid PROXY protocol header; dropping");
                            drop(permit);
                            return;
                        }
                        Err(_) => {
                            debug!(peer = %peer, "PROXY protocol header timed out; dropping");
                            drop(permit);
                            return;
                        }
                    }
                } else {
                    peer
                };

                // Per-client abuse admission, keyed on the real client address.
                let client_permit = match abuse.admit(peer.ip()) {
                    Ok(p) => p,
                    Err(reason) => {
                        metrics.rate_limited();
                        warn!(peer = %peer, reason = reason.as_str(), "connection rejected by rate limit");
                        drop(permit);
                        return;
                    }
                };
                metrics.accepted_connection();
                let throttle_bucket = client_permit.throttle_bucket();

                let resources = ConnectionResources {
                    config,
                    users,
                    command_auth,
                    metrics: metrics.clone(),
                    abuse,
                    dns_resolver,
                    throttle_bucket,
                    #[cfg(feature = "plugins")]
                    plugins,
                };
                if let Some(listener) = tls_listener {
                    match tokio::time::timeout(handshake_timeout, listener.accept(stream)).await {
                        Ok(Ok(Some(stream))) => {
                            let conn = Connection::new(
                                ClientStream::Tls(Box::new(stream)),
                                peer,
                                local,
                                resources,
                            );
                            if let Err(e) = conn.handle().await {
                                debug!(peer = %peer, error = %e, "connection ended");
                            }
                        }
                        // An ACME TLS-ALPN-01 challenge was answered during the
                        // handshake; the connection carries no SOCKS traffic.
                        Ok(Ok(None)) => {
                            debug!(peer = %peer, "answered ACME TLS-ALPN-01 challenge");
                        }
                        Ok(Err(e)) => {
                            debug!(peer = %peer, error = %e, "TLS handshake failed");
                        }
                        Err(_) => {
                            debug!(peer = %peer, "TLS handshake timed out");
                        }
                    }
                } else {
                    let conn = Connection::new(ClientStream::Tcp(stream), peer, local, resources);
                    if let Err(e) = conn.handle().await {
                        debug!(peer = %peer, error = %e, "connection ended");
                    }
                }
                metrics.closed_connection();
                drop(client_permit);
                drop(permit); // release the slot when the connection finishes
            });

            // Reap finished connection tasks so the set cannot grow without
            // bound (non-blocking; drains all that are already done).
            reap_finished(&mut conns);
        }

        // Shutdown was signalled: the accept loop has stopped. Reap anything that
        // finished since the last sweep so `active` counts only still-running
        // connections (and we skip the drain entirely if none remain).
        reap_finished(&mut conns);
        // Let the in-flight connections finish on their own, up to a bounded
        // window, then abort whatever is left so the process exits promptly.
        // Dropping `conns` would abort everything immediately, so drain first.
        let active = conns.len();
        if active > 0 {
            let drain_timeout = self.process_config.shutdown_drain_timeout;
            if drain_timeout.is_zero() {
                // `shutdown.draintimeout: 0` cuts in-flight connections at once,
                // skipping the (immediately-elapsing) timeout and its misleading
                // "drain timed out" warning. Warn that live connections are being
                // severed: 0 is the most aggressive setting, not an "unlimited"
                // drain, so an operator who set it expecting that sees the loss.
                warn!(
                    active,
                    "shutdown: draintimeout is 0, cutting in-flight connections immediately without draining"
                );
                conns.shutdown().await;
            } else {
                info!(
                    active,
                    drain_timeout_secs = drain_timeout.as_secs(),
                    "shutdown: draining in-flight connections"
                );
                let drained = tokio::time::timeout(drain_timeout, async {
                    while let Some(outcome) = conns.join_next().await {
                        note_task_outcome(outcome);
                    }
                })
                .await;
                if drained.is_err() {
                    warn!(
                        remaining = conns.len(),
                        "shutdown drain timed out; aborting remaining connections"
                    );
                    conns.shutdown().await;
                } else {
                    info!("shutdown: all in-flight connections drained");
                }
            }
        }

        Ok(())
    }
}

async fn load_users(config: &Config) -> Result<UserDb> {
    let Some(path) = config.userlist.clone() else {
        return Ok(UserDb::new());
    };
    // Read and hash-parse the userlist off the runtime threads so a large file on
    // slow storage cannot stall a worker during startup or reload. The outer `?`
    // surfaces a join failure (only on a panic — `spawn_blocking` is never
    // cancelled) with context; the inner `?` surfaces the load/parse error.
    let db = tokio::task::spawn_blocking(move || -> Result<UserDb> {
        let db = UserDb::load(&path)?;
        info!(users = db.len(), path = %path.display(), "loaded userlist");
        Ok(db)
    })
    .await
    .map_err(|e| {
        crate::errors::Error::Io(std::io::Error::other(format!(
            "userlist load task failed: {e}"
        )))
    })??;
    Ok(db)
}

/// Builds the external auth verifier when `auth.command` is configured. The
/// verified-credential cache lives inside it, so it is rebuilt (and the cache
/// reset) on reload, matching how the userlist cache behaves.
fn build_command_auth(config: &Config) -> Option<Arc<CommandAuth>> {
    let command = config.auth_command.as_ref()?;
    match CommandAuth::new(command) {
        Some(auth) => {
            info!(program = %command[0], "external auth command enabled");
            Some(Arc::new(auth))
        }
        None => None,
    }
}

/// Returns `true` if `peer` falls within one of the trusted PROXY-protocol
/// upstream networks.
///
/// The peer address is canonicalized first so an IPv4 trust CIDR still matches
/// an upstream that arrives IPv4-mapped (`::ffff:a.b.c.d`) on a dual-stack
/// listener — `Cidr::contains` itself treats a mapped address as IPv6 and would
/// otherwise reject a legitimately trusted IPv4 upstream.
fn is_trusted_proxy_upstream(trusted: &[Cidr], peer: IpAddr) -> bool {
    let canonical = peer.to_canonical();
    trusted.iter().any(|cidr| cidr.contains(canonical))
}

/// Logs a connection task that ended abnormally. The connection body itself
/// logs ordinary errors, so a failure surfaced here means the task panicked;
/// `JoinError::is_cancelled` only arises from `JoinSet::shutdown` (the drain's
/// abort path), which joins those tasks itself and never routes them here.
fn note_task_outcome(outcome: std::result::Result<(), tokio::task::JoinError>) {
    if let Err(e) = outcome {
        warn!(error = %e, "connection task terminated abnormally");
    }
}

/// Reaps every connection task that has already finished (non-blocking), keeping
/// the tracking set bounded during normal operation and giving an accurate
/// active count at shutdown. Panicking tasks are logged via [`note_task_outcome`].
fn reap_finished(conns: &mut JoinSet<()>) {
    while let Some(outcome) = conns.try_join_next() {
        note_task_outcome(outcome);
    }
}

/// A valid-but-very-likely-a-mistake configuration combination, surfaced as a
/// startup/reload warning. Kept as a pure enumeration (evaluated by
/// [`config_footguns`]) so the decision logic can be unit-tested without
/// capturing log output, and so callers can warn on the *effective* config.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Footgun {
    NoClientRules,
    NoSocksRules,
    OpenProxyNoAuth,
    AcmeWithProxyProtocol,
}

/// Returns the [`Footgun`]s the given config trips, in warning order.
///
/// This must be evaluated on the config that is actually in effect. On reload
/// that means *after* `preserve_process_config` restores the restart-only fields
/// (`internal`, `tls`) to their live values: a reload that merely *requests* a
/// loopback `internal` must not suppress [`Footgun::OpenProxyNoAuth`] while the
/// public listener is still live until restart.
fn config_footguns(config: &Config) -> Vec<Footgun> {
    let mut footguns = Vec::new();
    if !config.rules.has_scope(Scope::Client) {
        footguns.push(Footgun::NoClientRules);
    }
    if !config.rules.has_scope(Scope::Socks) {
        footguns.push(Footgun::NoSocksRules);
    }
    if config.noauth_on_non_loopback_listener() {
        footguns.push(Footgun::OpenProxyNoAuth);
    }
    // Evaluated here (rather than only at bind) so it also fires on reload, where
    // `proxyprotocol` is live and can be enabled while ACME (restart-only) stays
    // active.
    if matches!(config.tls.as_ref(), Some(crate::config::TlsConfig::Acme(_)))
        && !config.proxy_protocol.is_empty()
    {
        footguns.push(Footgun::AcmeWithProxyProtocol);
    }
    footguns
}

fn warn_config_footguns(config: &Config) {
    for footgun in config_footguns(config) {
        match footgun {
            Footgun::NoClientRules => {
                warn!("no 'client' rules defined — all incoming connections will be denied");
            }
            Footgun::NoSocksRules => {
                warn!("no 'socks' rules defined — all requests will be denied");
            }
            Footgun::OpenProxyNoAuth => {
                warn!(
                    listen = %config.internal,
                    "the SOCKS 'none' (no-authentication) method is offered on a non-loopback listener; combined with permissive 'socks'/'client' rules this is an open proxy — drop 'none' from 'socksmethod' (require 'username'), tighten the rules, or bind 'internal' to loopback"
                );
            }
            Footgun::AcmeWithProxyProtocol => {
                warn!(
                    listen = %config.internal,
                    "ACME and proxyprotocol are both configured: TLS-ALPN-01 validation connections that reach this listener without a trusted PROXY header (for example Let's Encrypt connecting directly) are rejected by the proxy-protocol admission gate, so certificate issuance/renewal fails unless every validation connection is proxied through a trusted PROXY-protocol upstream (TCP passthrough)"
                );
            }
        }
    }
}

fn warn_restart_required_changes(old: &Config, new: &Config) {
    if old.internal != new.internal {
        warn!(
            current = %old.internal,
            requested = %new.internal,
            "internal listener changes require a restart"
        );
    }
    if old.metrics_listen != new.metrics_listen {
        warn!(
            current = ?old.metrics_listen,
            requested = ?new.metrics_listen,
            "metrics listener changes require a restart"
        );
    }
    if old.metrics_allow_public != new.metrics_allow_public {
        warn!(
            current = old.metrics_allow_public,
            requested = new.metrics_allow_public,
            "metrics.allowpublic changes require a restart"
        );
    }
    if old.tls != new.tls {
        warn!("TLS listener changes require a restart");
    }
    if old.max_connections != new.max_connections {
        warn!(
            current = old.max_connections,
            requested = new.max_connections,
            "maxconnections changes require a restart"
        );
    }
    if old.shutdown_drain_timeout != new.shutdown_drain_timeout {
        warn!(
            current_secs = old.shutdown_drain_timeout.as_secs(),
            requested_secs = new.shutdown_drain_timeout.as_secs(),
            "shutdown.draintimeout changes require a restart"
        );
    }
    if old.log_outputs != new.log_outputs
        || old.log_file != new.log_file
        || old.log_format != new.log_format
        || old.log_rotate_size != new.log_rotate_size
        || old.log_rotate_keep != new.log_rotate_keep
    {
        warn!("logging changes require a restart");
    }
}

fn preserve_process_config(config: &mut Config, process_config: &Config) {
    config.internal = process_config.internal;
    config.metrics_listen = process_config.metrics_listen;
    config.metrics_allow_public = process_config.metrics_allow_public;
    config.tls = process_config.tls.clone();
    config.max_connections = process_config.max_connections;
    config.shutdown_drain_timeout = process_config.shutdown_drain_timeout;
    config.log_outputs = process_config.log_outputs.clone();
    config.log_file = process_config.log_file.clone();
    config.log_format = process_config.log_format;
    config.log_rotate_size = process_config.log_rotate_size;
    config.log_rotate_keep = process_config.log_rotate_keep;
}

struct AbortOnDrop<T>(JoinHandle<T>);

impl<T> Drop for AbortOnDrop<T> {
    fn drop(&mut self) {
        self.0.abort();
    }
}

/// Base delay for the accept-failure backoff; doubles per consecutive failure.
const ACCEPT_BACKOFF_BASE: Duration = Duration::from_millis(5);
/// Cap for the accept-failure backoff, so a persistent failure still retries
/// about once a second rather than stalling the listener for longer.
const ACCEPT_BACKOFF_MAX: Duration = Duration::from_secs(1);

/// Capped exponential backoff for a run of `attempt` consecutive accept
/// failures, clamped to `ACCEPT_BACKOFF_MAX`. Callers pass the 1-based streak
/// length; `0` and `1` both yield the base delay.
fn accept_backoff(attempt: u32) -> Duration {
    crate::util::capped_exponential_backoff(attempt, ACCEPT_BACKOFF_BASE, ACCEPT_BACKOFF_MAX)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn accept_backoff_grows_then_caps() {
        // Doubles from the base, then clamps at the max (never zero, never above).
        assert_eq!(accept_backoff(0), ACCEPT_BACKOFF_BASE);
        assert_eq!(accept_backoff(1), ACCEPT_BACKOFF_BASE);
        assert_eq!(accept_backoff(2), ACCEPT_BACKOFF_BASE * 2);
        assert_eq!(accept_backoff(3), ACCEPT_BACKOFF_BASE * 4);
        assert_eq!(accept_backoff(u32::MAX), ACCEPT_BACKOFF_MAX);
        // Monotonic non-decreasing and always within bounds.
        let mut prev = Duration::ZERO;
        for attempt in 0..40 {
            let d = accept_backoff(attempt);
            assert!(d >= prev && d >= ACCEPT_BACKOFF_BASE && d <= ACCEPT_BACKOFF_MAX);
            prev = d;
        }
    }

    #[test]
    fn proxy_trust_matches_ipv4_mapped_upstream() {
        let trusted: Vec<Cidr> = vec!["10.0.0.0/8".parse().unwrap()];

        // A plain IPv4 trusted upstream matches.
        assert!(is_trusted_proxy_upstream(
            &trusted,
            "10.1.2.3".parse().unwrap()
        ));
        // The same upstream arriving IPv4-mapped on a dual-stack listener must
        // also match the IPv4 trust CIDR — the regression this fixes.
        assert!(is_trusted_proxy_upstream(
            &trusted,
            "::ffff:10.1.2.3".parse().unwrap()
        ));

        // An untrusted source is rejected whether it arrives plain or mapped.
        assert!(!is_trusted_proxy_upstream(
            &trusted,
            "203.0.113.7".parse().unwrap()
        ));
        assert!(!is_trusted_proxy_upstream(
            &trusted,
            "::ffff:203.0.113.7".parse().unwrap()
        ));
    }

    #[test]
    fn proxy_trust_matches_native_ipv6_upstream() {
        // A native (non-mapped) IPv6 upstream still matches an IPv6 trust CIDR.
        let trusted: Vec<Cidr> = vec!["2001:db8::/32".parse().unwrap()];
        assert!(is_trusted_proxy_upstream(
            &trusted,
            "2001:db8::1".parse().unwrap()
        ));
        assert!(!is_trusted_proxy_upstream(
            &trusted,
            "2001:dead::1".parse().unwrap()
        ));
    }

    #[tokio::test]
    async fn bind_reports_local_addr() {
        let cfg = Config::parse(
            "internal: 127.0.0.1 port = 0\nsocks pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }",
        )
        .unwrap();
        let server = Server::bind(cfg).await.unwrap();
        let addr = server.local_addr().unwrap();
        assert_eq!(addr.ip().to_string(), "127.0.0.1");
        assert_ne!(addr.port(), 0);
    }

    #[tokio::test]
    async fn bind_refuses_public_metrics_without_allowpublic() {
        // `bind` runs the startup validation, so an unauthenticated metrics
        // endpoint on a non-loopback address is refused before anything binds.
        let cfg = Config::parse(
            "internal: 127.0.0.1 port = 0\nmetrics.listen: 0.0.0.0:0\nsocks pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }",
        )
        .unwrap();
        let Err(err) = Server::bind(cfg).await else {
            panic!("bind should refuse a public metrics endpoint without metrics.allowpublic");
        };
        assert!(err.to_string().contains("metrics.allowpublic"), "{err}");
    }

    #[tokio::test]
    async fn reload_updates_runtime_config_for_new_connections() {
        let server = Server::bind(
            Config::parse(
                "internal: 127.0.0.1 port = 0\nhandshaketimeout: 7\nshutdown.draintimeout: 5\nclient pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }\nsocks pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }",
            )
            .unwrap(),
        )
        .await
        .unwrap();
        let listen = server.local_addr().unwrap();

        server
            .reload(
                Config::parse(
                    "internal: 127.0.0.1 port = 1\nhandshaketimeout: 11\nmaxconnections: 7\nshutdown.draintimeout: 20\nclient pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }\nsocks block { from: 0.0.0.0/0 to: 0.0.0.0/0 }",
                )
                .unwrap(),
            )
            .await
            .unwrap();

        let state = server.state.read().await;
        assert_eq!(
            state.config.handshake_timeout,
            std::time::Duration::from_secs(11)
        );
        assert_eq!(state.config.internal, listen);
        assert_eq!(state.config.max_connections, 1024);
        // Non-reloadable process-level settings keep their startup values rather
        // than the reloaded ones.
        assert_eq!(
            state.config.shutdown_drain_timeout,
            std::time::Duration::from_secs(5)
        );
        assert_eq!(state.config.rules.rules.len(), 2);
    }

    #[tokio::test]
    async fn reload_preserves_metrics_allowpublic() {
        // Start with a public metrics endpoint that was explicitly allowed at
        // bind time.
        let server = Server::bind(
            Config::parse(
                "internal: 127.0.0.1 port = 0\nmetrics.listen: 0.0.0.0:0\nmetrics.allowpublic: true\nclient pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }\nsocks pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }",
            )
            .unwrap(),
        )
        .await
        .unwrap();

        // A reload that turns `metrics.allowpublic` off must NOT drift the
        // runtime state: the flag is restart-only, the listener stays bound, and
        // `state.config` keeps the startup value so it never claims the live
        // public endpoint is disallowed. (Toggling it actually takes effect only
        // on restart, which `warn_restart_required_changes` surfaces.)
        server
            .reload(
                Config::parse(
                    "internal: 127.0.0.1 port = 0\nmetrics.listen: 0.0.0.0:0\nmetrics.allowpublic: false\nclient pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }\nsocks pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }",
                )
                .unwrap(),
            )
            .await
            .unwrap();

        let state = server.state.read().await;
        assert!(
            state.config.metrics_allow_public,
            "metrics.allowpublic must keep its startup value across reload"
        );
    }

    #[test]
    fn config_footguns_flags_open_proxy_and_missing_rules() {
        // Loopback, fully ruled, no TLS: nothing to warn about.
        let clean = Config::parse(
            "internal: 127.0.0.1 port = 1080\nclient pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }\nsocks pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }",
        )
        .unwrap();
        assert!(config_footguns(&clean).is_empty());

        // Public listener offering the default `none` method: an open proxy.
        let open = Config::parse(
            "internal: 0.0.0.0 port = 1080\nclient pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }\nsocks pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }",
        )
        .unwrap();
        assert_eq!(config_footguns(&open), vec![Footgun::OpenProxyNoAuth]);

        // No rule scopes at all: both deny-everything footguns trip.
        let no_rules = Config::parse("internal: 127.0.0.1 port = 1080").unwrap();
        let footguns = config_footguns(&no_rules);
        assert!(footguns.contains(&Footgun::NoClientRules));
        assert!(footguns.contains(&Footgun::NoSocksRules));
        assert!(!footguns.contains(&Footgun::OpenProxyNoAuth));
    }

    #[test]
    fn config_footguns_flags_acme_with_proxyprotocol() {
        // ACME validates via TLS-ALPN-01, which the proxyprotocol admission gate
        // would reject for direct (un-PROXY-headed) Let's Encrypt connections.
        let cfg = Config::parse(
            "internal: 127.0.0.1 port = 1080\ntls.acme.domains: proxy.example.com\ntls.acme.cache: /var/lib/alighieri/acme\nproxyprotocol: 10.0.0.0/8\nclient pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }\nsocks pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }",
        )
        .unwrap();
        assert_eq!(config_footguns(&cfg), vec![Footgun::AcmeWithProxyProtocol]);
    }

    #[tokio::test]
    async fn reload_footgun_warnings_reflect_effective_listener() {
        // Bind a public, no-auth listener — an open proxy that trips the footgun.
        let server = Server::bind(
            Config::parse(
                "internal: 0.0.0.0 port = 0\nclient pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }\nsocks pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }",
            )
            .unwrap(),
        )
        .await
        .unwrap();
        assert!(config_footguns(&server.process_config).contains(&Footgun::OpenProxyNoAuth));

        // A reload that *requests* a loopback `internal` must not make the
        // warning disappear: `internal` is restart-only, so the public listener
        // is still live and the effective config still trips the footgun.
        server
            .reload(
                Config::parse(
                    "internal: 127.0.0.1 port = 0\nclient pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }\nsocks pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }",
                )
                .unwrap(),
            )
            .await
            .unwrap();

        let state = server.state.read().await;
        assert!(
            config_footguns(&state.config).contains(&Footgun::OpenProxyNoAuth),
            "open-proxy footgun must reflect the still-live public listener, not the requested loopback"
        );
    }

    #[tokio::test]
    async fn run_can_only_start_once() {
        let server = Arc::new(
            Server::bind(
                Config::parse(
                    "internal: 127.0.0.1 port = 0\nclient pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }\nsocks pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }",
                )
                .unwrap(),
            )
            .await
            .unwrap(),
        );
        let running = server.clone();
        let handle = tokio::spawn(async move { running.run().await });
        while !server.has_run.load(Ordering::Acquire) {
            tokio::task::yield_now().await;
        }

        let err = server.run().await.unwrap_err();
        assert!(err.to_string().contains("already running"));
        handle.abort();
    }

    #[tokio::test]
    async fn metrics_addr_remains_available_after_run_starts() {
        let server = Arc::new(
            Server::bind(
                Config::parse(
                    "internal: 127.0.0.1 port = 0\nmetrics.listen: 127.0.0.1:0\nclient pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }\nsocks pass { from: 0.0.0.0/0 to: 0.0.0.0/0 }",
                )
                .unwrap(),
            )
            .await
            .unwrap(),
        );
        let before = server.metrics_addr().unwrap().unwrap();
        let running = server.clone();
        let handle = tokio::spawn(async move { running.run().await });
        while !server.has_run.load(Ordering::Acquire) {
            tokio::task::yield_now().await;
        }

        assert_eq!(server.metrics_addr().unwrap(), Some(before));
        handle.abort();
    }
}