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    async fn server_check_token(conn: &mut BoxConn, expected: &str) -> std::io::Result<()> {
189        let line = read_preamble_line(conn).await?;
190        let preamble: TokenPreamble = serde_json::from_str(&line).map_err(std::io::Error::other)?;
191        if crate::auth::constant_time_eq(preamble.bearer.as_bytes(), expected.as_bytes()) {
192            Ok(())
193        } else {
194            Err(std::io::Error::new(
195                std::io::ErrorKind::PermissionDenied,
196                "bad bearer token",
197            ))
198        }
199    }
200
201    /// How to bind a listener.
202    pub struct BindSpec {
203        /// Address family + address.
204        pub endpoint: Endpoint,
205        /// TLS (TCP only). `None` = plaintext.
206        pub tls: Option<TlsServer>,
207        /// Required bearer token for network connections.
208        pub token: Option<String>,
209    }
210
211    /// A bound listener over one transport family.
212    pub enum Listener {
213        /// Unix-domain.
214        Unix(UnixListener),
215        /// TCP (TLS/token applied at accept-time).
216        Tcp {
217            /// Underlying listener.
218            listener: TcpListener,
219            /// Server TLS material, if any.
220            tls: Option<TlsServer>,
221            /// Required bearer token, if any.
222            token: Option<String>,
223        },
224    }
225
226    impl Listener {
227        /// Bind a listener per `spec`.
228        pub async fn bind(spec: &BindSpec) -> std::io::Result<Listener> {
229            match &spec.endpoint {
230                Endpoint::Unix { path } => {
231                    if let Some(parent) = path.parent() {
232                        tokio::fs::create_dir_all(parent).await?;
233                    }
234                    let _ = tokio::fs::remove_file(path).await;
235                    Ok(Listener::Unix(UnixListener::bind(path)?))
236                }
237                Endpoint::Tcp { addr } => Ok(Listener::Tcp {
238                    listener: TcpListener::bind(addr).await?,
239                    tls: spec.tls.clone(),
240                    token: spec.token.clone(),
241                }),
242            }
243        }
244
245        /// The actually-bound TCP address (resolves `:0`); `None` for Unix.
246        pub fn local_addr(&self) -> Option<String> {
247            match self {
248                Listener::Unix(_) => None,
249                Listener::Tcp { listener, .. } => listener.local_addr().ok().map(|a| a.to_string()),
250            }
251        }
252
253        /// Accept one connection, performing the TLS handshake + token check
254        /// (TCP) when configured.
255        pub async fn accept(&self) -> std::io::Result<BoxConn> {
256            match self {
257                Listener::Unix(l) => {
258                    let (stream, _addr) = l.accept().await?;
259                    Ok(Box::new(stream))
260                }
261                Listener::Tcp {
262                    listener,
263                    tls,
264                    token,
265                } => {
266                    let (stream, _addr) = listener.accept().await?;
267                    let mut conn: BoxConn = match tls {
268                        None => Box::new(stream),
269                        Some(t) => Box::new(t.acceptor.accept(stream).await?),
270                    };
271                    if let Some(expected) = token {
272                        server_check_token(&mut conn, expected).await?;
273                    }
274                    Ok(conn)
275                }
276            }
277        }
278    }
279}
280
281#[cfg(any(test, feature = "testkit"))]
282pub use server::{BindSpec, Listener, TlsServer, tls_server_from_pem};
283
284#[cfg(test)]
285mod tests {
286    use super::*;
287    use tokio::io::AsyncReadExt as _;
288
289    async fn echo_once(listener: Listener) {
290        let mut c = listener.accept().await.expect("accept");
291        let mut buf = [0u8; 5];
292        c.read_exact(&mut buf).await.expect("read");
293        c.write_all(&buf).await.expect("write");
294        c.flush().await.expect("flush");
295    }
296
297    #[tokio::test]
298    async fn tcp_tls_token_round_trip() {
299        let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap();
300        let cert_pem = cert.cert.pem().into_bytes();
301        let key_pem = cert.key_pair.serialize_pem().into_bytes();
302
303        let listener = Listener::bind(&BindSpec {
304            endpoint: Endpoint::Tcp {
305                addr: "127.0.0.1:0".into(),
306            },
307            tls: Some(tls_server_from_pem(&cert_pem, &key_pem).unwrap()),
308            token: Some("s3cr3t".into()),
309        })
310        .await
311        .unwrap();
312        let addr = listener.local_addr().unwrap();
313        let server = tokio::spawn(echo_once(listener));
314
315        let mut c = connect(&ConnectSpec {
316            endpoint: Endpoint::Tcp { addr },
317            tls: Some(tls_client_from_pem(&cert_pem, "localhost").unwrap()),
318            token: Some("s3cr3t".into()),
319        })
320        .await
321        .unwrap();
322        c.write_all(b"hello").await.unwrap();
323        let mut got = [0u8; 5];
324        c.read_exact(&mut got).await.unwrap();
325        assert_eq!(&got, b"hello");
326        server.await.unwrap();
327    }
328
329    #[tokio::test]
330    async fn bad_token_is_rejected() {
331        let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap();
332        let cert_pem = cert.cert.pem().into_bytes();
333        let key_pem = cert.key_pair.serialize_pem().into_bytes();
334        let listener = Listener::bind(&BindSpec {
335            endpoint: Endpoint::Tcp {
336                addr: "127.0.0.1:0".into(),
337            },
338            tls: Some(tls_server_from_pem(&cert_pem, &key_pem).unwrap()),
339            token: Some("right".into()),
340        })
341        .await
342        .unwrap();
343        let addr = listener.local_addr().unwrap();
344        tokio::spawn(async move {
345            let _ = listener.accept().await;
346        });
347        let r = connect(&ConnectSpec {
348            endpoint: Endpoint::Tcp { addr },
349            tls: Some(tls_client_from_pem(&cert_pem, "localhost").unwrap()),
350            token: Some("wrong".into()),
351        })
352        .await;
353        // connect() sends the token then returns Ok; the server rejects on
354        // accept and drops the connection, so a subsequent read sees EOF/err.
355        if let Ok(mut c) = r {
356            let mut b = [0u8; 1];
357            assert!(c.read(&mut b).await.map(|n| n == 0).unwrap_or(true));
358        }
359    }
360
361    #[tokio::test]
362    async fn unix_round_trip() {
363        let dir = tempfile::tempdir().unwrap();
364        let path = dir.path().join("t.sock");
365        let listener = Listener::bind(&BindSpec {
366            endpoint: Endpoint::Unix { path: path.clone() },
367            tls: None,
368            token: None,
369        })
370        .await
371        .unwrap();
372        let server = tokio::spawn(echo_once(listener));
373        let mut c = connect(&ConnectSpec {
374            endpoint: Endpoint::Unix { path },
375            tls: None,
376            token: None,
377        })
378        .await
379        .unwrap();
380        c.write_all(b"world").await.unwrap();
381        let mut got = [0u8; 5];
382        c.read_exact(&mut got).await.unwrap();
383        assert_eq!(&got, b"world");
384        server.await.unwrap();
385    }
386}
prospero_core/caliband/transport.rs

prospero_core/caliband/
transport.rs
