liminal-server 0.23.1

//! Opening and refreshing a direct path retain distinct responsibilities.
use super::{PunchError, PunchedLink, Puncher, UdpCarrier, fresh_nonce, require};
use crate::link_handshake::MachineIdentity;
use serde_json::json;
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
use std::sync::Arc;

impl Puncher {
    /// A new open or recovery gets fresh admitted control streams after an outage.
    /// # Errors
    /// The replacement broker did not admit the device's own subscription.
    pub fn rebind_broker(&mut self, mut broker: super::WireBroker) -> Result<(), PunchError> {
        broker.subscribe(self.local.machine())?;
        self.broker = broker;
        Ok(())
    }
    /// A later open observes once on the retained socket; idle never calls this.
    /// # Errors
    /// The echo or broker refused the new opening's observation.
    pub fn prepare_open(&mut self) -> Result<(), PunchError> {
        if self.observed.is_some() {
            return Ok(());
        }
        let nonce = fresh_nonce()?;
        let mut request = [0; 20];
        request[..4].copy_from_slice(b"LMOB");
        request[4..].copy_from_slice(&nonce);
        let (answer, source) = self
            .carrier()?
            .observe_raw(&request, self.echo, self.bound)?;
        self.observed = Some(observed_endpoint(&nonce, self.echo, source, &answer)?);
        self.register()
    }
    /// Shares the device's one receive worker with its direct paths.
    /// # Errors
    /// The device has not observed its socket.
    pub fn shared_carrier(&self) -> Result<Arc<UdpCarrier>, PunchError> {
        self.carrier
            .clone()
            .ok_or(PunchError::Named("device_not_observed"))
    }

    /// Reads the configured peer's registered wire addresses once.
    /// # Errors
    /// Broker refusal or malformed lookup answer.
    pub fn lookup_addresses(&mut self, peer: MachineIdentity) -> Result<Vec<String>, PunchError> {
        self.validate_peer(peer)?;
        let value = self.broker.call(
            "net.broker.lookup",
            &json!({"device": hex::encode(peer.ed25519_public())}),
        )?;
        serde_json::from_value(
            value
                .get("addresses")
                .cloned()
                .ok_or(PunchError::Named("invalid_broker_answer"))?,
        )
        .map_err(|_| PunchError::Named("invalid_broker_answer"))
    }

    /// Reopens the mapping while retaining the keyed link and all message counters.
    /// # Errors
    /// The broker or peer refused the refreshed mapping.
    pub fn refresh(&mut self, peer: MachineIdentity, index: u32) -> Result<(), PunchError> {
        self.validate_peer(peer)?;
        let lookup = self.broker.call(
            "net.broker.lookup",
            &json!({"device": hex::encode(peer.ed25519_public())}),
        )?;
        let endpoint: SocketAddr = serde_json::from_value(
            lookup
                .get("observed")
                .cloned()
                .ok_or(PunchError::Named("peer_not_observed"))?,
        )
        .map_err(|_| PunchError::Named("peer_not_observed"))?;
        let nonce = punch_nonce(false)?;
        self.probe(endpoint)?;
        self.request_punch(peer, &nonce)?;
        let word = self.broker.word(peer, Some(&nonce))?;
        require(word.observed == Some(endpoint), "peer_observation_changed")?;
        self.carrier()?.refresh_endpoint(index, endpoint)?;
        Ok(())
    }

    /// Answers a mapping refresh without a second KK handshake.
    /// # Errors
    /// Invalid broker word, probe failure, or unknown keyed link.
    pub fn accept_refresh(&mut self, peer: MachineIdentity, index: u32) -> Result<(), PunchError> {
        self.validate_peer(peer)?;
        let word = self.broker.word(peer, None)?;
        let endpoint = word
            .observed
            .ok_or(PunchError::Named("peer_not_observed"))?;
        self.probe(endpoint)?;
        self.request_punch(peer, &word.nonce)?;
        self.carrier()?.refresh_endpoint(index, endpoint)?;
        Ok(())
    }
    /// Opens the NAT mapping, requests the peer, then follows its return word.
    /// # Errors
    /// Refuses unavailable observations, broker failures, invalid words or KK proof.
    pub fn punch(&mut self, peer: MachineIdentity, index: u32) -> Result<PunchedLink, PunchError> {
        self.validate_peer(peer)?;
        require(self.observed.take().is_some(), "device_not_observed")?;
        let nonce = punch_nonce(true)?;
        let lookup = self.broker.call(
            "net.broker.lookup",
            &json!({"device": hex::encode(peer.ed25519_public())}),
        )?;
        let endpoint: SocketAddr = serde_json::from_value(
            lookup
                .get("observed")
                .cloned()
                .ok_or(PunchError::Named("peer_not_observed"))?,
        )
        .map_err(|_| PunchError::Named("peer_not_observed"))?;
        self.probe(endpoint)?;
        self.request_punch(peer, &nonce)?;
        let word = self.broker.word(peer, Some(&nonce))?;
        require(word.observed == Some(endpoint), "peer_observation_changed")?;
        self.handshake(peer, endpoint, index)
    }

    /// Answers the broker's word after sending a mapping-opening probe.
    /// # Errors
    /// Refuses unexpected words, broker failures or an unauthenticated KK peer.
    pub fn accept(&mut self, peer: MachineIdentity, index: u32) -> Result<PunchedLink, PunchError> {
        self.validate_peer(peer)?;
        require(self.observed.take().is_some(), "device_not_observed")?;
        let word = self.broker.word(peer, None)?;
        let endpoint = word
            .observed
            .ok_or(PunchError::Named("peer_not_observed"))?;
        self.probe(endpoint)?;
        // The return word proves both mapping-opening sends preceded the handshake.
        self.request_punch(peer, &word.nonce)?;
        self.handshake(peer, endpoint, index)
    }

    /// The shared carrier holding completed direct links.
    /// # Errors
    /// Refuses use before a successful observation.
    pub fn carrier(&self) -> Result<&UdpCarrier, PunchError> {
        self.carrier
            .as_deref()
            .ok_or(PunchError::Named("device_not_observed"))
    }
}

pub(super) fn observed_endpoint(
    nonce: &[u8; 16],
    expected: SocketAddr,
    source: SocketAddr,
    answer: &[u8],
) -> Result<SocketAddr, PunchError> {
    let length = answer.len();
    if source != expected || length < 23 || &answer[..4] != b"LMOA" || answer[4..20] != *nonce {
        return Err(PunchError::Named("invalid_broker_echo"));
    }
    let address = match (answer[20], length) {
        (4, 27) => SocketAddr::from((
            Ipv4Addr::new(answer[21], answer[22], answer[23], answer[24]),
            u16::from_be_bytes([answer[25], answer[26]]),
        )),
        (6, 39) => SocketAddr::from((
            Ipv6Addr::from(
                <[u8; 16]>::try_from(&answer[21..37])
                    .map_err(|_| PunchError::Named("invalid_broker_echo"))?,
            ),
            u16::from_be_bytes([answer[37], answer[38]]),
        )),
        _ => return Err(PunchError::Named("invalid_broker_echo")),
    };
    require(address.port() != 0, "invalid_broker_echo")?;
    Ok(address)
}

/// A validated broker instruction still requires the recorded peer's KK proof.
#[derive(Debug)]
pub struct Instruction {
    pub peer: MachineIdentity,
    pub endpoint: SocketAddr,
    pub nonce: String,
}
impl Puncher {
    /// Shares the existing device socket while admitting a separate incoming subscription.
    /// # Errors
    /// Subscription admission or socket setup failed.
    pub fn fork_listener(
        &self,
        local: super::LocalMachine,
        mut broker: super::WireBroker,
    ) -> Result<Self, PunchError> {
        broker.subscribe(local.machine())?;
        broker.words.set_read_timeout(None)?;
        Ok(Self {
            local,
            socket: Arc::clone(&self.socket),
            broker,
            echo: self.echo,
            bound: self.bound,
            observed: self.observed,
            carrier: Some(self.shared_carrier()?),
            attempted: true,
            requests: Arc::clone(&self.requests),
        })
    }
    /// Parks on a broker instruction with no idle timer.
    /// # Errors
    /// The broker ended or supplied an invalid instruction.
    pub fn next_instruction(&mut self) -> Result<Instruction, PunchError> {
        loop {
            let Some(liminal::protocol::Frame::Deliver {
                stream_id: 1,
                envelope,
                ..
            }) = crate::link_carrier::read_frame(&mut self.broker.words)?
            else {
                return Err(PunchError::Named("punch_word_missing"));
            };
            require(
                Some(envelope.schema_id) == self.broker.schema,
                "invalid_punch_schema",
            )?;
            let word: super::Word = serde_json::from_slice(&envelope.payload)
                .map_err(|_| PunchError::Named("invalid_punch_word"))?;
            let mut key = [0; 32];
            hex::decode_to_slice(&word.from, &mut key)
                .map_err(|_| PunchError::Named("invalid_punch_peer"))?;
            require(
                hex::decode(&word.nonce).is_ok_and(|nonce| nonce.len() == 16),
                "invalid_punch_nonce",
            )?;
            if self
                .requests
                .lock()
                .map_err(|_| PunchError::Named("punch_request_state_poisoned"))?
                .contains(&word.nonce)
            {
                continue;
            }
            let peer = MachineIdentity::new(key);
            self.validate_peer(peer)?;
            let endpoint = word
                .observed
                .ok_or(PunchError::Named("peer_not_observed"))?;
            require(endpoint.port() != 0, "peer_not_observed")?;
            return Ok(Instruction {
                peer,
                endpoint,
                nonce: word.nonce,
            });
        }
    }
    /// Answers one recorded peer instruction using the existing socket and counters.
    /// # Errors
    /// Broker refusal, probe failure, or the peer's KK proof failed.
    pub fn answer_instruction(
        &mut self,
        word: &Instruction,
        index: u32,
        refresh: bool,
    ) -> Result<(), PunchError> {
        self.probe(word.endpoint)?;
        self.request_punch(word.peer, &word.nonce)?;
        if refresh {
            self.carrier()?.refresh_peer(word.peer, word.endpoint)?;
        } else {
            self.handshake(word.peer, word.endpoint, index)?;
        }
        Ok(())
    }
}

fn punch_nonce(opening: bool) -> Result<String, PunchError> {
    let mut nonce = fresh_nonce()?;
    // The high bit tells the peer whether to keep its keys or start a new handshake.
    if opening {
        nonce[0] |= 128;
    } else {
        nonce[0] &= 127;
    }
    Ok(hex::encode(nonce))
}
impl Instruction {
    pub(crate) fn opening(&self) -> bool {
        matches!(
            self.nonce.as_bytes().first(),
            Some(b'8'..=b'9' | b'a'..=b'f' | b'A'..=b'F')
        )
    }
}