exfiltrate 0.3.0

// SPDX-License-Identifier: MIT OR Apache-2.0
//! Native and browser-side server plumbing for command execution and replies.
//!
//! # Concurrency, on native
//!
//! A command used to run inline in the connection's read loop, which meant the
//! socket was not read again until it returned. One slow command therefore
//! stalled every other command on the connection, and a `Cancel` could not even
//! be *received*, let alone acted on.
//!
//! So the loop now does three things and only three things: read a frame,
//! dispatch it, go back to reading. Commands run on their own threads, and every
//! byte that leaves the connection goes through a single writer thread fed by a
//! channel. That writer is what makes concurrent replies safe:
//! `send_socket_frame` deliberately does not touch the socket's blocking mode,
//! "to avoid race with reader thread", so having several threads call it at once
//! would interleave their frames. One writer, and a reply plus its attachments
//! submitted as one indivisible job, preserves that.
//!
//! # Which transport
//!
//! Whatever the address selects. `127.0.0.1:1337` is TCP as before, `unix:/path`
//! is a Unix socket whose access control is the containing directory's `0700`
//! mode, and `fd:3` adopts a socket a parent process already connected. The
//! last two exist because a sandbox that forbids `bind(2)` on a port is common,
//! and answering that with a better error message was never a fix. See
//! [`exfiltrate_internal::transport`].
//!
//! # Concurrency, in a browser
//!
//! The same, by the same means. This section used to say a browser could have
//! no equivalent and that the reasons were structural; that was wrong, and it is
//! worth saying so rather than quietly deleting it.
//!
//! Two obstacles were cited. Both are real. Neither is load-bearing.
//!
//! * *The socket handle is a `JsValue`, so it is neither `Send` nor `Sync`.*
//!   Still true. But the native side does not move its socket either — it moves
//!   bytes to a single writer, and `Vec<u8>` is `Send`. So the browser's socket
//!   worker keeps the handle and performs every write.
//! * *The `onmessage` callback owns the event loop.* Only while commands run
//!   inside it. They no longer do: each gets its own worker, the callback
//!   returns promptly, and that is exactly what makes a `Cancel` receivable.
//!
//! Cancellation across workers is an `AtomicBool`, which works because workers
//! share linear memory. The arrangement lives in the `wasm32` submodule.

// The reconnect state machine is target-independent *so that it can be tested
// natively*, but its only non-test caller is the wasm32 proxy loop. So on a
// native non-test build every item in it is dead, and `scripts/check` runs with
// `-D warnings`.
//
// Allowed rather than `#[cfg(any(target_arch = "wasm32", test))]`: keeping the
// module compiled on the host means a change that breaks it is caught by an
// ordinary `cargo check`, which is most of the value of having split it out.
#[cfg_attr(not(any(target_arch = "wasm32", test)), allow(dead_code))]
mod reconnect;
#[cfg(target_arch = "wasm32")]
mod wasm32;

use crate::commands::COMMANDS;
// `Response` is named only on the native side outside tests: the chunk sink is
// typed on it, and a browser has no chunk sink.
#[cfg_attr(target_arch = "wasm32", allow(unused_imports))]
use exfiltrate_internal::command::{CommandContext, Response};
use exfiltrate_internal::rpc::{CommandInvocation, CommandResponse};
use std::sync::LazyLock;

#[cfg(not(target_arch = "wasm32"))]
use exfiltrate_internal::command::StreamError;
#[cfg(not(target_arch = "wasm32"))]
use exfiltrate_internal::rpc::Chunk;
#[cfg(not(target_arch = "wasm32"))]
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

#[cfg(not(target_arch = "wasm32"))]
use exfiltrate_internal::auth;
#[cfg(not(target_arch = "wasm32"))]
use exfiltrate_internal::rpc::{AuthChallenge, RPC};
#[cfg(not(target_arch = "wasm32"))]
use exfiltrate_internal::transport::{Address, Listener, Stream, Transport};
#[cfg(not(target_arch = "wasm32"))]
use exfiltrate_internal::wire::{
    BACKOFF_DURATION, InFlightMessage, MAX_ATTACHMENTS, send_socket_frame,
};
#[cfg(not(target_arch = "wasm32"))]
use std::collections::HashMap;
#[cfg(not(target_arch = "wasm32"))]
use std::sync::{Arc, Mutex};

/// How many commands one connection may have in flight at once.
///
/// A peer that pipelines faster than the application can execute would otherwise
/// spawn threads without bound. The limit is generous — nobody debugging a
/// program runs sixty-four concurrent commands by accident — and exceeding it is
/// answered with an error rather than silently queued, so the client learns what
/// happened.
#[cfg(not(target_arch = "wasm32"))]
const MAX_CONCURRENT_COMMANDS: usize = 64;

/// The exfiltrate server.
///
/// Listens for connections from the CLI and executes commands.
/// The implementation differs based on the target architecture:
/// *   **Native**: opens a TCP listener on the configured address.
/// *   **WASM**: connects to the proxy via WebSocket.
pub struct Server {}

/// One frame group that must reach the socket without another writer's frames
/// getting between its parts.
#[cfg(not(target_arch = "wasm32"))]
type WriteJob = Vec<Vec<u8>>;

#[cfg(not(target_arch = "wasm32"))]
#[derive(Default)]
struct RunningCommands {
    tokens: HashMap<u32, Arc<AtomicBool>>,
}

/// What one connection has proved about itself so far.
///
/// `challenge` is `Some` exactly when this server requires a credential, so
/// "does this connection need to authenticate" and "what was it challenged
/// with" are one piece of state rather than two that can disagree.
#[cfg(not(target_arch = "wasm32"))]
struct Session {
    challenge: Option<AuthChallenge>,
    authenticated: bool,
}

#[cfg(not(target_arch = "wasm32"))]
impl Session {
    /// Whether this connection may issue commands.
    fn admitted(&self) -> bool {
        self.challenge.is_none() || self.authenticated
    }
}

#[cfg(not(target_arch = "wasm32"))]
fn do_stream(stream: Stream) {
    std::thread::Builder::new()
        .name("exfiltrate::server do_stream".to_string())
        .spawn(move || connection_loop(stream))
        .expect("exfiltrate could not spawn a connection thread");
}

#[cfg(not(target_arch = "wasm32"))]
fn connection_loop(mut stream: Stream) {
    let config = crate::config_snapshot();
    let (writer, write_jobs) = std::sync::mpsc::channel::<WriteJob>();

    let mut write_stream = match stream.try_clone_transport() {
        Ok(clone) => clone,
        Err(error) => {
            crate::diagnostic(&format!("exfiltrate: cannot split the connection: {error}"));
            return;
        }
    };
    // The single writer. Everything else submits jobs to it; nothing else
    // touches the socket's write half.
    let writer_thread = std::thread::Builder::new()
        .name("exfiltrate::server write".to_string())
        .spawn(move || {
            for job in write_jobs {
                for frame in job {
                    if let Err(error) = send_socket_frame(&frame, &mut write_stream) {
                        crate::diagnostic(&format!("exfiltrate: write failed: {error}"));
                        return;
                    }
                }
            }
        });
    if let Err(error) = writer_thread {
        crate::diagnostic(&format!("exfiltrate: cannot spawn the writer: {error}"));
        return;
    }

    let subscriber = crate::events::attach(config.event_queue_capacity);
    let running = Arc::new(Mutex::new(RunningCommands::default()));
    let mut in_flight_message = InFlightMessage::new();
    // A challenge per connection, so a proof captured from one is worth nothing
    // on the next. Failing to produce one closes the connection rather than
    // serving it unauthenticated.
    let mut session = Session {
        challenge: None,
        authenticated: false,
    };
    if auth_token().is_some() {
        match auth::challenge() {
            Ok(challenge) => session.challenge = Some(challenge),
            Err(error) => {
                crate::diagnostic(&format!(
                    "exfiltrate: refusing a connection because a challenge could not be \
                     generated: {error}"
                ));
                return;
            }
        }
    }

    loop {
        // Events are unsolicited, so nothing else will prompt us to send them.
        // Draining here and on every idle turn is what makes `watch` prompt
        // without a second thread per connection.
        if !flush_events(subscriber, &writer) {
            break;
        }

        match in_flight_message.read_stream(&mut stream) {
            Err(error) => {
                if error.kind() != std::io::ErrorKind::UnexpectedEof {
                    crate::diagnostic(&format!("exfiltrate: read failed: {error}"));
                }
                break;
            }
            Ok(exfiltrate_internal::wire::ReadStatus::WouldBlock) => {
                std::thread::sleep(BACKOFF_DURATION);
            }
            Ok(exfiltrate_internal::wire::ReadStatus::Progress) => continue,
            Ok(exfiltrate_internal::wire::ReadStatus::Completed(frame)) => {
                let rpc = match rmp_serde::from_slice::<RPC>(&frame) {
                    Ok(rpc) => rpc,
                    Err(error) => {
                        crate::diagnostic(&format!(
                            "exfiltrate: could not parse a message from the client: {error}. \
                             This usually means the CLI and the linked library are different \
                             versions; run `exfiltrate status` to compare them."
                        ));
                        break;
                    }
                };
                if !dispatch(rpc, subscriber, &writer, &running, &config, &mut session) {
                    break;
                }
            }
            Ok(_) => {
                crate::diagnostic("exfiltrate: unknown read status; ignoring");
            }
        }
    }

    crate::events::detach(subscriber);
    // Dropping the writer closes the channel, which ends the writer thread once
    // it has drained whatever a command finished writing.
    drop(writer);
}

/// Handles one inbound message. Returns false when the connection should end.
#[cfg(not(target_arch = "wasm32"))]
fn dispatch(
    rpc: RPC,
    subscriber: u64,
    writer: &std::sync::mpsc::Sender<WriteJob>,
    running: &Arc<Mutex<RunningCommands>>,
    config: &crate::Config,
    session: &mut Session,
) -> bool {
    // Everything except the handshake itself waits until the peer has proved it
    // is allowed to be here. Answering a `Command` with a refusal rather than
    // silence is what makes a client that never saw the challenge — an older
    // one, say — report something a person can act on.
    if !session.admitted() && !matches!(rpc, RPC::Hello(_) | RPC::AuthProof(_)) {
        if let Some(reply_id) = rpc.reply_id() {
            return send(
                writer,
                RPC::CommandResponse(CommandResponse::new(
                    false,
                    format!(
                        "this application requires a token. Set ${} on both ends, or pass \
                         --token.",
                        auth::TOKEN_ENV
                    )
                    .into(),
                    reply_id,
                )),
            );
        }
        crate::diagnostic("exfiltrate: a peer sent traffic before authenticating; closing");
        return false;
    }

    match rpc {
        RPC::Hello(peer) => {
            if !peer.is_compatible() {
                let local = config.build_info();
                crate::diagnostic(&format!("exfiltrate: {}", peer.skew_message(&local)));
            }
            // The challenge goes first, so a client holding our `Hello` already
            // knows whether one was coming.
            if let Some(challenge) = &session.challenge
                && !session.authenticated
                && !send(writer, RPC::AuthChallenge(challenge.clone()))
            {
                return false;
            }
            send(writer, RPC::Hello(config.build_info()))
        }
        RPC::AuthProof(offered) => {
            let (Some(challenge), Some(token)) = (&session.challenge, auth_token()) else {
                crate::diagnostic("exfiltrate: a peer offered a token where none is required");
                return send(
                    writer,
                    RPC::AuthResult(exfiltrate_internal::rpc::AuthResult {
                        ok: true,
                        message: String::new(),
                    }),
                );
            };
            let outcome = auth::verify(&token, challenge, &offered.proof);
            let accepted = outcome.is_ok();
            session.authenticated = accepted;
            if !accepted {
                crate::diagnostic("exfiltrate: a peer offered a token that was not accepted");
            }
            let answered = send(writer, RPC::AuthResult(auth::result_for(&outcome)));
            // A rejected peer starts over. That is not politeness: reconnecting
            // is what makes each guess cost a fresh challenge and another
            // derivation, on top of the penalty `verify` already applied.
            answered && accepted
        }
        RPC::Command(command) => {
            let reply_id = command.reply_id;
            let token = Arc::new(AtomicBool::new(false));
            {
                let mut state = match running.lock() {
                    Ok(state) => state,
                    Err(_) => return false,
                };
                if state.tokens.len() >= MAX_CONCURRENT_COMMANDS {
                    return send(
                        writer,
                        RPC::CommandResponse(CommandResponse::new(
                            false,
                            format!(
                                "this connection already has {MAX_CONCURRENT_COMMANDS} commands \
                                 in flight; wait for one to finish"
                            )
                            .into(),
                            reply_id,
                        )),
                    );
                }
                state.tokens.insert(reply_id, token.clone());
            }

            let command_writer = writer.clone();
            let worker_running = running.clone();
            let spawned = std::thread::Builder::new()
                .name(format!("exfiltrate::command {}", command.name))
                .spawn(move || {
                    run_command(command, token, &command_writer);
                    if let Ok(mut state) = worker_running.lock() {
                        state.tokens.remove(&reply_id);
                    }
                });
            if let Err(error) = spawned {
                if let Ok(mut state) = running.lock() {
                    state.tokens.remove(&reply_id);
                }
                return send(
                    writer,
                    RPC::CommandResponse(CommandResponse::new(
                        false,
                        format!("could not spawn a thread for this command: {error}").into(),
                        reply_id,
                    )),
                );
            }
            true
        }
        RPC::Cancel(cancel) => {
            let known = running
                .lock()
                .ok()
                .and_then(|state| {
                    state
                        .tokens
                        .get(&cancel.reply_id)
                        .map(|token| token.store(true, Ordering::Relaxed))
                })
                .is_some();
            if !known {
                crate::diagnostic(&format!(
                    "exfiltrate: cancel for reply {} arrived after the command finished",
                    cancel.reply_id
                ));
            }
            true
        }
        RPC::Subscribe(subscription) => {
            if let Err(error) = crate::events::subscribe(subscriber, &subscription.topic) {
                crate::diagnostic(&format!("exfiltrate: subscribe refused: {error}"));
            }
            true
        }
        RPC::Unsubscribe(subscription) => {
            if let Err(error) = crate::events::unsubscribe(subscriber, &subscription.topic) {
                crate::diagnostic(&format!("exfiltrate: unsubscribe refused: {error}"));
            }
            true
        }
        RPC::CommandResponse(_) | RPC::Chunk(_) | RPC::Event(_) => {
            crate::diagnostic(
                "exfiltrate: the client sent a message only a server may send; closing",
            );
            false
        }
        _ => {
            // `RPC` is non-exhaustive: a newer client may know variants this
            // build does not. Ignoring one is correct; closing the connection
            // over it would make every future protocol addition a flag day.
            crate::diagnostic("exfiltrate: ignoring an RPC variant this build does not know");
            true
        }
    }
}

/// What a server listening at `address` must ask its clients for.
///
/// Separated from the bind path so the rule can be tested: the server itself is
/// a process-wide singleton, and a rule that is only reachable through one is a
/// rule nobody checks.
#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, PartialEq, Eq)]
enum Credential {
    /// Nothing. The address already restricts who can reach it.
    None,
    /// A token the application or the environment supplied.
    Configured(String),
    /// A token invented for this run, because the address is reachable from
    /// elsewhere and nothing was configured.
    ///
    /// Nobody has seen this one yet, so it has to be printed.
    Generated(String),
}

#[cfg(not(target_arch = "wasm32"))]
impl Credential {
    fn token(&self) -> Option<String> {
        match self {
            Credential::None => None,
            Credential::Configured(token) | Credential::Generated(token) => Some(token.clone()),
        }
    }
}

/// Decides what `address` has to be served with.
///
/// A configured token is always honoured, even where the address would not have
/// needed one — setting one and having it ignored would be the surprise. Absent
/// one, a loopback address, a Unix socket and an inherited descriptor are all
/// already restricted to this machine and need nothing.
///
/// Anything else gets a token invented here rather than a refusal. Refusing
/// would leave someone who deliberately typed `0.0.0.0:1337` with no debug
/// server and an instruction to go and do more work, which is exactly the
/// outcome [`BindFailure::Warn`](crate::BindFailure::Warn) exists to avoid: a
/// debugging facility should not be the reason something does not work.
#[cfg(not(target_arch = "wasm32"))]
fn credential_for(
    address: &Address,
    configured: Option<String>,
) -> Result<Credential, exfiltrate_internal::auth::AuthError> {
    if let Some(token) = configured {
        return Ok(Credential::Configured(token));
    }
    if address.is_local() {
        return Ok(Credential::None);
    }
    Ok(Credential::Generated(auth::generate_token()?))
}

/// The token this server requires, if it requires one./// The token this server requires, if it requires one.
///
/// Set once by the bind path; see [`credential_for`]. `None` here means either
/// "no server is running" or "this address needs no credential", and both
/// answer the only question a connection asks the same way.
#[cfg(not(target_arch = "wasm32"))]
static ENFORCED_TOKEN: std::sync::OnceLock<Option<String>> = std::sync::OnceLock::new();

/// The token this server requires, if it requires one.
#[cfg(not(target_arch = "wasm32"))]
fn auth_token() -> Option<String> {
    ENFORCED_TOKEN.get().cloned().flatten()
}

/// Executes one command and writes its chunks and final response.
#[cfg(not(target_arch = "wasm32"))]
fn run_command(
    command: CommandInvocation,
    token: Arc<AtomicBool>,
    writer: &std::sync::mpsc::Sender<WriteJob>,
) {
    let reply_id = command.reply_id;
    let seq = Arc::new(AtomicU64::new(0));
    let chunk_writer = writer.clone();
    let chunk_seq = seq.clone();
    let context = CommandContext::new(
        token,
        Arc::new(move |payload: Response| {
            let chunk = RPC::Chunk(Chunk {
                reply_id,
                seq: chunk_seq.fetch_add(1, Ordering::Relaxed),
                payload,
            });
            let bytes = rmp_serde::to_vec(&chunk)
                .map_err(|error| StreamError::Disconnected(error.to_string()))?;
            chunk_writer
                .send(vec![bytes])
                .map_err(|_| StreamError::Disconnected("the client disconnected".to_string()))
        }),
    );

    let mut response = do_command(command, &context);
    if response.response.attachment_count() > MAX_ATTACHMENTS {
        response.success = false;
        response.response =
            format!("response exceeds the {MAX_ATTACHMENTS}-attachment limit").into();
    }
    let attachments = response.response.split_data();
    response.num_attachments = match u32::try_from(attachments.len()) {
        Ok(count) => count,
        Err(_) => {
            crate::diagnostic(&format!(
                "exfiltrate: command {reply_id} had too many attachments"
            ));
            return;
        }
    };

    let mut job = match rmp_serde::to_vec(&RPC::CommandResponse(response)) {
        Ok(bytes) => vec![bytes],
        Err(error) => {
            crate::diagnostic(&format!(
                "exfiltrate: cannot encode reply {reply_id}: {error}"
            ));
            return;
        }
    };
    job.extend(attachments);
    // One job, so a concurrent command's reply cannot land between a response
    // and the attachments that belong to it.
    let _ = writer.send(job);
}

#[cfg(not(target_arch = "wasm32"))]
fn send(writer: &std::sync::mpsc::Sender<WriteJob>, rpc: RPC) -> bool {
    match rmp_serde::to_vec(&rpc) {
        Ok(bytes) => writer.send(vec![bytes]).is_ok(),
        Err(error) => {
            crate::diagnostic(&format!("exfiltrate: cannot encode {rpc}: {error}"));
            true
        }
    }
}

/// Writes whatever events have queued for this connection. False means the
/// connection is gone.
#[cfg(not(target_arch = "wasm32"))]
fn flush_events(subscriber: u64, writer: &std::sync::mpsc::Sender<WriteJob>) -> bool {
    for event in crate::events::drain(subscriber) {
        if !send(writer, RPC::Event(event)) {
            return false;
        }
    }
    true
}

/// Looks up and runs a command, isolating a panic in it from the connection.
fn do_command(command: CommandInvocation, context: &CommandContext) -> CommandResponse {
    // Cloned out of the registry lock before executing: a command that
    // registers another command, or that asks for the command list, would
    // otherwise deadlock against a lock this function was still holding.
    let name = command.name.clone();
    let found = COMMANDS.lock_sync_read().contains_key(name.as_str());
    if !found {
        return CommandResponse::new(
            false,
            format!(
                "command not found: {name}. Run `exfiltrate list` to see what this \
                 application actually registered."
            )
            .into(),
            command.reply_id,
        );
    }

    let result = crate::panics::isolate(&name, || {
        let registry = COMMANDS.lock_sync_read();
        let Some(matcher) = registry.get(name.as_str()) else {
            return Err(Response::String(format!(
                "command {name} was unregistered while it was being invoked"
            )));
        };
        matcher.execute_with(command.args, context)
    });

    match result {
        Ok(response) => CommandResponse::new(true, response, command.reply_id),
        Err(response) => CommandResponse::new(false, response, command.reply_id),
    }
}

/// Global singleton server instance.
///
/// Lazily initializes the server on first access (which happens in `exfiltrate::begin_with`).
pub static SERVER: LazyLock<Server> = LazyLock::new(Server::new);

impl Server {
    fn new() -> Server {
        #[cfg(not(target_arch = "wasm32"))]
        {
            Self::new_native()
        }
        #[cfg(target_arch = "wasm32")]
        {
            Self::new_web()
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    fn new_native() -> Server {
        let config = crate::config_snapshot();
        let text = exfiltrate_internal::wire::resolve_addr(config.addr.as_deref());
        let address = match Address::parse(&text) {
            Ok(address) => address,
            Err(error) => {
                report_bind_failure_message(
                    &config,
                    format!("exfiltrate: {text} is not an address exfiltrate understands: {error}"),
                );
                return Server {};
            }
        };
        // The guard that used to be implicit in a hardcoded 127.0.0.1 and
        // stopped being implicit the moment the address became configurable.
        // Refusing here is the whole of it: a bound port is reachable, and a
        // debug command is not read-only.
        let credential =
            match credential_for(&address, auth::resolve_token(config.token.as_deref())) {
                Ok(credential) => credential,
                // The one case that still fails closed. We know this address needs a
                // credential and we cannot make one, so there is nothing to fall
                // back to that would not be worse than not listening.
                Err(error) => {
                    report_bind_failure_message(
                        &config,
                        format!(
                            "exfiltrate: {address} is reachable from outside this machine, so it\
                         \n  needs a token, and one could not be generated: {error}\
                         \n  No debug server was started. Set ${} on both ends, or listen\
                         \n  somewhere only local callers can reach.",
                            auth::TOKEN_ENV
                        ),
                    );
                    return Server {};
                }
            };
        // Decided once, here, rather than re-resolved per connection. A
        // generated token is not in `Config` at all, and even a configured one
        // could drift if the environment changed after the bind — the socket
        // and the credential guarding it have to be the same decision.
        let _ = ENFORCED_TOKEN.set(credential.token());

        let listener = match Listener::bind(&address) {
            Ok(listener) => listener,
            Err(error) => {
                report_bind_failure(&config, &address, &error);
                // Continue without a server. `begin` returning normally with no
                // listener is the whole point of the Warn policy: a debugging
                // facility must not be the reason the program under debug dies.
                return Server {};
            }
        };

        // With port 0 the interesting address is the one the OS chose, not the
        // one we asked for — and nothing can find it unless we say what it is.
        let bound = listener
            .resolved_address()
            .unwrap_or_else(|| address.clone())
            .to_string();
        crate::diagnostic(&format!("exfiltrate: listening on {bound}"));
        match &credential {
            // Generated because nothing was configured, which makes stderr the
            // only place anyone could learn it. Printing is not a leak here; it
            // is the whole delivery mechanism.
            Credential::Generated(token) => crate::diagnostic(&format!(
                "exfiltrate: {bound} is reachable from outside this machine, so it needs\
                 \n  a token. This run's token is:\
                 \n\
                 \n      {token}\
                 \n\
                 \n  Pass it with `exfiltrate --token {token}`, or export\
                 \n  {env}={token} on both ends. It changes every run; set\
                 \n  {env} here to pin one instead, and it will not be printed.\
                 \n  A token authenticates but does not encrypt: debug output crossing a\
                 \n  hostile network still wants a tunnel.",
                env = auth::TOKEN_ENV
            )),
            Credential::Configured(token) if config.announce_token => crate::diagnostic(&format!(
                "exfiltrate: this run's token is {token} — pass it with `exfiltrate --token \
                 {token}` or export {}={token}",
                auth::TOKEN_ENV
            )),
            // Deliberately not printed: you already have it somewhere, and
            // repeating it here would put it in the logs as well.
            Credential::Configured(_) => {
                crate::diagnostic("exfiltrate: connections must present a token")
            }
            Credential::None => {}
        }

        // An inherited descriptor is not an address anyone else can dial — the
        // number only means anything in this process — so advertising it would
        // be telling the registry something untrue.
        if config.instance_registry && !matches!(address, Address::Fd(_)) {
            crate::instances::advertise(&config.build_info().app_name, &bound);
        }

        let spawned = std::thread::Builder::new()
            .name("exfiltrate::listen".to_string())
            .spawn(move || {
                loop {
                    match listener.accept() {
                        Ok(Some(stream)) => do_stream(stream),
                        // An inherited descriptor is one connection, not a
                        // stream of them; there is nothing left to wait for.
                        Ok(None) => break,
                        Err(error) => {
                            // An accept failure is usually a transient resource
                            // limit. Panicking here used to take the debugged
                            // program down for something it did not do.
                            crate::diagnostic(&format!(
                                "exfiltrate: accept failed: {error}; still listening"
                            ));
                            std::thread::sleep(BACKOFF_DURATION);
                        }
                    }
                }
            });
        if let Err(error) = spawned {
            crate::diagnostic(&format!(
                "exfiltrate: could not spawn the listen thread: {error}"
            ));
        }
        Server {}
    }

    #[cfg(target_arch = "wasm32")]
    fn new_web() -> Server {
        wasm32::wasm32_go();
        Server {}
    }
}

/// Reports a failure to bind, honouring the configured policy.
#[cfg(not(target_arch = "wasm32"))]
fn report_bind_failure(config: &crate::Config, address: &Address, error: &std::io::Error) {
    let message = match error.kind() {
        std::io::ErrorKind::AddrInUse => format!(
            "exfiltrate: {address} is already in use, so no debug server was started.\n  \
             Another copy of this program is probably running. Set EXFILTRATE_ADDR, or pass \
             `Config::default().with_addr(\"127.0.0.1:0\")` to take any free port — \
             `exfiltrate instances` will list it."
        ),
        std::io::ErrorKind::PermissionDenied => format!(
            "exfiltrate: permission denied binding {address}, so no debug server was started.\n  \
             You may be running in a sandbox that forbids listening sockets. Try a transport \
             the sandbox does allow: `EXFILTRATE_ADDR=unix:/path/to/socket` needs only a \
             directory this process can write, and `EXFILTRATE_ADDR=fd:3` needs nothing at \
             all beyond a connected socketpair(2) the parent process passed in."
        ),
        _ => {
            format!("exfiltrate: could not bind {address}: {error}; no debug server was started.")
        }
    };
    report_bind_failure_message(config, message);
}

/// The policy half of [`report_bind_failure`], for failures with no `io::Error`.
#[cfg(not(target_arch = "wasm32"))]
fn report_bind_failure_message(config: &crate::Config, message: String) {
    match config.on_bind_failure {
        crate::BindFailure::Warn => crate::diagnostic(&message),
        crate::BindFailure::Silent => {}
        crate::BindFailure::Panic => panic!("{message}"),
    }
}

#[cfg(test)]
#[path = "server_tests.rs"]
mod tests;