wsb-daemon 0.248.6265

//! The machine's peer service: its identity, its listener, and the peers it
//! trusts.
//!
//! One of these per daemon. It owns the identity every connection
//! authenticates with, the TCP listener peers dial, the pairing window a
//! person opens deliberately, and the store that remembers all of it.

use std::net::{IpAddr, SocketAddr, TcpListener, TcpStream};
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime};

use wsb::peer::identity::{MachineIdentity, PairedPeer, PairingCode, PAIRING_WINDOW};
use wsb::peer::model::{PeerId, PeerState};
use wsb::peer::transport::{self, Expect, Role, Session};

use super::store::MachineStore;

pub use wsb::peer::DEFAULT_PEER_PORT;

/// The pairing window a person opened.
#[derive(Debug, Clone)]
pub struct PairingOffer {
    pub code: PairingCode,
    pub opened_at: SystemTime,
}

impl PairingOffer {
    fn live(&self) -> bool {
        self.opened_at
            .elapsed()
            .map(|since| since < PAIRING_WINDOW)
            .unwrap_or(false)
    }
}

/// Everything peer-related that belongs to the machine.
pub struct PeerService {
    identity: Arc<MachineIdentity>,
    store: Arc<MachineStore>,
    /// The pairing offer currently open, if any. One at a time: two open
    /// windows are two chances for a person to type a code into the wrong one.
    offer: Mutex<Option<PairingOffer>>,
    /// Which peers are connected right now, and what went wrong for the ones
    /// that are not. Observed, never persisted.
    live: Mutex<std::collections::BTreeMap<PeerId, Liveness>>,
    peer_port: u16,
    listen: IpAddr,
}

/// What is true about a peer at this moment.
#[derive(Debug, Clone)]
pub struct Liveness {
    pub state: PeerState,
    pub detail: Option<String>,
    pub since: SystemTime,
}

impl Liveness {
    /// How long this state has held, in seconds.
    ///
    /// The state alone cannot distinguish "connecting" from "has been trying
    /// to connect for four minutes", and the second is a fault while the
    /// first is normal.
    pub fn held_for(&self) -> u64 {
        self.since.elapsed().map(|held| held.as_secs()).unwrap_or(0)
    }
}

impl PeerService {
    pub fn new(
        identity: Arc<MachineIdentity>,
        store: Arc<MachineStore>,
        listen: IpAddr,
        peer_port: u16,
    ) -> Self {
        Self {
            identity,
            store,
            offer: Mutex::new(None),
            live: Mutex::new(std::collections::BTreeMap::new()),
            peer_port,
            listen,
        }
    }

    pub fn peer_port(&self) -> u16 {
        self.peer_port
    }

    /// The address the peer listener is bound to.
    pub fn peer_address(&self) -> IpAddr {
        self.listen
    }

    /// Open a pairing window and hand back the code to read out.
    pub fn open_pairing(&self) -> Result<PairingCode, String> {
        let code = PairingCode::mint();
        let mut offer = self
            .offer
            .lock()
            .map_err(|_| "the pairing window is unusable after a panic".to_string())?;
        *offer = Some(PairingOffer {
            code: code.clone(),
            opened_at: SystemTime::now(),
        });
        Ok(code)
    }

    pub fn close_pairing(&self) {
        if let Ok(mut offer) = self.offer.lock() {
            *offer = None;
        }
    }

    /// The offer, if one is open and still inside its window.
    pub fn pairing_offer(&self) -> Option<PairingOffer> {
        let mut offer = self.offer.lock().ok()?;
        match offer.as_ref() {
            Some(open) if open.live() => Some(open.clone()),
            // An expired window is closed here rather than left to look open.
            Some(_) => {
                *offer = None;
                None
            }
            None => None,
        }
    }

    /// Dial a peer we are being introduced to, and pin it.
    ///
    /// The address is where to knock; the identity that comes back is what is
    /// remembered. A machine that answers at that address and holds the code
    /// becomes a peer — under the id its own key derives to, not under
    /// anything it claims.
    pub fn pair_with(&self, address: &str, code: &PairingCode) -> Result<PairedPeer, String> {
        let mut stream = dial(address)?;
        let session = transport::handshake_announcing(
            (stream.try_clone().map_err(|e| e.to_string())?, stream.peer_addr().ok()),
            Role::Initiator,
            &self.identity,
            Expect::Pairing(code),
            self.peer_port,
        )?;
        // The address the operator typed is the one that works from here, so
        // that is what is remembered — not the one the far end announces about
        // itself, which may be behind a different route.
        let peer = self.remember(&session, address)?;
        session.shutdown();
        let _ = &mut stream;
        Ok(peer)
    }

    /// Record a peer proven by a completed handshake.
    fn remember(&self, session: &Session, address: &str) -> Result<PairedPeer, String> {
        let peer = PairedPeer {
            id: session.peer().clone(),
            name: session.peer_name().to_string(),
            public_key: wsb::peer::identity::base64_encode(session.peer_key().as_bytes()),
            fingerprint: wsb::peer::identity::Fingerprint::of(session.peer_key()),
            address: address.to_string(),
            paired_at: now(),
        };
        self.store.add_peer(&peer)?;
        Ok(peer)
    }

    /// Open a connection to a peer we already trust.
    pub fn connect(&self, peer: &PeerId) -> Result<Session, String> {
        let known = self
            .store
            .peer(peer)?
            .ok_or_else(|| format!("{peer} is not a paired peer"))?;
        let key = known.verifying_key()?;
        self.mark(peer, PeerState::Connecting, None);
        let stream = match dial(&known.address) {
            Ok(stream) => stream,
            Err(error) => {
                // Not reachable is ORDINARY: a laptop is closed, a machine is
                // asleep. Recorded as offline rather than as an error.
                self.mark(peer, PeerState::Offline, Some(error.clone()));
                return Err(error);
            }
        };
        let announced = (
            match stream.try_clone() {
                Ok(clone) => clone,
                Err(error) => {
                    self.mark(peer, PeerState::Error, Some(error.to_string()));
                    return Err(error.to_string());
                }
            },
            stream.peer_addr().ok(),
        );
        match transport::handshake_announcing(
            announced,
            Role::Initiator,
            &self.identity,
            Expect::Pinned(&key),
            self.peer_port,
        ) {
            Ok(session) => {
                bound(&session)?;
                self.mark(peer, PeerState::Connected, None);
                self.store.note_seen(peer, now())?;
                Ok(session)
            }
            Err(error) => {
                // A handshake that fails against a pinned key is not "offline":
                // something answered and was not who it should be.
                self.mark(peer, PeerState::Error, Some(error.clone()));
                Err(error)
            }
        }
    }

    /// Accept one inbound connection, authenticating it against what is known.
    ///
    /// A connection arriving while a pairing window is open is offered the
    /// pairing handshake; otherwise it must be a peer already pinned. Both
    /// paths end with an identity this machine has proven, never claimed.
    pub fn accept(&self, stream: TcpStream) -> Result<Session, String> {
        if let Some(offer) = self.pairing_offer() {
            let announced = (
                stream.try_clone().map_err(|e| e.to_string())?,
                stream.peer_addr().ok(),
            );
            let session = transport::handshake_announcing(
                announced,
                Role::Responder,
                &self.identity,
                Expect::Pairing(&offer.code),
                self.peer_port,
            )?;
            // Where to call it back: the address this connection came from,
            // with the port it announced. The port it dialled FROM is an
            // ephemeral one the operating system chose, and remembering that
            // would leave this machine unable to start a conversation.
            let address = session.dialable_address().ok_or_else(|| {
                "the peer connected from an address this machine cannot name, so there \
                 is nowhere to call it back"
                    .to_string()
            })?;
            self.remember(&session, &address)?;
            self.close_pairing();
            self.mark(session.peer(), PeerState::Connected, None);
            bound(&session)?;
            return Ok(session);
        }

        // No pairing window: the caller must be somebody already trusted. The
        // handshake reads their key first, so this looks them up by what they
        // present and refuses anything unknown.
        let peers = self.store.peers()?;
        // A responder cannot know which peer is dialling before the handshake,
        // so it accepts any pinned key and then checks membership. Expect the
        // key it presents, then verify it is one of ours.
        let announced = (
            stream.try_clone().map_err(|e| e.to_string())?,
            stream.peer_addr().ok(),
        );
        let session = transport::handshake_announcing(
            announced,
            Role::Responder,
            &self.identity,
            Expect::Any,
            self.peer_port,
        )?;
        let presented = wsb::peer::identity::base64_encode(session.peer_key().as_bytes());
        let known = peers
            .iter()
            .find(|peer| peer.public_key == presented && peer.id == *session.peer());
        match known {
            Some(peer) => {
                self.mark(&peer.id, PeerState::Connected, None);
                self.store.note_seen(&peer.id, now())?;
                // Identity is pinned; location is not. A machine that moved to
                // a new address is the same machine, and remembering the old
                // one would leave this side unable to dial it.
                if let Some(address) = session.dialable_address() {
                    if address != peer.address {
                        let mut moved = peer.clone();
                        moved.address = address;
                        let _ = self.store.add_peer(&moved);
                    }
                }
                bound(&session)?;
                Ok(session)
            }
            None => Err(format!(
                "a machine that is not paired with this one connected ({}); \
                 run `wsb peer pair` on both to introduce them",
                wsb::peer::identity::Fingerprint::of(session.peer_key())
            )),
        }
    }

    /// Bind the peer listener.
    pub fn listen(&self) -> Result<TcpListener, String> {
        let address = SocketAddr::new(self.listen, self.peer_port);
        TcpListener::bind(address).map_err(|e| {
            format!(
                "cannot listen for peers on {address}: {e}. Another daemon may hold it, \
                 or this machine may not be able to bind that address."
            )
        })
    }

    pub fn mark(&self, peer: &PeerId, state: PeerState, detail: Option<String>) {
        if let Ok(mut live) = self.live.lock() {
            live.insert(
                peer.clone(),
                Liveness {
                    state,
                    detail,
                    since: SystemTime::now(),
                },
            );
        }
    }

    /// What is true about a peer right now. A peer nothing has been observed
    /// about is offline — not "unknown", which is a state with no action.
    pub fn liveness(&self, peer: &PeerId) -> Liveness {
        self.live
            .lock()
            .ok()
            .and_then(|live| live.get(peer).cloned())
            .unwrap_or(Liveness {
                state: PeerState::Offline,
                detail: None,
                since: SystemTime::UNIX_EPOCH,
            })
    }
}

/// How long one read or one write on a peer connection may block.
///
/// Connecting was bounded and talking was not: a peer that accepted a
/// connection and then never answered held the asking thread for ever — the
/// one reconciliation thread, or the thread a plugin's request was waiting
/// on. Each read and each write gets this long; a transfer that keeps moving
/// is never cut short, because every chunk is a read of its own.
pub const PEER_IO_BOUND: Duration = Duration::from_secs(120);

fn bound(session: &Session) -> Result<(), String> {
    session.set_timeout(PEER_IO_BOUND)
}

/// Connect to `host:port`, with a bound on how long that may take.
fn dial(address: &str) -> Result<TcpStream, String> {
    use std::net::ToSocketAddrs;
    let mut resolved = address
        .to_socket_addrs()
        .map_err(|e| format!("cannot resolve {address}: {e}"))?;
    let target = resolved
        .next()
        .ok_or_else(|| format!("{address} resolves to no address"))?;
    TcpStream::connect_timeout(&target, Duration::from_secs(10))
        .map_err(|e| format!("cannot reach {address}: {e}"))
}

pub fn now() -> u64 {
    SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .map(|since| since.as_secs())
        .unwrap_or(0)
}

/// Serve peers until the daemon stops.
///
/// One thread accepting, one thread per connection. Threads rather than tasks
/// because everything behind a peer request blocks: reading trees, hashing
/// files, writing them. On the runtime's threads that work would starve the
/// workbench, and making it async would mean an async filesystem for no gain.
pub fn spawn_accept_loop(
    machine: std::sync::Arc<crate::machine::MachineState>,
    listener: TcpListener,
) {
    let address = listener.local_addr().ok();
    std::thread::Builder::new()
        .name("wsb-peer-accept".to_string())
        .spawn(move || {
            if let Some(address) = address {
                tracing::info!(%address, "listening for peers");
            }
            for incoming in listener.incoming() {
                let stream = match incoming {
                    Ok(stream) => stream,
                    Err(error) => {
                        tracing::warn!(%error, "a peer connection could not be accepted");
                        continue;
                    }
                };
                let machine = machine.clone();
                // One thread per peer: a connection that stalls must not stop
                // the next one being accepted.
                let spawned = std::thread::Builder::new()
                    .name("wsb-peer-serve".to_string())
                    .spawn(move || serve_one(machine, stream));
                if let Err(error) = spawned {
                    tracing::warn!(%error, "cannot serve a peer connection");
                }
            }
        })
        .expect("the peer accept thread starts");
}

/// Authenticate one inbound connection and serve it until it goes away.
fn serve_one(machine: std::sync::Arc<crate::machine::MachineState>, stream: TcpStream) {
    let peer_address = stream
        .peer_addr()
        .map(|address| address.to_string())
        .unwrap_or_else(|_| "an unknown address".to_string());

    let mut session = match machine.peers.accept(stream) {
        Ok(session) => session,
        Err(error) => {
            // A refused connection is worth saying out loud: it is either a
            // machine that should have been paired, or one that should not be
            // connecting at all.
            tracing::warn!(%error, address = %peer_address, "a peer connection was refused");
            return;
        }
    };
    let peer = session.peer().clone();
    tracing::debug!(%peer, address = %peer_address, "a peer connected");

    let blocking = crate::machine::BlockingMachine {
        state: machine.clone(),
    };
    match crate::peer::connection::serve(&blocking, &mut session, &peer) {
        Ok(()) => {
            // A connection that ended CLEANLY is a conversation that finished,
            // not a machine that went away. Continuity opens one connection per
            // pass and closes it, so marking "offline" here left a healthy peer
            // reading OFFLINE between passes — for all but a few milliseconds
            // of every three seconds, while it was working perfectly.
            //
            // What "offline" means is that a dial failed, and that is where it
            // is recorded.
            machine.peers.mark(&peer, PeerState::Connected, None);
            tracing::debug!(%peer, "a peer connection finished");
        }
        Err(error) => {
            machine
                .peers
                .mark(&peer, PeerState::Error, Some(error.clone()));
            tracing::warn!(%peer, %error, "a peer connection ended badly");
        }
    }
}