dig_nat/
tunnel.rs

1//! Byte-stream adapter over a relay tunnel — turns the message-oriented [`RelayTunnel`] (RLY-002
2//! send/recv of opaque payloads) into a tokio [`AsyncRead`] + [`AsyncWrite`] duplex stream, so the
3//! SAME mTLS handshake + [`yamux`](crate::mux) multiplexing that runs over a direct TCP connection
4//! runs UNCHANGED over the tier-6 relayed path.
5//!
6//! ## Why this exists (the security crux)
7//!
8//! The relayed tier is the last resort, but it MUST NOT be a weaker connection than a direct one. By
9//! carrying the identical [`dig_tls`] mTLS session over this adapter, a relayed [`PeerConnection`]
10//! presents the same CA-chained [`NodeCert`](dig_tls::NodeCert), the same `peer_id =
11//! SHA-256(SPKI DER)` pin, and the same #1204 BLS binding as a direct connection. The relay only ever
12//! forwards TLS records it cannot read — it is an untrusted byte forwarder, never a trusted party
13//! (§5.4 recipient-sealing still applies as a layer ABOVE this transport). The mTLS layer, not the
14//! relay, authenticates the peer.
15//!
16//! ## Framing
17//!
18//! A [`RelayTunnel`] forwards whole payloads A→relay→B in order. TLS is itself a length-delimited
19//! record protocol, so this adapter needs no framing of its own: each `poll_write` ships up to
20//! [`MAX_RELAY_PAYLOAD`](crate::relay::MAX_RELAY_PAYLOAD) bytes as one RLY-002 frame, and each
21//! `poll_read` drains one inbound payload at a time (buffering the remainder across reads). Payload
22//! order is preserved by the single relay socket; a dropped frame under backpressure surfaces as a
23//! stream error, which fails the dial (acceptable for a last-resort tier — the strategy has already
24//! exhausted every more-direct method).
25
26use std::io;
27use std::pin::Pin;
28use std::task::{Context, Poll};
29
30use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
31
32use crate::relay::{RelayTunnel, MAX_RELAY_PAYLOAD};
33
34/// A duplex byte stream over one [`RelayTunnel`]: writes become RLY-002 relayed frames to the peer,
35/// reads deliver the payloads the relay forwards back. Wrap it exactly like a [`tokio::net::TcpStream`]
36/// — the mTLS connector/acceptor and yamux run over it identically.
37pub struct RelayTunnelStream {
38    /// The underlying relay tunnel (RLY-002 send/recv over the persistent reservation socket).
39    tunnel: RelayTunnel,
40    /// Bytes from the most recently received payload not yet copied to a reader, and the read cursor
41    /// into them. A single inbound payload can satisfy several small `poll_read`s (e.g. TLS reads the
42    /// 5-byte record header then the body), so the remainder is held here rather than dropped.
43    read_carry: Vec<u8>,
44    read_pos: usize,
45}
46
47impl RelayTunnelStream {
48    /// Adapt a [`RelayTunnel`] into a byte-stream. The caller then runs the mTLS handshake over it.
49    pub fn new(tunnel: RelayTunnel) -> Self {
50        RelayTunnelStream {
51            tunnel,
52            read_carry: Vec::new(),
53            read_pos: 0,
54        }
55    }
56}
57
58impl AsyncRead for RelayTunnelStream {
59    fn poll_read(
60        self: Pin<&mut Self>,
61        cx: &mut Context<'_>,
62        buf: &mut ReadBuf<'_>,
63    ) -> Poll<io::Result<()>> {
64        let this = self.get_mut();
65
66        // Serve from the carry-over buffer first — one relay payload may fill many small reads.
67        if this.read_pos >= this.read_carry.len() {
68            match this.tunnel.poll_recv(cx) {
69                Poll::Ready(Some(payload)) => {
70                    this.read_carry = payload;
71                    this.read_pos = 0;
72                }
73                // The reservation dropped: signal clean EOF so the mTLS layer reports a closed
74                // connection rather than hanging.
75                Poll::Ready(None) => return Poll::Ready(Ok(())),
76                Poll::Pending => return Poll::Pending,
77            }
78        }
79
80        let remaining = &this.read_carry[this.read_pos..];
81        let n = remaining.len().min(buf.remaining());
82        buf.put_slice(&remaining[..n]);
83        this.read_pos += n;
84        Poll::Ready(Ok(()))
85    }
86}
87
88impl AsyncWrite for RelayTunnelStream {
89    fn poll_write(
90        self: Pin<&mut Self>,
91        _cx: &mut Context<'_>,
92        buf: &[u8],
93    ) -> Poll<io::Result<usize>> {
94        // Ship at most one relay frame per call; the caller loops for a larger buffer. TLS records are
95        // far smaller than the cap, so in practice one write == one record == one frame.
96        let n = buf.len().min(MAX_RELAY_PAYLOAD);
97        match self.tunnel.send(buf[..n].to_vec()) {
98            Ok(()) => Poll::Ready(Ok(n)),
99            Err(e) => Poll::Ready(Err(io::Error::new(io::ErrorKind::BrokenPipe, e))),
100        }
101    }
102
103    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
104        // `send` hands the frame straight to the reservation's outbound sink — nothing is buffered here.
105        Poll::Ready(Ok(()))
106    }
107
108    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
109        // Dropping the stream drops the tunnel, which deregisters its routing; there is no half-close
110        // to signal over RLY-002, so shutdown is a no-op success.
111        Poll::Ready(Ok(()))
112    }
113}
dig_nat/tunnel.rs

dig_nat/
tunnel.rs
