breakmancer 0.9.0

//! Controller side of the protocol (run on the developer's laptop).

use std::{
    io::{self, Write},
    sync::LazyLock,
    time::Duration,
};

use clap::{Parser, ValueEnum};
use indoc::indoc;
use regex::{self, Regex};
use rustyline_async::{Readline, ReadlineEvent};
use tokio::signal;

use crate::{
    cli::ConnectVia,
    protocol::{
        self, AuthVersion, BreakpointId, Command, Connection, ControllerId, ExitBreak, IdString,
        OutputType, SecureMsg, TerminateCmd, XWingKeypair,
    },
    sigint,
    transport::{tcp::TcpConnectParams, ConnectMethod, TransportListener},
    util,
};

#[derive(Clone, Debug)]
enum BinaryLoc {
    /// Assume breakmancer will already be on the PATH.
    SystemPath,

    /// Use a hosted version of breakmancer, downloaded as needed.
    FetchHosted,

    /// Use the provided path.
    Provided(String),
}

impl BinaryLoc {
    /// Parse `--bin` argument.
    fn parse_arg(value: &str) -> Result<BinaryLoc, String> {
        match value {
            "@PATH" => Ok(BinaryLoc::SystemPath),
            "@FETCH" => Ok(BinaryLoc::FetchHosted),
            path => match path.chars().next() {
                None => Err("Path may not be empty")?,
                Some('.' | '/') => Ok(BinaryLoc::Provided(path.to_owned())),
                _ => Err("Path must start with '.' or '/'")?,
            },
        }
    }

    /// Convert to a string suitable for passing to `--bin` (after
    /// shell escaping).
    fn format_arg(&self) -> &str {
        match self {
            BinaryLoc::SystemPath => "@PATH",
            BinaryLoc::FetchHosted => "@FETCH",
            BinaryLoc::Provided(path) => path,
        }
    }
}

/// Start a new session.
///
/// When the controller starts, it will give you a command to run on
/// the breakpoint host. The breakpoint end will try to open a connection
/// to the callback address, at which point a secure handshake
/// will occur and a debugging session can start.
#[derive(Parser, Debug)]
pub struct Cli {
    /// Authentication protocol.
    #[arg(long, value_enum)]
    auth: Option<AuthVersion>,

    /// Transport method to use.
    #[arg(long = "via")]
    transport: Option<ConnectVia>,

    /// Callback address and port when using TCP transport.
    ///
    /// IP address or domain name that the developer's machine can be reached at,
    /// and the public-facing port to use. This must not be
    /// obstructed by NAT or firewall.
    ///
    /// IPV6 addresses must be enclosed in `[square brackets]`.
    #[arg(long = "tcp-call", value_name = "HOST:PORT")]
    tcp_call: Option<String>,

    /// Local port to listen on when using TCP transport.
    ///
    /// If not specified, defaults to the port specified in
    /// `--tcp-call`.  Set this value if you use port-forwarding and
    /// have differing external and internal ports.
    #[arg(long = "tcp-listen")]
    tcp_listen: Option<u16>,

    #[arg(verbatim_doc_comment)]
    #[doc = "Binary path to use in breakpoint command line.

This only changes what the suggested invocation will be when printing \
the breakmancer command to use on the breakpoint side; you can always \
adjust it manually as needed.

- Use the special string `@PATH` to expect a binary called `breakmancer` \
  on the system path.
- Use the special string `@FETCH` to create a command line that will \
  fetch the binary on demand from a server.
- Use any path starting with `/` or `.` to specify a path to a binary on \
  the breakpoint side, e.g. `/opt/bin/breakmancer` or `./bin/breakmancer`.
"]
    #[arg(long = "bin")]
    binary: Option<String>,
}

/// Session params, processed from CLI args.
struct Params {
    pub connect: ConnectMethod,
    pub binary: BinaryLoc,
}

/// Ask the user if the protocol will work for them.
///
/// Return Err if can't understand answer or no options are acceptable
/// to user.
fn protocol_wizard() -> Result<AuthVersion, String> {
    println!(
        "Will you be able to view stdout from your process as it runs? \
         This is required for finding the breakpoint's verification string during \
         connection setup.\n"
    );
    print!("Answer [y/n]: ");
    io::stdout().flush().unwrap();
    let mut answer = String::new();
    io::stdin().read_line(&mut answer).unwrap();

    println!();
    println!("============================================================");
    println!();

    match answer.trim().to_lowercase().as_str() {
        "y" | "yes" => Ok(AuthVersion::DualPub),
        "n" | "no" => Err("This program currently cannot meet your needs. \
                 If after reviewing your answers you find they are unchanged, \
                 consider submitting a bug report describing your use-case \
                 and its constraints."
            .to_string()),
        _ => Err("Unrecognized answer".to_string()),
    }
}

/// Ask the user how they want to run the breakpoint binary.
///
/// If the answer is a Provided, it will be normalized to start with
/// '/' or '.' if needed.
///
/// Return Err if can't understand answer.
fn binary_source_wizard() -> Result<BinaryLoc, String> {
    println!("How will you be running breakmancer on the breakpoint side?");
    println!();
    println!("1) It's already on the $PATH (e.g. installed system-wide)");
    println!("2) I'd like it to be downloaded temporarily from the maintainer's server");
    println!("3) I've installed the binary in some other location");
    println!();
    print!("Choice [1/2/3]: ");
    io::stdout().flush().unwrap();

    let mut answer = String::new();
    io::stdin().read_line(&mut answer).unwrap();
    println!();

    #[allow(clippy::items_after_statements)]
    fn ask_for_path() -> Result<String, String> {
        println!("Please enter the path on the breakpoint side:\n");
        let mut answer = String::new();
        io::stdin().read_line(&mut answer).unwrap();
        println!();

        let answer = answer.trim_end_matches(['\r', '\n']).to_string();
        if answer.is_empty() {
            Err("Provided path was empty".to_string())
        } else if answer.starts_with(['/', '.']) {
            Ok(answer)
        } else {
            // Put it into the expected format
            Ok(format!("./{answer}"))
        }
    }

    let answer = match answer.trim().to_lowercase().as_str() {
        "1" => Ok(BinaryLoc::SystemPath),
        "2" => Ok(BinaryLoc::FetchHosted),
        "3" => Ok(BinaryLoc::Provided(ask_for_path()?)),
        _ => Err("Unrecognized answer".to_string()),
    };

    println!("============================================================");
    println!();

    answer
}

// Sub-wizard for TCP details.
fn tcp_wizard() -> Result<TcpConnectParams, String> {
    println!("What network address should the breakpoint connect to?");
    println!();
    println!("This must be a HOST:PORT pair. Host may be a domain/host name or IP address.");
    println!();
    print!("> ");
    io::stdout().flush().unwrap();

    let mut callback_addr = String::new();
    io::stdin().read_line(&mut callback_addr).unwrap();
    println!();
    let callback_addr = callback_addr.trim().to_string();
    util::parse_address(&callback_addr)?;

    println!(
        "Should the controller listen on a different local port than the one \
         the breakpoint will connect to? For example, if port-forwarding is in \
         effect."
    );
    println!();
    println!(
        "Either enter a numeric port, or leave this blank to listen on \
              the same port as specified above."
    );
    println!();
    print!("> ");
    io::stdout().flush().unwrap();

    let mut local_port = String::new();
    io::stdin().read_line(&mut local_port).unwrap();
    println!();
    let local_port = local_port.trim();
    let local_port = match local_port {
        "" => None,
        _ => Some(
            local_port
                .parse::<u16>()
                .map_err(|err| format!("Not a valid port number: {err}"))?,
        ),
    };

    Ok(TcpConnectParams {
        callback_addr,
        local_port,
    })
}

/// Ask the user what transport they want to use.
fn transport_wizard() -> Result<ConnectMethod, String> {
    println!(indoc! {
        "How do you want the breakpoint to connect back to this controller?

         1) Magic Wormhole: Both sides connect to `magic-wormhole` rendezvous \
            and transit servers.

            No network configuration is required, and may use a \
            low-latency, direct connection if hole-punching succeeds.

            If a relay server is required (due to firewall or NAT), may \
            be higher latency, and this adds another party that can observe \
            the encrypted traffic. May be less robust if you're using multiple \
            breakpoints. Controller's IP address may be visible in breakpoint \
            logs.

         2) Direct TCP: Controller listens on a TCP port and breakpoint initiates \
            a connection.

            This is the lowest latency and most reliable option -- if you can \
            make it work. Requires more knowledge and control of your network \
            (port-forwarding, manually discovering your IP address, etc.)

            May require exposing controller's IP address to the public (if \
            breakpoint command or logs are public)."
    });
    println!();
    print!("Choice [1/2]: ");
    io::stdout().flush().unwrap();

    let mut protocol_answer = String::new();
    io::stdin().read_line(&mut protocol_answer).unwrap();
    println!();

    let answer = match protocol_answer.trim().to_lowercase().as_str() {
        "1" => ConnectMethod::Wormhole,
        "2" => ConnectMethod::Tcp(tcp_wizard()?),
        _ => Err("Unrecognized answer".to_string())?,
    };

    println!("============================================================");
    println!();

    Ok(answer)
}

/// Format a connection method as a string containing one or more CLI
/// arguments intended for the Controller.
fn controller_transport_args(connect: &ConnectMethod) -> String {
    match &connect {
        ConnectMethod::Tcp(params) => {
            let mut response = format!(
                "--via=tcp --tcp-call={}",
                util::shell_escape(&params.callback_addr)
            );
            if let Some(local_port) = params.local_port {
                response += &format!(" --tcp-listen={local_port}");
            }
            response
        }
        ConnectMethod::Wormhole => String::from("--via=wormhole"),
    }
}

/// Format a connection method as a string containing one or more CLI
/// arguments intended for the Breakpoint.
fn breakpoint_transport_args(connect: &ConnectMethod) -> String {
    match &connect {
        ConnectMethod::Tcp(params) => {
            format!(
                "--via=tcp:1 --tcp-call={}",
                util::shell_escape(&params.callback_addr)
            )
        }
        ConnectMethod::Wormhole => String::from("--via=wormhole:1"),
    }
}

/// Parse CLI args into session params.
///
/// If any necessary args are missing, enter guided mode.
fn parse_args(args: &Cli) -> Result<Params, String> {
    let connect_cli = match args.transport {
        None => None,
        Some(ConnectVia::Tcp) => {
            // TODO: Consider calling tcp_wizard() -- needs a little
            // refactoring first, though.
            let tcp_addr = args.tcp_call.as_ref().ok_or(String::from(
                "Must specify TCP callback parameters with --tcp-call",
            ))?;

            // TODO: Move into TcpConnectParams constructor?
            util::parse_address(tcp_addr)
                .map_err(|err| format!("Invalid callback address: {err}"))?;
            Some(ConnectMethod::Tcp(TcpConnectParams {
                callback_addr: tcp_addr.clone(),
                local_port: args.tcp_listen,
            }))
        }
        Some(ConnectVia::Wormhole) => Some(ConnectMethod::Wormhole),
    };

    let binary_maybe = args
        .binary
        .as_ref()
        .map(|choice| BinaryLoc::parse_arg(choice).map_err(|err| format!("--bin: {err}")))
        .transpose()?;

    // Now we've exhausted all of the things we can parse without
    // asking the user questions about missing params.
    let mut tips = vec![];

    if args.auth.is_none() {
        let chosen = protocol_wizard()?;
        let arg_val = chosen
            .to_possible_value()
            .expect("None of the selectable AuthVersion variants should be skipped.");
        tips.push(format!("--auth={}", arg_val.get_name()));
    }

    let connect = if let Some(connect) = connect_cli {
        connect
    } else {
        let chosen = transport_wizard()?;
        tips.push(controller_transport_args(&chosen));
        chosen
    };

    let binary = if let Some(choice) = binary_maybe {
        choice
    } else {
        let chosen = binary_source_wizard()?;
        tips.push(format!("--bin={}", util::shell_escape(chosen.format_arg())));
        chosen
    };

    if !tips.is_empty() {
        println!(
            "Tip: The following command line options contain your selections, \
             and will allow you to skip those questions in the future:\n\n  {}",
            tips.join(" "),
        );
        println!();
        println!("        ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~        ");
        println!();
    }

    Ok(Params { connect, binary })
}

/// Print bytes as string (lossy Unicode conversion) and add a newline
/// on the end if there wasn't one already.
fn print_with_trailing_newline(data: &[u8]) {
    print!("{}", String::from_utf8_lossy(data));
    if !data.ends_with("\n".as_bytes()) {
        println!();
    }
}

const CMD_EXIT: &str = "exit";
// Use a simple regex to keep the dependency small
static CMD_EXIT_PATTERN: LazyLock<Regex> =
    LazyLock::new(|| Regex::new(r" *(?:exit|Exit|EXIT) *").unwrap());

/// What the user did when we asked them for a command at the REPL.
enum ReplInput {
    /// A command the user wants to have run.
    Command(String),
    /// An interrupt of some sort, possibly signaling a desire to
    /// leave the REPL.
    Interrupted,
}

/// Prompt the user for a command at the REPL.
async fn ask_user_next_command(
    cmd_history: &mut Vec<String>,
    sigint_manager: &mut sigint::Manager,
) -> Result<ReplInput, String> {
    // Set up a new readline each time we need one, because
    // otherwise we can't use regular printlns everywhere. (It has
    // to be dropped in order to get the terminal out of raw mode
    // again.) Maybe
    // change to a persistent readline once we switch to a unified
    // logging system or something. The big downside is having to
    // maintain an external history and re-import it repeatedly.
    let (mut readline, rl_writer) = Readline::new(">> ".to_string())
        .map_err(|err| format!("Could not set up input prompt: {err}"))?;
    for cmd in &mut *cmd_history {
        readline.add_history_entry((*cmd).to_string());
    }

    // Suppress the default ^C handler we have set up.
    let suppress_sigint = sigint_manager.suppress();
    let readline_result = readline.readline().await;
    suppress_sigint.release();

    // Prevent dangling prompt
    readline.flush().unwrap();
    // Keep the writer open this long. Otherwise we get `Couldn't read
    // input: line writers closed` and the session keeps restarting.
    drop(rl_writer);

    match readline_result {
        Ok(ReadlineEvent::Line(command)) => Ok(ReplInput::Command(command)),
        Ok(ReadlineEvent::Eof | ReadlineEvent::Interrupted) => Ok(ReplInput::Interrupted),
        Err(err) => Err(format!("Couldn't read input: {err}")),
    }
}

/// Tell the breakpoint to exit (continue);
///
/// Consumes the connection, as it will stop working immediately anyhow.
async fn hangup(mut conn: Connection) -> Result<(), String> {
    conn.send(SecureMsg::ExitBreak(ExitBreak {}))
        .await
        .map_err(|err| format!("Could not tell breakpoint to exit: {err}"))?;
    Ok(())
}

/// What the user wants to have happen after the session ends.
enum AfterSession {
    /// Breakpoint should exit, but controller should wait for a new connection.
    Iterate,
    /// Both breakpoint and controller should exit.
    Exit,
}

/// Listen for a connection and perform one shell session.
///
/// The `cmd_history` will be updated with new commands run in this
/// session.
///
/// An Ok return means the session ended normally, without an error
/// (that is, the user asked to end the session). The Ok arm indicates
/// what should happen next.
///
/// The connection is consumed; when the function ends, the connection
/// is dropped, which cancels its tasks.
async fn run_session(
    mut conn: Connection,
    cmd_history: &mut Vec<String>,
    sigint_manager: &mut sigint::Manager,
) -> Result<AfterSession, String> {
    println!("Session ready. You can enter single-line commands. Use `{CMD_EXIT}` to exit.");

    // Each iteration of the loop is a command/response interaction on the REPL.
    let mut cmd_seq = 0;
    loop {
        // Try to get a command to run, or get an interrupt/exit
        // request. Convert these to Some vs. None.
        let command: Option<String> =
            match ask_user_next_command(cmd_history, sigint_manager).await? {
                ReplInput::Command(command) => {
                    if command.is_empty() {
                        continue; // just re-show the prompt
                    }

                    if CMD_EXIT_PATTERN.is_match(&command) {
                        // Another way of exiting the session, besides
                        // ^C/^D. We'll treat it the same way as an
                        // interrupt.
                        None
                    } else {
                        Some(command)
                    }
                }
                ReplInput::Interrupted => None,
            };

        // If there was an interrupt/exit, figure out if they meant it
        // (and return) or whether they actually want to stay in the
        // REPL.
        let Some(command) = command else {
            match ask_user_exit_choice().await {
                ExitChoice::Resume => {
                    // User maybe interrupted by accident, and just
                    // wants to go back to entering commands.
                    continue;
                }
                ExitChoice::Iterate => {
                    hangup(conn).await?;
                    return Ok(AfterSession::Iterate);
                }
                ExitChoice::Exit => {
                    hangup(conn).await?;
                    return Ok(AfterSession::Exit);
                }
            }
        };

        // We actually have a command to run! Save it in the history
        // if it's new and not space-prefixed. (ignorespace,
        // ignoredups)
        if !command.starts_with(' ') && Some(&command) != cmd_history.last() {
            cmd_history.push(command.clone());
        }
        conn.send(SecureMsg::Command(Command {
            command: command.clone(),
            seq: cmd_seq,
        }))
        .await
        .map_err(|err| format!("Could not send command to breakpoint: {err}"))?;

        // Once we send the message, we expect zero or more output
        // lines and then an outcome.

        // Whether we're currently trying to kill the remote command
        // (e.g. because it's hanging).
        let mut killing = false;
        loop {
            tokio::select! {
                // Always check the ^C first, to ensure it is handled
                // on a fast stream.
                biased;

                // We turn ^C into a termination command
                _ = signal::ctrl_c() => {
                    if !killing {
                        println!(); // get onto a new line after the ^C
                        println!("Attempting to terminate command...");
                        conn.send(
                            SecureMsg::TerminateCmd(TerminateCmd { seq: cmd_seq })
                        ).await.map_err(|err| format!("Could not send termination command to breakpoint: {err}"))?;
                        killing = true;
                    }
                },

                // But otherwise just stream the output
                msg = conn.recv() => match msg?.ok_or("Error reading from breakpoint.")? {
                    SecureMsg::OutputLine(output) => {
                        if output.seq != cmd_seq {
                            return Err(format!(
                                "Received output for command #{} instead of #{}",
                                output.seq, cmd_seq
                            ));
                        }
                        let log_tag = match output.gender {
                            OutputType::Stdout => "out",
                            OutputType::Stderr => "err",
                        };
                        print!("[{log_tag}] ");
                        print_with_trailing_newline(&output.bytes);
                    }
                    SecureMsg::Finished(outcome) => {
                        if outcome.seq != cmd_seq {
                            return Err(format!(
                                "Received completion notification for command #{} instead of #{}",
                                outcome.seq, cmd_seq
                            ));
                        }
                        match outcome.exit_code {
                            Some(status) => println!("[exit: {status}]"),
                            None => println!("[exit: terminated by signal]"),
                        }
                        break;
                    }
                    wrong_variant => Err(format!(
                        "Received unexpected message type: {}",
                        Into::<&str>::into(wrong_variant)
                    ))?,
                },
            };
        }

        cmd_seq += 1;
    }
}

/// User's confirmation of how they want to exit the REPL.
enum ExitChoice {
    /// They don't actually want to exit, and want to stay in the REPL.
    Resume,
    /// See [`AfterSession::Iterate`].
    Iterate,
    /// See [`AfterSession::Exit`].
    Exit,
}

/// The user has used an interrupt while the REPL was idle -- what do they want to do next?
async fn ask_user_exit_choice() -> ExitChoice {
    println!(indoc! {
        "You've left the command shell. Do you want to resume, or exit?

         [1] Resume: Go back to issuing commands to the breakpoint.
         [2] Iterate: Exit from breakpoint on remote end, running script normally, but leave controller running with same keys.
         [3] Exit: Exit from breakpoint and controller. For additional debugging you'll need to provision the breakpoint with a new key and secret."
    });
    println!();

    let (mut readline, mut rl_writer) = Readline::new("Choice [1/2/3]: ".to_string()).unwrap();
    let choice = loop {
        match readline.readline().await {
            Ok(ReadlineEvent::Line(choice)) => match choice.as_str() {
                "1" => break ExitChoice::Resume,
                "2" => break ExitChoice::Iterate,
                "3" => break ExitChoice::Exit,
                _ => (),
            },
            Ok(ReadlineEvent::Eof) => (),
            Ok(ReadlineEvent::Interrupted) => break ExitChoice::Exit,
            Err(err) => {
                rl_writer
                    .write_all(format!("Failed to read input, exiting: {err}\n").as_bytes())
                    .unwrap();
                break ExitChoice::Exit;
            }
        }
    };
    readline.flush().unwrap(); // prevent dangling prompt
    choice
}

/// Print connection instructions for the user.
fn print_connection_instructions(
    controller_keypair: &XWingKeypair,
    params: &Params,
    listener: &TransportListener,
) {
    let controller_id = ControllerId::from_key(&controller_keypair.public_key);
    let binary_base = match &params.binary {
        BinaryLoc::SystemPath => String::from("breakmancer"),
        BinaryLoc::FetchHosted => String::from(
            "curl -sS https://codeberg.org/timmc/breakmancer/raw/branch/master/fetch-and-run.sh \
             \\\n    | bash -s --",
        ),
        BinaryLoc::Provided(path) => util::shell_escape(path),
    };

    println!(
        indoc! {r#"Run the following command from the environment you want to inspect:

          {} break --auth=dual-pub {} -i {}

        (Now {} for callback...)"#},
        binary_base,
        breakpoint_transport_args(&params.connect),
        &controller_id.id_string(),
        listener.describe_listening(),
    );
}

pub async fn run(args: &Cli) -> Result<(), String> {
    let params = parse_args(args)?;

    // Allow override by debug keys
    let (controller_keypair, debug_breakpoint_id) = match crate::debug_utils::get_debug_keys() {
        Some(keys) => (
            keys.controller_keypair,
            Some(BreakpointId::from_key(&keys.breakpoint_keypair.public_key)),
        ),
        None => (XWingKeypair::new(), None),
    };

    // We need this signal manager for suppressing ^C during certain
    // operations.
    //
    // Except... that's not quite true. We actually *already* suppress
    // ^C at all times, because rustyline-async and tokio's ctrl_c
    // both register a signal handler, and that removes the default
    // action of ^C. So what this manager *actually* does is register
    // yet another handler that exits on ^C, and then provides a way
    // to *specifically* suppress that one handler for certain
    // operations.
    let mut sigint_manager = sigint::Manager::setup();

    let controller_id = ControllerId::from_key(&controller_keypair.public_key);
    let mut transport_listener = params.connect.new_listener(&controller_id).await?;

    print_connection_instructions(&controller_keypair, &params, &transport_listener);

    // Share the readline (and its history) across sessions.
    let mut command_history = Vec::<String>::new();

    loop {
        let backoff = Duration::from_secs(10);
        // New connection created (and destroyed) on every loop
        // iteration. This minimizes lingering state to reduce the
        // opportunity for vulnerabilities or other bugs (including
        // leftover in-flight messages).
        let conn_partial = match protocol::controller_open_connection(
            controller_keypair.clone(),
            &mut transport_listener,
        )
        .await
        {
            Ok(conn) => conn,
            Err(err) => {
                println!(
                    "Failed to start controller connection; waiting {backoff:?} and trying again: {err}"
                );
                tokio::time::sleep(backoff).await;
                continue;
            }
        };

        let expected_breakpoint_id = match debug_breakpoint_id.as_ref() {
            None => {
                print!("Enter the verification string the breakpoint has printed to stdout: ");
                io::stdout().flush().unwrap();
                let mut expected_breakpoint_id = String::new();
                io::stdin()
                    .read_line(&mut expected_breakpoint_id)
                    .map_err(|err| format!("Error while reading input: {err}"))?;
                BreakpointId::from_received_string(expected_breakpoint_id.trim())
                    .map_err(|err| format!("Not a valid verification string: {err}"))?
            }
            Some(debug_value) => {
                println!("[WARNING] Using debug keys for breakpoint.");
                debug_value.clone()
            }
        };
        let (conn, breakpoint_intro) = match conn_partial
            .verify_breakpoint_and_complete_setup(&expected_breakpoint_id)
            .await
        {
            Ok(conn_and_info) => conn_and_info,
            Err(err) => {
                println!("Unable to complete connection: {err}");
                continue;
            }
        };

        if let Some(which) = breakpoint_intro.which {
            println!("Breakpoint: {which}");
        }

        match run_session(conn, &mut command_history, &mut sigint_manager).await {
            Ok(AfterSession::Iterate) => {
                println!("Will wait for new connection...");
            }
            Ok(AfterSession::Exit) => {
                return Ok(());
            }
            Err(err) => {
                println!("Encountered error, restarting connection: {err}");
            }
        }
    }
}