apeirogond 0.42.0

// SPDX-License-Identifier: GPL-3.0-or-later

//! `Hello`/`Welcome` on control stream 0 (PROTOCOL.md 4).

use a7n_proto::v1::envelope::Body;
use a7n_proto::v1::{Capability, Envelope, HandshakeError, Hello, PeerInfo, Welcome};
use a7n_proto::{FrameKind, PROTOCOL_VERSION, encode_frame, limits};
use quinn::{Connection, RecvStream, SendStream};

use crate::link::{Link, LinkError, Role, peer_machine_id};

/// Major versions must be equal; name the older machine when they differ.
fn check_version(
    local: u32,
    remote: u32,
    local_name: &str,
    remote_name: &str,
) -> Result<(), LinkError> {
    if local == remote {
        return Ok(());
    }
    let (name, version) = if local < remote {
        (local_name, remote)
    } else {
        (remote_name, local)
    };
    Err(LinkError::UpdateRequired {
        name: name.to_owned(),
        version,
    })
}

/// Capabilities both sides list. Values this build does not know (from a
/// newer peer) are not in the intersection.
pub(crate) fn common_capabilities(local: &[i32], remote: &[i32]) -> Vec<Capability> {
    local
        .iter()
        .filter(|value| remote.contains(value))
        .filter_map(|&value| Capability::try_from(value).ok())
        .filter(|&capability| capability != Capability::Unspecified)
        .collect()
}

/// Frames `body` into `buf` (cleared first, so one buffer serves a whole
/// link) and writes it to `send`.
pub(crate) async fn send_envelope(
    send: &mut SendStream,
    buf: &mut Vec<u8>,
    body: Body,
) -> Result<(), LinkError> {
    buf.clear();
    encode_frame(&Envelope { body: Some(body) }, buf);
    send.write_all(buf).await?;
    Ok(())
}

/// Reads one frame of `kind` from `recv`, keeping surplus bytes in `buf`.
///
/// Cancel-safe: `RecvStream::read_chunk` is (quinn `recv_stream.rs`), and
/// bytes reach `buf` only after it completes, so dropping the future loses
/// nothing and the next call resumes from `buf`.
///
/// The frame must finish within `FRAME_READ_TIMEOUT` once its first byte
/// arrived (PROTOCOL.md 2).
pub(crate) async fn read_frame(
    recv: &mut RecvStream,
    buf: &mut Vec<u8>,
    kind: FrameKind,
) -> Result<Envelope, LinkError> {
    let mut envelope: Envelope = crate::frame::read(
        recv,
        buf,
        kind,
        crate::frame::ChunkBound::Fixed(usize::MAX),
        Some(limits::FRAME_READ_TIMEOUT),
    )
    .await?
    .ok_or(LinkError::StreamClosed)?;
    let info = match &mut envelope.body {
        Some(Body::Hello(hello)) => hello.info.as_mut(),
        Some(Body::Welcome(welcome)) => welcome.info.as_mut(),
        _ => None,
    };
    if let Some(info) = info {
        info.discard_invalid_monitors();
    }
    Ok(envelope)
}

fn into_link(
    connection: Connection,
    streams: (SendStream, RecvStream),
    buf: Vec<u8>,
    info: PeerInfo,
    local: &PeerInfo,
    role: Role,
) -> Result<Link, LinkError> {
    Ok(Link {
        peer_id: peer_machine_id(&connection)?,
        capabilities: common_capabilities(&local.capabilities, &info.capabilities),
        connection,
        control_send: streams.0,
        control_recv: streams.1,
        control_buf: buf,
        peer_info: info,
        role,
    })
}

/// Advertise only the hardware address of the interface carrying this link.
pub(crate) fn local_info(
    connection: &Connection,
    endpoint: &quinn::Endpoint,
    local: &PeerInfo,
) -> PeerInfo {
    let source = connection
        .local_ip()
        .or_else(|| {
            endpoint
                .local_addr()
                .ok()
                .map(|addr| addr.ip())
                .filter(|ip| !ip.is_unspecified())
        })
        .or_else(|| a7n_platform::local_addr_for(connection.remote_address()));
    let mut info = local.clone();
    info.hw_addr = source
        .and_then(a7n_platform::hw_addr_for)
        .map_or_else(Vec::new, |mac| mac.to_vec());
    info
}

/// Initiator side: opens stream 0, sends `Hello`, reads `Welcome`.
pub(crate) async fn initiate(connection: Connection, local: &PeerInfo) -> Result<Link, LinkError> {
    let (mut send, mut recv) = connection.open_bi().await?;
    let hello = Hello {
        protocol_version: PROTOCOL_VERSION,
        info: Some(local.clone()),
    };
    send_envelope(&mut send, &mut Vec::new(), Body::Hello(hello)).await?;
    let mut buf = Vec::new();
    let Some(Body::Welcome(welcome)) = read_frame(&mut recv, &mut buf, FrameKind::Control)
        .await?
        .body
    else {
        return Err(LinkError::InvalidHello);
    };
    let info = welcome.info.clone().unwrap_or_default();
    check_version(
        PROTOCOL_VERSION,
        welcome.protocol_version,
        &local.name,
        &info.name,
    )?;
    if welcome.error() != HandshakeError::None {
        return Err(LinkError::InvalidHello);
    }
    into_link(connection, (send, recv), buf, info, local, Role::Initiator)
}

/// Responder side: accepts stream 0, reads `Hello`, answers `Welcome`.
///
/// `accepts_drive` says whether this initiator may drive us.
pub(crate) async fn respond(
    connection: Connection,
    local: &PeerInfo,
    accepts_drive: bool,
) -> Result<Link, LinkError> {
    let (mut send, mut recv) = connection.accept_bi().await?;
    let mut buf = Vec::new();
    let first = read_frame(&mut recv, &mut buf, FrameKind::FirstHello).await?;
    let Some(Body::Hello(hello)) = first.body else {
        return reject(
            &mut send,
            local,
            HandshakeError::InvalidHello,
            LinkError::InvalidHello,
        )
        .await;
    };
    let info = hello.info.clone().unwrap_or_default();
    if let Err(error) = check_version(
        PROTOCOL_VERSION,
        hello.protocol_version,
        &local.name,
        &info.name,
    ) {
        return reject(&mut send, local, HandshakeError::UnsupportedVersion, error).await;
    }
    let welcome = Welcome {
        protocol_version: PROTOCOL_VERSION,
        info: Some(local.clone()),
        accepts_drive,
        error: HandshakeError::None.into(),
    };
    send_envelope(&mut send, &mut Vec::new(), Body::Welcome(welcome)).await?;
    into_link(connection, (send, recv), buf, info, local, Role::Responder)
}

/// Sends a `Welcome` carrying `code`, waits until the peer has it, and
/// returns `error`.
async fn reject(
    send: &mut SendStream,
    local: &PeerInfo,
    code: HandshakeError,
    error: LinkError,
) -> Result<Link, LinkError> {
    let welcome = Welcome {
        protocol_version: PROTOCOL_VERSION,
        info: Some(local.clone()),
        accepts_drive: false,
        error: code.into(),
    };
    send_envelope(send, &mut Vec::new(), Body::Welcome(welcome)).await?;
    // Best effort: without this the close could overtake the Welcome.
    let _ = send.finish();
    let _ = send.stopped().await;
    Err(error)
}

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

    #[test]
    fn check_version_major_mismatch_names_older_machine_to_update() {
        // Arrange
        let cases = [
            (1, 2, Some("update laptop to protocol v2")),
            (2, 1, Some("update desktop to protocol v2")),
            (1, 1, None),
        ];

        for (local, remote, expected) in cases {
            // Act
            let result = check_version(local, remote, "laptop", "desktop");

            // Assert
            assert_eq!(
                result.err().map(|error| error.to_string()).as_deref(),
                expected,
                "{local} vs {remote}"
            );
        }
    }

    #[test]
    fn common_capabilities_unknown_remote_capability_is_dropped() {
        // Arrange
        let unknown = 9999;
        let local = [
            Capability::ClipboardText as i32,
            Capability::Gestures as i32,
        ];
        let remote = [Capability::ClipboardText as i32, unknown];

        // Act
        let common = common_capabilities(&local, &remote);

        // Assert
        assert_eq!(common, vec![Capability::ClipboardText]);
    }
}