wsb 0.258.6441

//! The channel two daemons talk on: authenticated, encrypted, and pinned to
//! the identities that were paired.
//!
//! # The handshake
//!
//! ```text
//! initiator → responder   ephemeral X25519 public key, identity public key
//! responder → initiator   ephemeral X25519 public key, identity public key,
//!                         signature over the transcript
//! initiator → responder   signature over the transcript
//! ```
//!
//! The transcript is both ephemeral keys, both identity keys and a per-
//! connection nonce from each side. Signing it with the long-term identity is
//! what binds the ephemeral keys to the machine: somebody in the middle can
//! substitute their own ephemeral key, but they cannot produce a signature
//! over a transcript containing it, so the substitution is detected before a
//! byte of anything else moves.
//!
//! The shared secret is the X25519 agreement, expanded by HKDF into two keys —
//! one per direction, so a frame can never be replayed back at the sender —
//! and each direction has its own counter, so a frame cannot be replayed
//! within a session either. The per-connection nonces are in the transcript,
//! so a recorded handshake cannot be replayed at all.
//!
//! # Pinning
//!
//! The identity key each side presents is checked against the one that was
//! PAIRED. A peer whose key changed is refused — not re-trusted, not prompted
//! about mid-connection. Re-pairing is a deliberate act a person performs.
//!
//! # Pairing
//!
//! The first connection has nothing pinned yet, so the same handshake runs
//! with one addition: both sides also prove they know the pairing code, by
//! including a MAC of the transcript under the code's derived key. A wrong
//! code fails the handshake having revealed nothing about the right one.

use std::io::{Read, Write};
use std::net::TcpStream;
use std::time::Duration;

use chacha20poly1305::aead::{Aead, KeyInit, Payload};
use chacha20poly1305::{ChaCha20Poly1305, Key, Nonce};
use ed25519_dalek::{Signature, VerifyingKey};
use x25519_dalek::{EphemeralSecret, PublicKey as X25519Public};

use super::identity::MachineIdentity;
use super::model::PeerId;

/// The largest single frame. A frame is one protocol message; the largest is
/// an `Apply` carrying file contents, and this bounds how much a peer can make
/// this daemon allocate before it has agreed to anything.
pub const MAX_FRAME: usize = 64 * 1024 * 1024;

/// How long a handshake may take before it is abandoned.
pub const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(20);

/// How long a single request/response exchange may take.
pub const EXCHANGE_TIMEOUT: Duration = Duration::from_secs(300);

/// Which end of the connection this is. Only the key labels differ, and they
/// must differ or both sides would encrypt with the same key.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Role {
    Initiator,
    Responder,
}

/// What the handshake established.
pub struct Session {
    stream: TcpStream,
    send: ChaCha20Poly1305,
    receive: ChaCha20Poly1305,
    send_counter: u64,
    receive_counter: u64,
    /// Who is on the other end, proven rather than claimed.
    peer: PeerId,
    peer_name: String,
    peer_key: VerifyingKey,
    peer_port: u16,
    /// Where this connection is with, as seen from here.
    peer_address: Option<std::net::SocketAddr>,
}

impl std::fmt::Debug for Session {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Session")
            .field("peer", &self.peer)
            .field("peer_name", &self.peer_name)
            .finish_non_exhaustive()
    }
}

impl Session {
    pub fn peer(&self) -> &PeerId {
        &self.peer
    }

    pub fn peer_name(&self) -> &str {
        &self.peer_name
    }

    pub fn peer_key(&self) -> &VerifyingKey {
        &self.peer_key
    }

    /// Where to dial this peer next time: the address it was reached at, with
    /// the port it says it listens on.
    ///
    /// `None` when the socket could not say who it is talking to, which is not
    /// a state to invent an address for.
    pub fn dialable_address(&self) -> Option<String> {
        self.peer_address
            .map(|address| match address.ip() {
                std::net::IpAddr::V6(ip) => format!("[{ip}]:{}", self.peer_port),
                ip => format!("{ip}:{}", self.peer_port),
            })
    }

    /// Send one message.
    pub fn send<T: serde::Serialize>(&mut self, message: &T) -> Result<(), String> {
        let plaintext =
            serde_json::to_vec(message).map_err(|e| format!("cannot encode a peer message: {e}"))?;
        if plaintext.len() > MAX_FRAME {
            return Err(format!(
                "a peer message of {} bytes is larger than the {MAX_FRAME}-byte limit",
                plaintext.len()
            ));
        }
        let nonce = counter_nonce(self.send_counter);
        self.send_counter = self.send_counter.checked_add(1).ok_or_else(|| {
            "this connection has sent as many frames as it may; reconnect".to_string()
        })?;
        let sealed = self
            .send
            .encrypt(
                &nonce,
                Payload {
                    msg: &plaintext,
                    aad: b"wsb-peer-frame-v1",
                },
            )
            .map_err(|_| "cannot encrypt a peer frame".to_string())?;

        self.stream
            .write_all(&(sealed.len() as u32).to_be_bytes())
            .and_then(|()| self.stream.write_all(&sealed))
            .and_then(|()| self.stream.flush())
            .map_err(|e| format!("cannot write to the peer: {e}"))
    }

    /// Receive one message.
    pub fn receive<T: serde::de::DeserializeOwned>(&mut self) -> Result<T, String> {
        let mut length = [0u8; 4];
        self.stream
            .read_exact(&mut length)
            .map_err(|e| format!("cannot read from the peer: {e}"))?;
        let length = u32::from_be_bytes(length) as usize;
        // Checked BEFORE allocating: the length is the peer's claim, and a
        // claim is not a reason to reserve a gigabyte.
        if length > MAX_FRAME + 64 {
            return Err(format!(
                "the peer announced a {length}-byte frame, over the {MAX_FRAME}-byte limit"
            ));
        }
        let mut sealed = vec![0u8; length];
        self.stream
            .read_exact(&mut sealed)
            .map_err(|e| format!("cannot read from the peer: {e}"))?;

        let nonce = counter_nonce(self.receive_counter);
        self.receive_counter = self.receive_counter.checked_add(1).ok_or_else(|| {
            "this connection has received as many frames as it may; reconnect".to_string()
        })?;
        let plaintext = self
            .receive
            .decrypt(
                &nonce,
                Payload {
                    msg: &sealed,
                    aad: b"wsb-peer-frame-v1",
                },
            )
            // Not "corrupt": a frame that does not authenticate is a frame
            // somebody changed, or one replayed out of order. Either way the
            // connection is over.
            .map_err(|_| "a peer frame did not authenticate; the connection is not trustworthy".to_string())?;

        serde_json::from_slice(&plaintext)
            .map_err(|e| format!("the peer sent a message this wsb does not understand: {e}"))
    }

    /// Send a request and read the answer.
    pub fn exchange<Q: serde::Serialize, A: serde::de::DeserializeOwned>(
        &mut self,
        request: &Q,
    ) -> Result<A, String> {
        self.send(request)?;
        self.receive()
    }

    /// Bound how long a single read or write may block.
    ///
    /// A command line is waiting behind some of these — a secret query, for
    /// one — and a peer that has stopped answering must not hold it there.
    /// Applied to both directions: a write that blocks for ever on a full
    /// buffer is as bad as a read that never returns.
    pub fn set_timeout(&self, limit: std::time::Duration) -> Result<(), String> {
        self.stream
            .set_read_timeout(Some(limit))
            .map_err(|e| format!("cannot bound how long a peer read waits: {e}"))?;
        self.stream
            .set_write_timeout(Some(limit))
            .map_err(|e| format!("cannot bound how long a peer write waits: {e}"))
    }

    pub fn shutdown(&self) {
        let _ = self.stream.shutdown(std::net::Shutdown::Both);
    }
}

/// The socket, and who is on the other end of it.
fn stream_and_port(stream: &mut TcpStream) -> (TcpStream, Option<std::net::SocketAddr>) {
    let address = stream.peer_addr().ok();
    (
        stream
            .try_clone()
            .expect("a connected socket can be duplicated"),
        address,
    )
}

/// The nonce for one frame: the counter, big-endian, in the low 8 bytes.
///
/// Unique per direction because each direction has its own key and its own
/// counter, which is what ChaCha20-Poly1305 requires and what a random nonce
/// per frame would only approximate.
fn counter_nonce(counter: u64) -> Nonce {
    let mut bytes = [0u8; 12];
    bytes[4..].copy_from_slice(&counter.to_be_bytes());
    *Nonce::from_slice(&bytes)
}

/// What the far end must prove.
pub enum Expect<'a> {
    /// A paired peer: this exact identity key, or nothing.
    Pinned(&'a VerifyingKey),
    /// A first introduction: whoever holds this pairing code.
    Pairing(&'a super::identity::PairingCode),
    /// Whoever it is, proven — but not pinned here.
    ///
    /// For the RESPONDER, which cannot know which of its peers is dialling
    /// until it has read the key. The handshake still proves the caller holds
    /// the identity it presents; membership is then checked by the service
    /// against the peers it has, and an unknown one is refused there. This is
    /// not "trust anyone": it is "prove who you are, and I will decide".
    Any,
}

/// Perform the handshake and return the encrypted session.
pub fn handshake(
    mut stream: TcpStream,
    role: Role,
    identity: &MachineIdentity,
    expect: Expect<'_>,
) -> Result<Session, String> {
    handshake_announcing(stream_and_port(&mut stream), role, identity, expect, 0)
}

/// The handshake, announcing the port this machine listens for peers on.
pub fn handshake_announcing(
    (mut stream, peer_address): (TcpStream, Option<std::net::SocketAddr>),
    role: Role,
    identity: &MachineIdentity,
    expect: Expect<'_>,
    listen_port: u16,
) -> Result<Session, String> {
    stream
        .set_read_timeout(Some(HANDSHAKE_TIMEOUT))
        .and_then(|()| stream.set_write_timeout(Some(HANDSHAKE_TIMEOUT)))
        .map_err(|e| format!("cannot set the handshake timeout: {e}"))?;
    // Nagle would hold a small frame waiting for a larger one that never
    // comes, which for a request/response protocol is latency for nothing.
    let _ = stream.set_nodelay(true);

    let ephemeral = EphemeralSecret::random_from_rng(rand::rngs::OsRng);
    let ephemeral_public = X25519Public::from(&ephemeral);

    use rand::RngCore;
    let mut nonce = [0u8; 32];
    rand::rngs::OsRng.fill_bytes(&mut nonce);

    let mine = Greeting {
        version: super::protocol::VERSION,
        ephemeral: super::identity::base64_encode(ephemeral_public.as_bytes()),
        identity: super::identity::base64_encode(identity.public_key().as_bytes()),
        name: identity.name().to_string(),
        nonce: super::identity::base64_encode(&nonce),
        peer_port: listen_port,
    };

    // The initiator speaks first, so both sides read exactly one greeting
    // before either signs anything.
    let theirs: Greeting = match role {
        Role::Initiator => {
            write_plain(&mut stream, &mine)?;
            read_plain(&mut stream)?
        }
        Role::Responder => {
            let theirs = read_plain(&mut stream)?;
            write_plain(&mut stream, &mine)?;
            theirs
        }
    };

    if theirs.version != super::protocol::VERSION {
        return Err(format!(
            "the peer speaks peer-protocol v{} and this wsb speaks v{} — update both",
            theirs.version,
            super::protocol::VERSION
        ));
    }

    let peer_key = VerifyingKey::from_bytes(&theirs.identity_bytes()?)
        .map_err(|e| format!("the peer presented an unusable identity key: {e}"))?;

    // Pinned identity is checked BEFORE the shared secret is used for
    // anything, so a stranger learns nothing beyond "this port speaks the peer
    // protocol".
    if let Expect::Pinned(expected) = expect {
        if expected != &peer_key {
            return Err(format!(
                "the peer's identity changed: expected {}, got {}. \
                 Pairing is not renewed silently — remove the peer and pair again \
                 if this is deliberate.",
                super::identity::Fingerprint::of(expected),
                super::identity::Fingerprint::of(&peer_key)
            ));
        }
    }

    // The transcript both sides sign. Ordered by ROLE, not by arrival, so both
    // ends compute the same bytes.
    let (initiator, responder) = match role {
        Role::Initiator => (&mine, &theirs),
        Role::Responder => (&theirs, &mine),
    };
    let transcript = transcript(initiator, responder);

    let signature = identity.sign(&transcript);
    let pairing_proof = match expect {
        Expect::Pairing(code) => Some(mac(&code.secret(), &transcript)),
        Expect::Pinned(_) | Expect::Any => None,
    };
    let mine_proof = Proof {
        signature: super::identity::base64_encode(&signature.to_bytes()),
        pairing: pairing_proof.map(|proof| super::identity::base64_encode(&proof)),
    };

    let theirs_proof: Proof = match role {
        Role::Responder => {
            write_plain(&mut stream, &mine_proof)?;
            read_plain(&mut stream)?
        }
        Role::Initiator => {
            let theirs = read_plain(&mut stream)?;
            write_plain(&mut stream, &mine_proof)?;
            theirs
        }
    };

    let their_signature =
        Signature::from_bytes(&fixed::<64>(&theirs_proof.signature, "the peer's signature")?);
    if !super::identity::verify(&peer_key, &transcript, &their_signature) {
        return Err(
            "the peer did not prove it holds the identity key it presented".to_string(),
        );
    }

    if let Expect::Pairing(code) = expect {
        let expected = mac(&code.secret(), &transcript);
        let presented = theirs_proof
            .pairing
            .as_deref()
            .ok_or_else(|| "the peer did not present a pairing proof".to_string())?;
        let presented = fixed::<32>(presented, "the peer's pairing proof")?;
        // Constant time: a comparison that returns early tells an attacker how
        // much of a guess was right.
        if !constant_time_eq(&expected, &presented) {
            return Err("the pairing code does not match".to_string());
        }
        let _ = code;
    }

    // Only now is the shared secret used. Both sides derive the same two keys
    // and take opposite ends of them.
    let shared = ephemeral.diffie_hellman(&X25519Public::from(theirs.ephemeral_bytes()?));
    let (initiator_key, responder_key) = derive(shared.as_bytes(), &transcript)?;
    let (send, receive) = match role {
        Role::Initiator => (initiator_key, responder_key),
        Role::Responder => (responder_key, initiator_key),
    };

    stream
        .set_read_timeout(Some(EXCHANGE_TIMEOUT))
        .and_then(|()| stream.set_write_timeout(Some(EXCHANGE_TIMEOUT)))
        .map_err(|e| format!("cannot set the exchange timeout: {e}"))?;

    Ok(Session {
        stream,
        send: ChaCha20Poly1305::new(&send),
        receive: ChaCha20Poly1305::new(&receive),
        send_counter: 0,
        receive_counter: 0,
        peer: super::identity::peer_id_of(&peer_key),
        peer_name: theirs.name,
        peer_key,
        peer_port: theirs.peer_port,
        peer_address,
    })
}

/// The unencrypted opening message. It carries public keys and a nonce and
/// nothing else: everything secret comes after both sides have proven who they
/// are.
/// Base64 on the wire rather than byte arrays: serde has no impl for an array
/// longer than 32, and a handshake that is readable in a packet capture is one
/// a person can diagnose. Nothing here is secret.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct Greeting {
    version: u32,
    ephemeral: String,
    identity: String,
    name: String,
    nonce: String,
    /// The port this machine listens for peers on.
    ///
    /// Announced because only the far end knows it: the port a connection
    /// arrives FROM is an ephemeral one the operating system chose, and
    /// dialling that back would reach nothing. Combined with the address the
    /// connection was seen at — which is the one thing that is known to be
    /// reachable — it is how each side learns where to call the other.
    peer_port: u16,
}

impl Greeting {
    fn ephemeral_bytes(&self) -> Result<[u8; 32], String> {
        fixed(&self.ephemeral, "the peer's ephemeral key")
    }

    fn identity_bytes(&self) -> Result<[u8; 32], String> {
        fixed(&self.identity, "the peer's identity key")
    }
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct Proof {
    signature: String,
    /// Present only while pairing.
    pairing: Option<String>,
}

/// Decode exactly N bytes, or say which field was wrong.
fn fixed<const N: usize>(text: &str, what: &str) -> Result<[u8; N], String> {
    let bytes = super::identity::base64_decode(text).map_err(|e| format!("{what}: {e}"))?;
    bytes
        .try_into()
        .map_err(|_| format!("{what} is not {N} bytes"))
}

/// The bytes both sides sign: everything either of them chose, in a fixed
/// order, with lengths, so no rearrangement produces the same transcript.
fn transcript(initiator: &Greeting, responder: &Greeting) -> Vec<u8> {
    use sha2::{Digest, Sha256};
    let mut hasher = Sha256::new();
    hasher.update(b"wsb-peer-handshake-v1");
    for greeting in [initiator, responder] {
        hasher.update(greeting.version.to_be_bytes());
        // The encoded forms, which are exactly what each side sent: hashing a
        // re-encoding would let two spellings of one key produce two
        // transcripts.
        for field in [&greeting.ephemeral, &greeting.identity, &greeting.nonce, &greeting.name] {
            hasher.update((field.len() as u32).to_be_bytes());
            hasher.update(field.as_bytes());
        }
        // Signed like everything else either side chose: an announced port a
        // man in the middle could rewrite would send the next connection
        // somewhere neither machine picked.
        hasher.update(greeting.peer_port.to_be_bytes());
    }
    hasher.finalize().to_vec()
}

/// Two directional keys from the agreement and the transcript.
fn derive(shared: &[u8], transcript: &[u8]) -> Result<(Key, Key), String> {
    let hkdf = hkdf::Hkdf::<sha2::Sha256>::new(Some(transcript), shared);
    let mut initiator = [0u8; 32];
    let mut responder = [0u8; 32];
    hkdf.expand(b"wsb-peer-initiator-v1", &mut initiator)
        .map_err(|_| "cannot derive the session keys".to_string())?;
    hkdf.expand(b"wsb-peer-responder-v1", &mut responder)
        .map_err(|_| "cannot derive the session keys".to_string())?;
    Ok((
        *Key::from_slice(&initiator),
        *Key::from_slice(&responder),
    ))
}

/// HMAC-SHA256 of the transcript under the pairing code's key.
fn mac(key: &[u8; 32], message: &[u8]) -> [u8; 32] {
    use hmac::{Hmac, Mac};
    let mut mac = <Hmac<sha2::Sha256> as Mac>::new_from_slice(key)
        .expect("HMAC accepts a 32-byte key");
    mac.update(message);
    mac.finalize().into_bytes().into()
}

/// Compare without leaking where the difference is.
fn constant_time_eq(left: &[u8; 32], right: &[u8; 32]) -> bool {
    let mut difference = 0u8;
    for (a, b) in left.iter().zip(right.iter()) {
        difference |= a ^ b;
    }
    difference == 0
}

/// A length-prefixed JSON message, before there is a session to encrypt with.
fn write_plain<T: serde::Serialize>(stream: &mut TcpStream, message: &T) -> Result<(), String> {
    let bytes = serde_json::to_vec(message)
        .map_err(|e| format!("cannot encode a handshake message: {e}"))?;
    stream
        .write_all(&(bytes.len() as u32).to_be_bytes())
        .and_then(|()| stream.write_all(&bytes))
        .and_then(|()| stream.flush())
        .map_err(|e| format!("cannot write the handshake: {e}"))
}

fn read_plain<T: serde::de::DeserializeOwned>(stream: &mut TcpStream) -> Result<T, String> {
    let mut length = [0u8; 4];
    stream
        .read_exact(&mut length)
        .map_err(|e| format!("cannot read the handshake: {e}"))?;
    let length = u32::from_be_bytes(length) as usize;
    // A handshake message is small and fixed in shape. Anything large here is
    // not a handshake, and reserving for it would be the cheapest denial of
    // service there is.
    if length > 8 * 1024 {
        return Err(format!("the peer opened with a {length}-byte handshake message"));
    }
    let mut bytes = vec![0u8; length];
    stream
        .read_exact(&mut bytes)
        .map_err(|e| format!("cannot read the handshake: {e}"))?;
    serde_json::from_slice(&bytes)
        .map_err(|e| format!("the peer sent a handshake this wsb does not understand: {e}"))
}

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