liminal-server 0.23.1

//! One outbound exporting link, admitted with this machine's own Lys pass.

use crate::config::types::{MachineConfig, OutboundLinkConfig};
use crate::link::LinkHost;
use crate::link_carrier::{CarrierError, WireCarrier, read_frame, write_frame};
use crate::link_handshake::{LinkHandshake, LinkHandshakeError, LinkTransport, NOISE_OVERHEAD};
use liminal::protocol::{
    Frame, LinkEnd, LinkOffer, LinkOpenOutcome, ProtocolVersion, REVOCATION_CLOSE_VERSION,
};
use liminal_protocol::credential::WireCredential;
use std::net::{TcpStream, ToSocketAddrs};

/// A keyed link and the carrier retained for its subsequent traffic.
#[derive(Debug)]
pub struct DialedLink {
    /// The configured link under the network.
    pub link_id: String,
    /// The placed prefixes this link exports.
    pub exports: Vec<String>,
    /// The finished handshake, bound to the importing peer's machine.
    pub transport: LinkTransport,
    /// The connection and application stream the handshake opened.
    pub carrier: WireCarrier,
}

/// The peer's answer to one exporting open; Held and Refused are final answers.
#[derive(Debug)]
pub enum LinkDialOutcome {
    /// The handshake finished and its keyed link is held by the caller.
    Opened(Box<DialedLink>),
    /// Lys retained the open under an approval draft.
    Held { grant: String, draft: String },
    /// The peer refused the open by name.
    Refused { reason: String },
}

/// Why an outbound open could not reach a peer's link answer.
#[derive(Debug, thiserror::Error)]
pub enum LinkDialError {
    /// The entry cannot be dialed as configured.
    #[error("outbound link configuration refused: {0}")]
    Configuration(String),
    /// A resolver or connection failed without exposing credentials.
    #[error("outbound link I/O failed: {0}")]
    Io(#[from] std::io::Error),
    /// Framing or the carrier's I/O failed.
    #[error(transparent)]
    Carrier(#[from] CarrierError),
    /// The KK peer proof did not finish.
    #[error(transparent)]
    Handshake(#[from] LinkHandshakeError),
    /// The peer refused machine-pass admission at Connect.
    #[error("peer refused Connect ({reason_code}): {reason:?}")]
    ConnectRefused {
        reason_code: u16,
        reason: Option<String>,
    },
    /// A Connect ack selected a version outside the offered range.
    #[error("peer selected an unoffered protocol version")]
    Version,
    /// The peer ended before answering `Connect` or `LinkOpen`.
    #[error("wire ended before {0}")]
    Ended(&'static str),
}

/// Dials one configured entry with this machine's recorded identity and pass.
pub struct LinkDialer<'a> {
    host: &'a LinkHost,
    network: Option<&'a crate::link_open::network::Network>,
    pass: Result<crate::link_pass::MachinePass, crate::link_pass::PassRefusal>,
    topology:
        Option<std::sync::Arc<crate::server::connection::services::LiminalConnectionServices>>,
}

impl std::fmt::Debug for LinkDialer<'_> {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("LinkDialer")
            .field("host", &self.host)
            .field(
                "pass",
                &self.pass.as_ref().map(crate::link_pass::debug::PassLength),
            )
            .field("topology", &self.topology)
            .finish()
    }
}

impl<'a> LinkDialer<'a> {
    /// Uses the host already built from this machine's verified configuration.
    #[must_use]
    pub fn new(host: &'a LinkHost, machine: &MachineConfig) -> Self {
        Self {
            host,
            pass: crate::link_pass::MachinePass::configured(machine),
            topology: None,
            network: None,
        }
    }

    pub(crate) const fn with_network(
        mut self,
        network: &'a crate::link_open::network::Network,
    ) -> Self {
        self.network = Some(network);
        self
    }

    /// Publishes the keyed open before returning it to its carrier owner.
    #[must_use]
    pub fn with_topology(
        mut self,
        services: std::sync::Arc<crate::server::connection::services::LiminalConnectionServices>,
    ) -> Self {
        self.topology = Some(services);
        self
    }

    /// Opens the exporting end with exactly one TCP connection attempt.
    ///
    /// DNS is resolved once and only its first address is attempted. A peer's
    /// Held or Refused answer is returned without reconnecting.
    ///
    /// # Errors
    /// Missing address/key, unreadable pass, admission refusal, malformed wire
    /// answer, I/O failure, or a handshake against an unexpected machine.
    pub fn open(&self, entry: &OutboundLinkConfig) -> Result<LinkDialOutcome, LinkDialError> {
        let address = entry.peer_address.as_deref().ok_or_else(|| {
            LinkDialError::Configuration(format!("link '{}' has no peer_address", entry.link_id))
        })?;
        let mut handshake = self.handshake(entry)?;
        let mut first = vec![0; NOISE_OVERHEAD];
        let length = handshake.write(&mut first)?;
        first.truncate(length);
        let (mut socket, interruption) = self.connect(address)?;
        let stream_id = 1;
        write_frame(
            &mut socket,
            &Frame::LinkOpen {
                flags: 0,
                stream_id,
                offer: LinkOffer {
                    link_id: entry.link_id.clone(),
                    end: LinkEnd::Export,
                    machine: *self.host.local().machine().ed25519_public(),
                    handshake: first,
                },
            },
        )?;
        let outcome = match read_frame(&mut socket)? {
            Some(Frame::LinkOpenAck {
                stream_id: answered_stream,
                outcome,
                ..
            }) if answered_stream == stream_id => outcome,
            None => return Err(LinkDialError::Ended("LinkOpenAck")),
            Some(_) => return Err(CarrierError::UnexpectedFrame.into()),
        };
        socket.set_read_timeout(None)?;
        socket.set_write_timeout(None)?;
        drop(interruption);
        match outcome {
            LinkOpenOutcome::Opened { handshake: answer } => {
                handshake.read(&answer)?;
                let transport = handshake.finish()?;
                let mut carrier = WireCarrier::new(socket, stream_id)?;
                let life = crate::link::LinkLife::new(*transport.peer());
                if let Some(services) = &self.topology {
                    let local = services.console().device().to_owned();
                    let services = std::sync::Arc::clone(services);
                    life.observe(
                        local,
                        std::sync::Arc::new(move |change| {
                            if let Err(error) = services.console().emit(change) {
                                tracing::error!(%error, "dialed link topology failed");
                            }
                        }),
                    )
                    .map_err(LinkDialError::Configuration)?;
                }
                carrier.install_life(life)?;
                Ok(LinkDialOutcome::Opened(Box::new(DialedLink {
                    link_id: entry.link_id.clone(),
                    exports: entry.exports.clone(),
                    transport,
                    carrier,
                })))
            }
            LinkOpenOutcome::Held { grant, draft } => Ok(LinkDialOutcome::Held { grant, draft }),
            LinkOpenOutcome::Refused { reason } => Ok(LinkDialOutcome::Refused { reason }),
        }
    }

    fn handshake(&self, entry: &OutboundLinkConfig) -> Result<LinkHandshake, LinkDialError> {
        if entry.link_id.is_empty()
            || entry
                .link_id
                .chars()
                .any(|c| c.is_whitespace() || c.is_control())
        {
            return Err(LinkDialError::Configuration(
                "link_id is not an identifier".to_owned(),
            ));
        }
        let peer = self
            .host
            .recorded_machine(&entry.peer_holder)
            .ok_or_else(|| {
                LinkDialError::Configuration(format!(
                    "peer_holder '{}' has no recorded machine key",
                    entry.peer_holder
                ))
            })?;
        for subject in &entry.exports {
            crate::auth_lys::SubjectBinding::new(
                "wire".to_owned(),
                subject.clone(),
                subject.clone(),
            )
            .map_err(|error| LinkDialError::Configuration(error.to_string()))?;
        }
        Ok(LinkHandshake::initiator(self.host.local(), peer)?)
    }

    fn connect(
        &self,
        address: &str,
    ) -> Result<(TcpStream, Option<std::sync::Arc<TcpStream>>), LinkDialError> {
        let pass = self
            .pass
            .as_ref()
            .map_err(|error| LinkDialError::Configuration(error.to_string()))?;
        let token = pass
            .for_open()
            .map_err(|error| LinkDialError::Configuration(error.to_string()))?;
        let token = token.trim();
        if token.is_empty() {
            return Err(LinkDialError::Configuration(
                "machine pass is empty".to_owned(),
            ));
        }
        let credential = WireCredential::lys_record(token.as_bytes());
        if credential.len() > crate::server::connection::pre_admission::MAX_CONNECT_AUTH_TOKEN_BYTES
        {
            return Err(LinkDialError::Configuration(
                "machine pass exceeds the Connect bound".to_owned(),
            ));
        }
        let (mut socket, interruption) = if let Some(network) = self.network {
            let (socket, interruption) = network
                .wire_socket(address)
                .map_err(|error| CarrierError::Lifecycle(error.0))?;
            (socket, Some(interruption))
        } else {
            let target = address.to_socket_addrs()?.next().ok_or_else(|| {
                LinkDialError::Configuration(
                    "peer_address resolved to no socket address".to_owned(),
                )
            })?;
            (TcpStream::connect(target)?, None)
        };
        socket.set_nodelay(true)?;
        write_frame(
            &mut socket,
            &Frame::Connect {
                flags: 0,
                min_version: ProtocolVersion::new(1, 0),
                max_version: REVOCATION_CLOSE_VERSION,
                auth_token: credential,
            },
        )?;
        match read_frame(&mut socket)? {
            Some(Frame::ConnectAck {
                selected_version, ..
            }) => {
                if selected_version < ProtocolVersion::new(1, 0)
                    || selected_version > REVOCATION_CLOSE_VERSION
                {
                    return Err(LinkDialError::Version);
                }
            }
            Some(Frame::ConnectError {
                reason_code,
                message,
                ..
            }) => {
                return Err(LinkDialError::ConnectRefused {
                    reason_code,
                    reason: message,
                });
            }
            None => return Err(LinkDialError::Ended("ConnectAck")),
            Some(_) => return Err(CarrierError::UnexpectedFrame.into()),
        }
        Ok((socket, interruption))
    }
}