prospero_core/caliband/
transport.rs

1//! Network-agnostic transport seam for the caliband protocol (ADR 0051).
2//!
3//! Turns an [`Endpoint`] (+ optional TLS + optional bearer token) into a duplex
4//! byte stream. The NDJSON protocol rides on top of a [`BoxConn`] unchanged —
5//! TLS and the token preamble are framing below it. Ported from
6//! `caliban-supervisor::transport`; prospero needs the client [`connect`] path
7//! in production, and the server [`Listener`] path only for `FakeCaliband`.
8
9use std::sync::Arc;
10
11use serde::{Deserialize, Serialize};
12use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt as _};
13use tokio::net::TcpStream;
14use tokio::net::UnixStream;
15use tokio_rustls::TlsConnector;
16use tokio_rustls::rustls::pki_types::pem::PemObject;
17use tokio_rustls::rustls::pki_types::{CertificateDer, ServerName};
18use tokio_rustls::rustls::{ClientConfig, RootCertStore};
19
20use crate::caliband::wire::Endpoint;
21
22/// A duplex byte stream over any transport family.
23pub trait Conn: AsyncRead + AsyncWrite + Unpin + Send {}
24impl<T: AsyncRead + AsyncWrite + Unpin + Send> Conn for T {}
25
26/// Boxed duplex connection handed to the NDJSON protocol layer.
27pub type BoxConn = Box<dyn Conn>;
28
29/// Client-side TLS material.
30#[derive(Clone)]
31pub struct TlsClient {
32    /// Handshake connector built from a trusted CA store.
33    pub connector: TlsConnector,
34    /// Expected server name (SNI / cert validation target).
35    pub server_name: String,
36}
37
38/// Install the `ring` crypto provider as the process default, exactly once.
39fn ensure_crypto_provider() {
40    use std::sync::Once;
41    static INIT: Once = Once::new();
42    INIT.call_once(|| {
43        let _ = tokio_rustls::rustls::crypto::ring::default_provider().install_default();
44    });
45}
46
47/// Build client TLS trusting `ca_pem`, verifying the server presents `server_name`.
48pub fn tls_client_from_pem(ca_pem: &[u8], server_name: &str) -> std::io::Result<TlsClient> {
49    ensure_crypto_provider();
50    let mut roots = RootCertStore::empty();
51    for cert in CertificateDer::pem_slice_iter(ca_pem) {
52        roots
53            .add(cert.map_err(|e| std::io::Error::other(e.to_string()))?)
54            .map_err(std::io::Error::other)?;
55    }
56    // A trust store with zero certs can never validate a peer — it always
57    // silently rejects. That only happens when `ca_pem` held no CERTIFICATE PEM
58    // block (an empty or garbage file), which is a misconfiguration; surface it
59    // as an error rather than a useless-but-`Ok` client.
60    if roots.is_empty() {
61        return Err(std::io::Error::new(
62            std::io::ErrorKind::InvalidData,
63            "no certificates found in CA PEM",
64        ));
65    }
66    let config = ClientConfig::builder()
67        .with_root_certificates(roots)
68        .with_no_client_auth();
69    Ok(TlsClient {
70        connector: TlsConnector::from(Arc::new(config)),
71        server_name: server_name.to_string(),
72    })
73}
74
75/// Bearer-token preamble: `{"bearer":"<token>"}\n`, sent after the TLS
76/// handshake so it travels encrypted. TCP only; Unix never sends it. The daemon
77/// refuses to start with a token but no TLS (see `require_token_tls` in
78/// `prosperod`), so this preamble is never written over a plaintext socket.
79#[derive(Serialize, Deserialize)]
80struct TokenPreamble {
81    bearer: String,
82}
83
84async fn client_send_token(conn: &mut BoxConn, token: &str) -> std::io::Result<()> {
85    let mut line = serde_json::to_vec(&TokenPreamble {
86        bearer: token.to_string(),
87    })
88    .map_err(std::io::Error::other)?;
89    line.push(b'\n');
90    conn.write_all(&line).await?;
91    conn.flush().await
92}
93
94/// How to dial a connection.
95pub struct ConnectSpec {
96    /// Target address.
97    pub endpoint: Endpoint,
98    /// TLS (TCP only).
99    pub tls: Option<TlsClient>,
100    /// Bearer token to present (TCP only).
101    pub token: Option<String>,
102}
103
104/// Dial a connection per `spec`: TLS handshake when configured, then the
105/// bearer-token preamble when a token is configured.
106pub async fn connect(spec: &ConnectSpec) -> std::io::Result<BoxConn> {
107    match &spec.endpoint {
108        Endpoint::Unix { path } => Ok(Box::new(UnixStream::connect(path).await?)),
109        Endpoint::Tcp { addr } => {
110            let stream = TcpStream::connect(addr).await?;
111            let mut conn: BoxConn = match &spec.tls {
112                None => Box::new(stream),
113                Some(t) => {
114                    let name = ServerName::try_from(t.server_name.clone())
115                        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput, e))?;
116                    Box::new(t.connector.connect(name, stream).await?)
117                }
118            };
119            if let Some(token) = &spec.token {
120                client_send_token(&mut conn, token).await?;
121            }
122            Ok(conn)
123        }
124    }
125}
126
127// ---- Server half: only compiled for the test harness (`FakeCaliband`). ----
128#[cfg(any(test, feature = "testkit"))]
129mod server {
130    use super::{BoxConn, Endpoint, TokenPreamble, ensure_crypto_provider};
131    use std::sync::Arc;
132    use tokio::io::AsyncReadExt as _;
133    use tokio::net::{TcpListener, UnixListener};
134    use tokio_rustls::TlsAcceptor;
135    use tokio_rustls::rustls::ServerConfig;
136    use tokio_rustls::rustls::pki_types::pem::PemObject;
137    use tokio_rustls::rustls::pki_types::{CertificateDer, PrivateKeyDer};
138
139    /// Server-side TLS material.
140    #[derive(Clone)]
141    pub struct TlsServer {
142        /// Handshake acceptor built from a cert chain + private key.
143        pub acceptor: TlsAcceptor,
144    }
145
146    /// Build server TLS from a PEM cert chain + private key.
147    pub fn tls_server_from_pem(cert_pem: &[u8], key_pem: &[u8]) -> std::io::Result<TlsServer> {
148        ensure_crypto_provider();
149        let certs: Vec<CertificateDer<'static>> = CertificateDer::pem_slice_iter(cert_pem)
150            .collect::<Result<_, _>>()
151            .map_err(|e| std::io::Error::other(e.to_string()))?;
152        let key: PrivateKeyDer<'static> = PrivateKeyDer::from_pem_slice(key_pem)
153            .map_err(|e| std::io::Error::other(e.to_string()))?;
154        let config = ServerConfig::builder()
155            .with_no_client_auth()
156            .with_single_cert(certs, key)
157            .map_err(std::io::Error::other)?;
158        Ok(TlsServer {
159            acceptor: TlsAcceptor::from(Arc::new(config)),
160        })
161    }
162
163    async fn read_preamble_line(conn: &mut BoxConn) -> std::io::Result<String> {
164        let mut buf = Vec::with_capacity(128);
165        let mut byte = [0u8; 1];
166        loop {
167            let n = conn.read(&mut byte).await?;
168            if n == 0 {
169                return Err(std::io::Error::new(
170                    std::io::ErrorKind::UnexpectedEof,
171                    "no token preamble",
172                ));
173            }
174            if byte[0] == b'\n' {
175                break;
176            }
177            buf.push(byte[0]);
178            if buf.len() > 4096 {
179                return Err(std::io::Error::new(
180                    std::io::ErrorKind::InvalidData,
181                    "token preamble too long",
182                ));
183            }
184        }
185        String::from_utf8(buf).map_err(std::io::Error::other)
186    }
187
188    /// Constant-time byte-string equality. Compares in time that depends only
189    /// on the *lengths* of the inputs, never on *where* they first differ, so a
190    /// bearer-token check can't be turned into a timing oracle that recovers
191    /// the token byte-by-byte. `subtle` isn't a dependency of this crate, so
192    /// this is a minimal hand-rolled version: bail early only on a length
193    /// mismatch (a token's length isn't secret), then XOR-accumulate every byte
194    /// so the loop always runs to completion regardless of the first mismatch.
195    fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
196        if a.len() != b.len() {
197            return false;
198        }
199        let mut diff: u8 = 0;
200        for (x, y) in a.iter().zip(b.iter()) {
201            diff |= x ^ y;
202        }
203        diff == 0
204    }
205
206    async fn server_check_token(conn: &mut BoxConn, expected: &str) -> std::io::Result<()> {
207        let line = read_preamble_line(conn).await?;
208        let preamble: TokenPreamble = serde_json::from_str(&line).map_err(std::io::Error::other)?;
209        if constant_time_eq(preamble.bearer.as_bytes(), expected.as_bytes()) {
210            Ok(())
211        } else {
212            Err(std::io::Error::new(
213                std::io::ErrorKind::PermissionDenied,
214                "bad bearer token",
215            ))
216        }
217    }
218
219    /// How to bind a listener.
220    pub struct BindSpec {
221        /// Address family + address.
222        pub endpoint: Endpoint,
223        /// TLS (TCP only). `None` = plaintext.
224        pub tls: Option<TlsServer>,
225        /// Required bearer token for network connections.
226        pub token: Option<String>,
227    }
228
229    /// A bound listener over one transport family.
230    pub enum Listener {
231        /// Unix-domain.
232        Unix(UnixListener),
233        /// TCP (TLS/token applied at accept-time).
234        Tcp {
235            /// Underlying listener.
236            listener: TcpListener,
237            /// Server TLS material, if any.
238            tls: Option<TlsServer>,
239            /// Required bearer token, if any.
240            token: Option<String>,
241        },
242    }
243
244    impl Listener {
245        /// Bind a listener per `spec`.
246        pub async fn bind(spec: &BindSpec) -> std::io::Result<Listener> {
247            match &spec.endpoint {
248                Endpoint::Unix { path } => {
249                    if let Some(parent) = path.parent() {
250                        tokio::fs::create_dir_all(parent).await?;
251                    }
252                    let _ = tokio::fs::remove_file(path).await;
253                    Ok(Listener::Unix(UnixListener::bind(path)?))
254                }
255                Endpoint::Tcp { addr } => Ok(Listener::Tcp {
256                    listener: TcpListener::bind(addr).await?,
257                    tls: spec.tls.clone(),
258                    token: spec.token.clone(),
259                }),
260            }
261        }
262
263        /// The actually-bound TCP address (resolves `:0`); `None` for Unix.
264        pub fn local_addr(&self) -> Option<String> {
265            match self {
266                Listener::Unix(_) => None,
267                Listener::Tcp { listener, .. } => listener.local_addr().ok().map(|a| a.to_string()),
268            }
269        }
270
271        /// Accept one connection, performing the TLS handshake + token check
272        /// (TCP) when configured.
273        pub async fn accept(&self) -> std::io::Result<BoxConn> {
274            match self {
275                Listener::Unix(l) => {
276                    let (stream, _addr) = l.accept().await?;
277                    Ok(Box::new(stream))
278                }
279                Listener::Tcp {
280                    listener,
281                    tls,
282                    token,
283                } => {
284                    let (stream, _addr) = listener.accept().await?;
285                    let mut conn: BoxConn = match tls {
286                        None => Box::new(stream),
287                        Some(t) => Box::new(t.acceptor.accept(stream).await?),
288                    };
289                    if let Some(expected) = token {
290                        server_check_token(&mut conn, expected).await?;
291                    }
292                    Ok(conn)
293                }
294            }
295        }
296    }
297}
298
299#[cfg(any(test, feature = "testkit"))]
300pub use server::{BindSpec, Listener, TlsServer, tls_server_from_pem};
301
302#[cfg(test)]
303mod tests {
304    use super::*;
305    use tokio::io::AsyncReadExt as _;
306
307    async fn echo_once(listener: Listener) {
308        let mut c = listener.accept().await.expect("accept");
309        let mut buf = [0u8; 5];
310        c.read_exact(&mut buf).await.expect("read");
311        c.write_all(&buf).await.expect("write");
312        c.flush().await.expect("flush");
313    }
314
315    #[tokio::test]
316    async fn tcp_tls_token_round_trip() {
317        let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap();
318        let cert_pem = cert.cert.pem().into_bytes();
319        let key_pem = cert.key_pair.serialize_pem().into_bytes();
320
321        let listener = Listener::bind(&BindSpec {
322            endpoint: Endpoint::Tcp {
323                addr: "127.0.0.1:0".into(),
324            },
325            tls: Some(tls_server_from_pem(&cert_pem, &key_pem).unwrap()),
326            token: Some("s3cr3t".into()),
327        })
328        .await
329        .unwrap();
330        let addr = listener.local_addr().unwrap();
331        let server = tokio::spawn(echo_once(listener));
332
333        let mut c = connect(&ConnectSpec {
334            endpoint: Endpoint::Tcp { addr },
335            tls: Some(tls_client_from_pem(&cert_pem, "localhost").unwrap()),
336            token: Some("s3cr3t".into()),
337        })
338        .await
339        .unwrap();
340        c.write_all(b"hello").await.unwrap();
341        let mut got = [0u8; 5];
342        c.read_exact(&mut got).await.unwrap();
343        assert_eq!(&got, b"hello");
344        server.await.unwrap();
345    }
346
347    #[tokio::test]
348    async fn bad_token_is_rejected() {
349        let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap();
350        let cert_pem = cert.cert.pem().into_bytes();
351        let key_pem = cert.key_pair.serialize_pem().into_bytes();
352        let listener = Listener::bind(&BindSpec {
353            endpoint: Endpoint::Tcp {
354                addr: "127.0.0.1:0".into(),
355            },
356            tls: Some(tls_server_from_pem(&cert_pem, &key_pem).unwrap()),
357            token: Some("right".into()),
358        })
359        .await
360        .unwrap();
361        let addr = listener.local_addr().unwrap();
362        tokio::spawn(async move {
363            let _ = listener.accept().await;
364        });
365        let r = connect(&ConnectSpec {
366            endpoint: Endpoint::Tcp { addr },
367            tls: Some(tls_client_from_pem(&cert_pem, "localhost").unwrap()),
368            token: Some("wrong".into()),
369        })
370        .await;
371        // connect() sends the token then returns Ok; the server rejects on
372        // accept and drops the connection, so a subsequent read sees EOF/err.
373        if let Ok(mut c) = r {
374            let mut b = [0u8; 1];
375            assert!(c.read(&mut b).await.map(|n| n == 0).unwrap_or(true));
376        }
377    }
378
379    #[tokio::test]
380    async fn unix_round_trip() {
381        let dir = tempfile::tempdir().unwrap();
382        let path = dir.path().join("t.sock");
383        let listener = Listener::bind(&BindSpec {
384            endpoint: Endpoint::Unix { path: path.clone() },
385            tls: None,
386            token: None,
387        })
388        .await
389        .unwrap();
390        let server = tokio::spawn(echo_once(listener));
391        let mut c = connect(&ConnectSpec {
392            endpoint: Endpoint::Unix { path },
393            tls: None,
394            token: None,
395        })
396        .await
397        .unwrap();
398        c.write_all(b"world").await.unwrap();
399        let mut got = [0u8; 5];
400        c.read_exact(&mut got).await.unwrap();
401        assert_eq!(&got, b"world");
402        server.await.unwrap();
403    }
404}
prospero_core/caliband/transport.rs

prospero_core/caliband/
transport.rs
