use capnp::traits::ImbueMut;
use futures::io::Cursor;
use libcfd_rpc::{quic_metadata_protocol_capnp, rpc_capnp, tunnelrpc_capnp};
use tokio::io::AsyncWriteExt as _;
mod common;
use common::{StubHook, TokioBridge};
fn hex(bytes: &[u8]) -> String {
bytes.iter().map(|x| format!("{x:02x}")).collect()
}
/// Golden bytes produced by capnp-go v2.18.0 (the wire format cloudflared
/// uses): each constant was generated by a Go program running against the
/// vendored `zombiezen.com/go/capnproto2` in this repo's cloudflared
/// checkout, then confirmed byte-identical to what this crate's serializer
/// emits for the same logical message (see `golden_messages_match_go_reference`).
const GOLDEN: &[(&str, &str)] = &[
(
"bootstrap",
"000000000500000000000000010001000800000000000000000000000100010000000000000000000000000000000000",
),
(
"call",
"000000003000000000000000010001000200000000000000000000000300030001000000000000009754e87fec9516f7000000000000000008000000010001000c0000000000020000000000000000000000000000000000000000000000000004000000010003008900000007000000000000000000000008000000000002001d000000820000002000000001000200050000008200000009000000820000006163636f756e742d7461672d313233000102030405060708090a0b0c0d0e0f10aabbccddeeff001122334455667788990000010000000000040000000000040049000000220000000d0000008200000011000000160000002d0000004a00000031000000620000003031323334353637383961626364656605000000a20000000d000000a2000000616c6c6f775f72656d6f74655f636f6e6669670000000000737570706f72745f646174616772616d5f76320000000000323032362e372e3300000000000000006c696e75782f616d64363400000000000a000001000000000000000001000100",
),
(
"bootstrap-return",
"000000000b00000000000000010001000300000000000000000000000200010000000000010000000000000000000000000000000000020003000000000000000100000017000000040000000100010001000000000000000000000000000000",
),
(
"register-return",
"0000000012000000000000000100010003000000000000000000000002000100010000000100000000000000000000000000000000000200040000000000010025000000070000000000000001000100010000000000000000000000010002000000000000000000050000008200000009000000220000000102030405060708090a0b0c0d0e0f106c687200000000000000000001000100",
),
(
"unregister-call",
"000000000e00000000000000010001000200000000000000000000000300030002000000010000009754e87fec9516f7000000000000000008000000010001000c00000000000200000000000000000000000000000000000000000000000000fcffffff0000000001000000070000000000000001000100",
),
(
"finish",
"00000000040000000000000001000100040000000000000000000000010000000100000001000000",
),
(
"empty-return",
"0000000009000000000000000100010003000000000000000000000002000100020000000100000000000000000000000000000000000200000000000000000001000000070000000000000001000100",
),
(
"release",
"00000000040000000000000001000100060000000000000000000000010000000000000001000000",
),
(
"connect-request",
"00000000130000000000000001000200000000000000000005000000c20000000d00000027000000687474703a2f2f6578616d706c652e636f6d2f706174680008000000000002000d0000005a0000001100000022000000110000004a0000001500000062000000487474704d6574686f64000000000000474554000000000048747470486f737400000000000000006578616d706c652e636f6d0000000000",
),
(
"connect-response",
"000000001000000000000000000002000000000000000000010000002700000008000000000002000d0000005a000000110000002200000011000000c2000000190000005a000000487474705374617475730000000000003230300000000000487474704865616465723a636f6e74656e742d7479706500746578742f706c61696e000000000000",
),
];
fn golden_hex(name: &str) -> Vec<u8> {
let (_, h) = GOLDEN.iter().find(|(n, _)| *n == name).expect("golden");
(0..h.len())
.step_by(2)
.map(|i| u8::from_str_radix(&h[i..i + 2], 16).unwrap())
.collect()
}
#[test]
fn golden_messages_match_go_reference() {
// bootstrap
{
let mut message = capnp::message::Builder::new_default();
let root = message.init_root::<rpc_capnp::message::Builder>();
let mut bs = root.init_bootstrap();
bs.set_question_id(0);
assert_eq!(
hex(&capnp::serialize::write_message_to_words(&message)),
golden_hex("bootstrap")
.iter()
.map(|b| format!("{b:02x}"))
.collect::<String>()
);
}
// call registerConnection
{
let mut message = capnp::message::Builder::new_default();
let root = message.init_root::<rpc_capnp::message::Builder>();
let mut call = root.init_call();
call.set_question_id(1);
let mut mt = call.reborrow().init_target();
mt.set_imported_cap(0);
call.reborrow().set_interface_id(0xf71695ec7fe85497);
call.reborrow().set_method_id(0);
call.reborrow().init_send_results_to().set_caller(());
let mut payload = call.reborrow().init_params();
let mut params = payload
.reborrow()
.init_content()
.init_as::<tunnelrpc_capnp::registration_server::register_connection_params::Builder>(
);
{
let mut a = params.reborrow().init_auth();
a.set_account_tag("account-tag-123");
a.set_tunnel_secret(&[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16]);
}
params.set_tunnel_id(&[
0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff, 0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77,
0x88, 0x99,
]);
params.set_conn_index(0);
{
let mut o = params.reborrow().init_options();
let mut c = o.reborrow().init_client();
c.set_client_id(b"0123456789abcdef");
let mut feats = c.reborrow().init_features(2);
feats.set(0, "allow_remote_config");
feats.set(1, "support_datagram_v2");
c.set_version("2026.7.3");
c.set_arch("linux/amd64");
o.set_origin_local_ip(&[10, 0, 0, 1]);
o.set_replace_existing(false);
o.set_compression_quality(0);
o.set_num_previous_attempts(1);
}
payload.reborrow().init_cap_table(0);
assert_eq!(
hex(&capnp::serialize::write_message_to_words(&message)),
golden_hex("call")
.iter()
.map(|b| format!("{b:02x}"))
.collect::<String>()
);
}
// bootstrap-return
{
let mut message = capnp::message::Builder::new_default();
let mut cap_table: capnp::private::layout::CapTable = Vec::new();
let mut root = message.init_root::<rpc_capnp::message::Builder>();
root.imbue_mut(&mut cap_table);
let mut ret = root.init_return();
ret.set_answer_id(0);
ret.set_release_param_caps(false);
let mut results = ret.init_results();
let mut content = results.reborrow().init_content();
content.set_as_capability(Box::new(StubHook));
let mut ctab = results.init_cap_table(1);
ctab.reborrow().get(0).set_sender_hosted(0);
assert_eq!(
hex(&capnp::serialize::write_message_to_words(&message)),
golden_hex("bootstrap-return")
.iter()
.map(|b| format!("{b:02x}"))
.collect::<String>()
);
}
// register-return
{
let mut message = capnp::message::Builder::new_default();
let root = message.init_root::<rpc_capnp::message::Builder>();
let mut ret = root.init_return();
ret.set_answer_id(1);
ret.set_release_param_caps(false);
let mut res = ret.init_results();
let mut rres = res
.reborrow()
.init_content()
.init_as::<tunnelrpc_capnp::registration_server::register_connection_results::Builder>(
);
let mut conn_resp = rres.reborrow().init_result();
let mut cd = conn_resp.reborrow().init_result().init_connection_details();
cd.set_uuid(&[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16]);
cd.set_location_name("lhr");
cd.set_tunnel_is_remotely_managed(false);
res.reborrow().init_cap_table(0);
assert_eq!(
hex(&capnp::serialize::write_message_to_words(&message)),
golden_hex("register-return")
.iter()
.map(|b| format!("{b:02x}"))
.collect::<String>()
);
}
// finish
{
let mut message = capnp::message::Builder::new_default();
let root = message.init_root::<rpc_capnp::message::Builder>();
let mut f = root.init_finish();
f.set_question_id(1);
f.set_release_result_caps(false);
assert_eq!(
hex(&capnp::serialize::write_message_to_words(&message)),
golden_hex("finish")
.iter()
.map(|b| format!("{b:02x}"))
.collect::<String>()
);
}
}
#[tokio::test]
async fn framing_round_trip() {
let (a, b) = tokio::io::duplex(4096);
let (mut a, mut b) = (a, TokioBridge(b));
let mut message = capnp::message::Builder::new_default();
let root = message.init_root::<rpc_capnp::message::Builder>();
let mut bs = root.init_bootstrap();
bs.set_question_id(7);
let bytes = capnp::serialize::write_message_to_words(&message);
a.write_all(&bytes).await.unwrap();
let reader = libcfd_rpc::io::read_message(&mut b).await.unwrap();
let root = reader.get_root::<rpc_capnp::message::Reader>().unwrap();
match root.reborrow().which().unwrap() {
rpc_capnp::message::Bootstrap(bs) => {
assert_eq!(bs.unwrap().get_question_id(), 7);
}
_ => panic!("expected bootstrap"),
}
}
#[tokio::test]
async fn golden_messages_parse() {
// The golden call message should parse as a call targeting the
// registration server.
let mut stream = Cursor::new(golden_hex("call"));
let reader = libcfd_rpc::io::read_message(&mut stream).await.unwrap();
let root = reader.get_root::<rpc_capnp::message::Reader>().unwrap();
match root.reborrow().which().unwrap() {
rpc_capnp::message::Call(c) => {
let c = c.unwrap();
assert_eq!(c.get_question_id(), 1);
assert_eq!(c.get_interface_id(), 0xf71695ec7fe85497);
assert_eq!(c.get_method_id(), 0);
let target = c.get_target().unwrap();
match target.reborrow().which().unwrap() {
rpc_capnp::message_target::ImportedCap(id) => assert_eq!(id, 0),
_ => panic!("expected importedCap"),
}
}
_ => panic!("expected call"),
}
}
#[tokio::test]
async fn golden_register_return_parses() {
// The golden register-return must decode to connectionDetails with the
// uuid/locationName written by capnp-go.
let mut stream = Cursor::new(golden_hex("register-return"));
let reader = libcfd_rpc::io::read_message(&mut stream).await.unwrap();
let root = reader.get_root::<rpc_capnp::message::Reader>().unwrap();
let ret = match root.reborrow().which().unwrap() {
rpc_capnp::message::Return(r) => r.unwrap(),
_ => panic!("expected return"),
};
assert_eq!(ret.get_answer_id(), 1);
let payload = match ret.reborrow().which().unwrap() {
rpc_capnp::return_::Results(r) => r.unwrap(),
_ => panic!("expected results"),
};
let rres = payload
.get_content()
.get_as::<tunnelrpc_capnp::registration_server::register_connection_results::Reader<'_>>()
.unwrap();
let conn_resp = rres.reborrow().get_result().unwrap();
match conn_resp.reborrow().get_result().which().unwrap() {
tunnelrpc_capnp::connection_response::result::ConnectionDetails(d) => {
let d = d.unwrap();
assert_eq!(
d.get_uuid().unwrap(),
&[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16]
);
assert_eq!(d.get_location_name().unwrap(), "lhr");
assert!(!d.get_tunnel_is_remotely_managed());
}
_ => panic!("expected connectionDetails"),
}
}
#[tokio::test]
async fn golden_release_parses() {
let mut stream = Cursor::new(golden_hex("release"));
let reader = libcfd_rpc::io::read_message(&mut stream).await.unwrap();
let root = reader.get_root::<rpc_capnp::message::Reader>().unwrap();
match root.reborrow().which().unwrap() {
rpc_capnp::message::Release(r) => {
let r = r.unwrap();
assert_eq!(r.get_id(), 0);
assert_eq!(r.get_reference_count(), 1);
}
_ => panic!("expected release"),
}
}
/// The crate's public encoder must produce the exact golden bytes capnp-go
/// generates for the same logical messages, proving wire compatibility with
/// cloudflared rather than mere self-consistency.
#[test]
fn golden_connect_messages_match_crate_encoding() {
use libcfd_rpc::quic::{ConnectRequest, ConnectResponse};
let request = ConnectRequest {
destination: "http://example.com/path".into(),
connection_type: libcfd_rpc::ConnectionType::Http,
metadata: vec![
("HttpMethod".into(), "GET".into()),
("HttpHost".into(), "example.com".into()),
],
};
let message = libcfd_rpc::quic::encode_connect_request(&request).unwrap();
assert_eq!(
hex(&capnp::serialize::write_message_to_words(&message)),
golden_hex("connect-request")
.iter()
.map(|b| format!("{b:02x}"))
.collect::<String>()
);
let response = ConnectResponse {
error: String::new(),
metadata: vec![
("HttpStatus".into(), "200".into()),
("HttpHeader:content-type".into(), "text/plain".into()),
],
};
let message = libcfd_rpc::quic::encode_connect_response(&response).unwrap();
assert_eq!(
hex(&capnp::serialize::write_message_to_words(&message)),
golden_hex("connect-response")
.iter()
.map(|b| format!("{b:02x}"))
.collect::<String>()
);
}
#[tokio::test]
async fn golden_connect_messages_parse() {
let mut stream = Cursor::new(golden_hex("connect-request"));
let reader = libcfd_rpc::io::read_message(&mut stream).await.unwrap();
let r = reader
.get_root::<quic_metadata_protocol_capnp::connect_request::Reader>()
.unwrap();
assert_eq!(r.get_dest().unwrap(), "http://example.com/path");
assert_eq!(
r.get_type().unwrap(),
quic_metadata_protocol_capnp::ConnectionType::Http
);
let md = r.get_metadata().unwrap();
assert_eq!(md.len(), 2);
assert_eq!(md.get(0).get_key().unwrap(), "HttpMethod");
assert_eq!(md.get(0).get_val().unwrap(), "GET");
assert_eq!(md.get(1).get_key().unwrap(), "HttpHost");
assert_eq!(md.get(1).get_val().unwrap(), "example.com");
let mut stream = Cursor::new(golden_hex("connect-response"));
let reader = libcfd_rpc::io::read_message(&mut stream).await.unwrap();
let r = reader
.get_root::<quic_metadata_protocol_capnp::connect_response::Reader>()
.unwrap();
assert_eq!(r.get_error().unwrap(), "");
let md = r.get_metadata().unwrap();
assert_eq!(md.len(), 2);
assert_eq!(md.get(0).get_key().unwrap(), "HttpStatus");
assert_eq!(md.get(0).get_val().unwrap(), "200");
assert_eq!(md.get(1).get_key().unwrap(), "HttpHeader:content-type");
assert_eq!(md.get(1).get_val().unwrap(), "text/plain");
}
#[tokio::test]
async fn read_message_rejects_oversized_segment() {
// Header claiming a single segment larger than capnp-go's 64 MiB
// decode limit; read_message must reject before allocating.
let mut bytes = vec![0u8; 8];
bytes[4..8].copy_from_slice(&((8 * 1024 * 1024 + 1) as u32).to_le_bytes());
let mut stream = Cursor::new(bytes);
match libcfd_rpc::io::read_message(&mut stream).await {
Err(libcfd_rpc::RpcError::Protocol(_)) => {}
Err(e) => panic!("expected protocol error, got {e}"),
Ok(_) => panic!("expected protocol error, got a message"),
}
}
#[tokio::test]
async fn read_message_rejects_segment_count_wraparound() {
// count-1 = u32::MAX wraps to segment_count 0.
let mut bytes = vec![0u8; 8];
bytes[0..4].copy_from_slice(&u32::MAX.to_le_bytes());
let mut stream = Cursor::new(bytes);
match libcfd_rpc::io::read_message(&mut stream).await {
Err(libcfd_rpc::RpcError::Protocol(_)) => {}
Err(e) => panic!("expected protocol error, got {e}"),
Ok(_) => panic!("expected protocol error, got a message"),
}
}
#[tokio::test]
async fn read_message_rejects_too_many_segments() {
// count-1 = 512 makes segment_count 513, above the 512 cap.
let mut bytes = vec![0u8; 8];
bytes[0..4].copy_from_slice(&512u32.to_le_bytes());
bytes[4..8].copy_from_slice(&1u32.to_le_bytes());
let mut stream = Cursor::new(bytes);
match libcfd_rpc::io::read_message(&mut stream).await {
Err(libcfd_rpc::RpcError::Protocol(_)) => {}
Err(e) => panic!("expected protocol error, got {e}"),
Ok(_) => panic!("expected protocol error, got a message"),
}
}
#[tokio::test]
async fn read_message_handles_byte_at_a_time_writes() {
// Writing the framed bytes one byte at a time must not corrupt framing.
let mut message = capnp::message::Builder::new_default();
let root = message.init_root::<rpc_capnp::message::Builder>();
let mut bs = root.init_bootstrap();
bs.set_question_id(11);
let bytes = capnp::serialize::write_message_to_words(&message);
let mut stream = Cursor::new(Vec::new());
for byte in &bytes {
futures::io::AsyncWriteExt::write_all(&mut stream, &[*byte])
.await
.unwrap();
}
futures::io::AsyncWriteExt::flush(&mut stream)
.await
.unwrap();
let mut cursor = Cursor::new(stream.into_inner());
let reader = libcfd_rpc::io::read_message(&mut cursor).await.unwrap();
let root = reader.get_root::<rpc_capnp::message::Reader>().unwrap();
match root.reborrow().which().unwrap() {
rpc_capnp::message::Bootstrap(bs) => {
assert_eq!(bs.unwrap().get_question_id(), 11);
}
_ => panic!("expected bootstrap"),
}
}
#[tokio::test]
async fn read_message_handles_tiny_buffer_reads() {
// A 1-byte duplex forces every poll_read to return at most one byte;
// read_exact must reassemble the message regardless. The writer runs as
// a task so it can drain while the reader fills its buffer.
let (a, b) = tokio::io::duplex(1);
let mut b = TokioBridge(b);
let mut message = capnp::message::Builder::new_default();
let root = message.init_root::<rpc_capnp::message::Builder>();
let mut bs = root.init_bootstrap();
bs.set_question_id(3);
let bytes = capnp::serialize::write_message_to_words(&message);
let writer = tokio::spawn(async move {
let mut a = a;
a.write_all(&bytes).await.unwrap();
});
let reader = libcfd_rpc::io::read_message(&mut b).await.unwrap();
let root = reader.get_root::<rpc_capnp::message::Reader>().unwrap();
match root.reborrow().which().unwrap() {
rpc_capnp::message::Bootstrap(bs) => {
assert_eq!(bs.unwrap().get_question_id(), 3);
}
_ => panic!("expected bootstrap"),
}
writer.await.unwrap();
}
#[tokio::test]
async fn read_message_multi_segment_round_trip() {
// Force a second segment with a tiny first segment and a long string.
let allocator = capnp::message::HeapAllocator::new().first_segment_words(8);
let mut message = capnp::message::Builder::new(allocator);
let root = message.init_root::<rpc_capnp::message::Builder>();
let mut call = root.init_call();
call.set_question_id(9);
let mut payload = call.reborrow().init_params();
let mut params = payload
.reborrow()
.init_content()
.init_as::<tunnelrpc_capnp::registration_server::register_connection_params::Builder>(
);
let mut auth = params.reborrow().init_auth();
auth.set_account_tag("a".repeat(200));
auth.set_tunnel_secret(&[7u8; 32]);
payload.reborrow().init_cap_table(0);
let mut bytes = Vec::new();
capnp::serialize::write_message(&mut bytes, &message).unwrap();
let segment_table_count = u32::from_le_bytes(bytes[0..4].try_into().unwrap());
assert!(segment_table_count >= 1, "expected at least two segments");
let second_segment_words = u32::from_le_bytes(bytes[8..12].try_into().unwrap());
assert!(second_segment_words > 0, "expected a second segment");
let mut stream = Cursor::new(bytes);
let reader = libcfd_rpc::io::read_message(&mut stream).await.unwrap();
let root = reader.get_root::<rpc_capnp::message::Reader>().unwrap();
match root.reborrow().which().unwrap() {
rpc_capnp::message::Call(c) => {
let c = c.unwrap();
assert_eq!(c.get_question_id(), 9);
let payload = c.get_params().unwrap();
let params = payload
.get_content()
.get_as::<tunnelrpc_capnp::registration_server::register_connection_params::Reader<'_>>(
)
.unwrap();
let auth = params.get_auth().unwrap();
assert_eq!(
auth.get_account_tag().unwrap().to_str().unwrap(),
&"a".repeat(200)
);
assert_eq!(auth.get_tunnel_secret().unwrap(), &[7u8; 32]);
}
_ => panic!("expected call"),
}
}
#[tokio::test]
async fn read_message_truncated_header_yields_eof() {
// Only four of the eight header bytes arrive before EOF.
let mut bytes = vec![0u8; 4];
bytes[0..4].copy_from_slice(&0u32.to_le_bytes());
let mut stream = Cursor::new(bytes);
match libcfd_rpc::io::read_message(&mut stream).await {
Err(libcfd_rpc::RpcError::Eof) => {}
Err(e) => panic!("expected eof, got {e}"),
Ok(_) => panic!("expected eof, got a message"),
}
}
#[tokio::test]
async fn read_message_truncated_body_yields_eof() {
// The header promises one segment of two words, but only two bytes of
// payload arrive before EOF.
let mut bytes = Vec::new();
bytes.extend_from_slice(&0u32.to_le_bytes());
bytes.extend_from_slice(&2u32.to_le_bytes());
bytes.extend_from_slice(&[0xab, 0xcd]);
let mut stream = Cursor::new(bytes);
match libcfd_rpc::io::read_message(&mut stream).await {
Err(libcfd_rpc::RpcError::Eof) => {}
Err(e) => panic!("expected eof, got {e}"),
Ok(_) => panic!("expected eof, got a message"),
}
}
#[tokio::test]
async fn read_message_rejects_corrupt_segment_count() {
// A segment count larger than the data actually present must not parse
// as a success: the reader hits EOF before the payload is complete.
let mut bytes = Vec::new();
bytes.extend_from_slice(&4u32.to_le_bytes());
bytes.extend_from_slice(&1u32.to_le_bytes());
bytes.extend_from_slice(&[0u8; 8]);
let mut stream = Cursor::new(bytes);
match libcfd_rpc::io::read_message(&mut stream).await {
Err(libcfd_rpc::RpcError::Eof) => {}
Err(e) => panic!("expected eof, got {e}"),
Ok(_) => panic!("expected eof, got a message"),
}
}