use liminal::protocol::{
CausalContext, Frame, LinkEnd, LinkOffer, LinkOpenOutcome, MessageEnvelope,
};
use liminal_server::config::{ServerConfig, validate};
use liminal_server::link_carrier::{
Carrier, CarrierError, CarrierReceive, LinkCarrier, WireCarrier,
};
use liminal_server::link_dial::{LinkDialOutcome, LinkDialer};
use liminal_server::link_handshake::{LinkHandshake, NOISE_OVERHEAD};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::sync::{
Arc, Mutex,
atomic::{AtomicUsize, Ordering},
};
use std::thread::JoinHandle;
mod support;
use support::federation::{Federation, LiveOpen, local_machine, read_frame, write_frame};
type TestResult = Result<(), Box<dyn std::error::Error>>;
fn configuration(links: &str) -> Result<ServerConfig, Box<dyn std::error::Error>> {
Ok(toml::from_str(&format!(
r#"
listen_address = "127.0.0.1:41001"
health_listen_address = "127.0.0.1:41002"
drain_timeout_ms = 0
persistence_path = "/private/tmp"
channels = [{{ name = "orders.created", durable = true }}]
routing_rules = []
[auth]
token = "test-bearer"
[auth.lys]
issuer = "https://issuer.example"
network_id = "network-a"
keys_path = "keys.json"
[auth.lys.machine]
identity_path = "machine.key"
pass_path = "machine.pass"
[auth.lys.machine.machines]
importer = "{}"
{links}
"#,
hex::encode(local_machine(2).machine().ed25519_public())
))?)
}
fn link_entry(holder: &str, exports: &str) -> String {
format!(
r#"
[[auth.lys.machine.links]]
link_id = "orders-link"
peer_holder = "{holder}"
peer_address = "127.0.0.1:41003"
exports = ["{exports}"]
"#
)
}
#[test]
fn a_duplicate_link_id_is_refused_at_config() -> TestResult {
let entry = link_entry("importer", "orders");
let mut config = configuration(&format!("{entry}{entry}"))?;
let error = validate(&mut config, None)
.err()
.ok_or("duplicate accepted")?;
assert!(
error
.to_string()
.contains("duplicate link_id 'orders-link'"),
"{error}"
);
Ok(())
}
#[test]
fn a_peer_holder_without_a_recorded_key_is_refused_at_config() -> TestResult {
let mut config = configuration(&link_entry("unknown", "orders"))?;
let error = validate(&mut config, None)
.err()
.ok_or("unknown holder accepted")?;
assert!(
error
.to_string()
.contains("peer_holder 'unknown' has no recorded machine key"),
"{error}"
);
Ok(())
}
#[test]
fn an_export_pattern_is_refused_at_config() -> TestResult {
let mut config = configuration(&link_entry("importer", "orders.*"))?;
let error = validate(&mut config, None)
.err()
.ok_or("pattern accepted")?;
assert!(
error
.to_string()
.contains("exports 'orders.*' is not a placed subject name"),
"{error}"
);
Ok(())
}
#[test]
fn a_named_outbound_link_and_an_addressless_link_are_valid_configuration() -> TestResult {
let entry = link_entry("importer", "orders");
let mut config = configuration(&entry)?;
validate(&mut config, None)?;
let mut config = configuration(&entry.replace("peer_address = \"127.0.0.1:41003\"", ""))?;
validate(&mut config, None)?;
Ok(())
}
struct CountedPeer {
address: SocketAddr,
stopping: Arc<Mutex<Option<SocketAddr>>>,
accepts: Arc<AtomicUsize>,
worker: Option<JoinHandle<Result<(), String>>>,
}
impl CountedPeer {
fn start(
supervisor: liminal_server::server::connection::ConnectionSupervisor,
) -> Result<Self, Box<dyn std::error::Error>> {
let listener = TcpListener::bind("127.0.0.1:0")?;
let address = listener.local_addr()?;
let stopping = Arc::new(Mutex::new(None));
let accepts = Arc::new(AtomicUsize::new(0));
let worker_stop = Arc::clone(&stopping);
let worker_accepts = Arc::clone(&accepts);
let worker = std::thread::spawn(move || {
loop {
let (stream, peer) = listener.accept().map_err(|error| error.to_string())?;
let stop_peer = *worker_stop.lock().map_err(|_| "stop peer lock poisoned")?;
if stop_peer == Some(peer) {
break;
}
worker_accepts.fetch_add(1, Ordering::SeqCst);
supervisor
.spawn_connection(stream)
.map_err(|error| error.to_string())?;
}
Ok(())
});
Ok(Self {
address,
stopping,
accepts,
worker: Some(worker),
})
}
fn finish(mut self) -> Result<usize, Box<dyn std::error::Error>> {
self.stop()?;
Ok(self.accepts.load(Ordering::SeqCst))
}
fn stop(&mut self) -> TestResult {
if self.worker.is_some() {
let interrupt =
socket2::Socket::new(socket2::Domain::IPV4, socket2::Type::STREAM, None)?;
interrupt.bind(&"127.0.0.1:0".parse::<SocketAddr>()?.into())?;
*self
.stopping
.lock()
.map_err(|_| "stop peer lock poisoned")? = interrupt.local_addr()?.as_socket();
interrupt.connect(&self.address.into())?;
}
if let Some(worker) = self.worker.take() {
worker.join().map_err(|_| "counted listener panicked")??;
}
Ok(())
}
}
impl Drop for CountedPeer {
fn drop(&mut self) {
if let Err(error) = self.stop() {
eprintln!("counted listener cleanup failed: {error}");
}
}
}
fn dial(pair: &Federation) -> Result<(LinkDialOutcome, usize), Box<dyn std::error::Error>> {
let counted = CountedPeer::start(pair.importer.supervisor.clone())?;
let mut entry = pair.outbound();
entry.peer_address = Some(counted.address.to_string());
let result = LinkDialer::new(&pair.exporter.host, &pair.exporter.config).open(&entry)?;
Ok((result, counted.finish()?))
}
#[test]
fn a_configured_link_dials_and_opens_the_exporting_end() -> TestResult {
let pair = Federation::start(LiveOpen::Outright)?;
let (outcome, accepts) = dial(&pair)?;
let LinkDialOutcome::Opened(open) = outcome else {
return Err("export end was not opened".into());
};
assert_eq!(open.transport.peer(), &pair.importer.identity());
assert_eq!(open.link_id, "orders-link");
assert_eq!(open.exports, ["orders"]);
assert_eq!(accepts, 1);
drop(open);
pair.shutdown()?;
Ok(())
}
#[test]
fn a_held_open_is_answered_by_its_draft_and_not_retried() -> TestResult {
let pair = Federation::start(LiveOpen::Held)?;
let (outcome, accepts) = dial(&pair)?;
assert!(
matches!(outcome, LinkDialOutcome::Held { grant, draft } if grant == "grant-link" && draft == "draft-link")
);
assert_eq!(accepts, 1);
assert_eq!(pair.live_asks(), 1);
pair.shutdown()?;
Ok(())
}
#[test]
fn a_refused_open_names_its_reason() -> TestResult {
let pair = Federation::start(LiveOpen::Refused)?;
let (outcome, accepts) = dial(&pair)?;
let LinkDialOutcome::Refused { reason } = outcome else {
return Err("open was not refused".into());
};
let reason: liminal_server::auth_lys::AccessRefusal = serde_json::from_str(&reason)?;
assert_eq!(reason.refused.resource_id(), "orders-link");
assert_eq!(reason.refused.action, "link_export");
assert_eq!(accepts, 1);
assert_eq!(pair.live_asks(), 1);
pair.shutdown()?;
Ok(())
}
#[test]
fn two_liminals_boot_with_lys_installed_and_know_each_others_keys() -> TestResult {
let pair = Federation::start(LiveOpen::Outright)?;
assert_ne!(pair.exporter.address, pair.importer.address);
assert_eq!(
pair.exporter.host.recorded_machine("importer"),
Some(pair.importer.identity())
);
assert_eq!(
pair.importer.host.recorded_machine("exporter"),
Some(pair.exporter.identity())
);
let subject = liminal_server::auth_lys::SubjectBinding::link(
"network-a".to_owned(),
"orders-link".to_owned(),
"orders".to_owned(),
)?;
for (server, action) in [
(&pair.exporter, liminal_server::link::LinkAction::Export),
(&pair.importer, liminal_server::link::LinkAction::Import),
] {
assert!(
!server
.host
.rights()
.holds(action, &subject, 0)?
.grant
.is_empty()
);
assert!(
server
.host
.rights()
.holds(action.other(), &subject, 0)
.is_err()
);
let mut subscriber = server.subscribe("orders")?;
let envelope = MessageEnvelope::new(
subscriber.selected_schema(),
CausalContext::independent(),
br#"{"message":"local-message"}"#.to_vec(),
);
server.publish("orders", envelope.clone())?;
assert_eq!(subscriber.receive()?, envelope);
}
pair.shutdown()?;
Ok(())
}
#[test]
fn a_third_machine_dialing_is_refused_at_the_handshake() -> TestResult {
let pair = Federation::start(LiveOpen::Outright)?;
let mut socket = pair.importer.connect_as(&pair.exporter.config.pass_path)?;
let mut third = LinkHandshake::initiator(&local_machine(3), pair.importer.identity())?;
let mut first = [0; NOISE_OVERHEAD];
let length = third.write(&mut first)?;
write_frame(
&mut socket,
&Frame::LinkOpen {
flags: 0,
stream_id: 1,
offer: LinkOffer {
link_id: "orders-link".to_owned(),
end: LinkEnd::Export,
machine: *pair.exporter.identity().ed25519_public(),
handshake: first[..length].to_vec(),
},
},
)?;
let Frame::LinkOpenAck {
stream_id: 1,
outcome: LinkOpenOutcome::Refused { reason },
..
} = read_frame(&mut socket)?
else {
return Err("third machine was not refused".into());
};
assert_eq!(
reason,
"handshake refused: peer is not the machine this link was opened with"
);
assert_eq!(pair.live_asks(), 1);
drop(socket);
pair.shutdown()?;
Ok(())
}
fn wire_pair() -> Result<(WireCarrier, TcpStream), Box<dyn std::error::Error>> {
let listener = TcpListener::bind("127.0.0.1:0")?;
let client = TcpStream::connect(listener.local_addr()?)?;
let (peer, _) = listener.accept()?;
Ok((WireCarrier::new(client, 1)?, peer))
}
#[test]
fn a_wire_carrier_moves_sealed_bytes_and_reports_its_end() -> TestResult {
let (mut carrier, mut peer) = wire_pair()?;
assert_eq!(carrier.carrier(), Carrier::Wire);
carrier.send("orders", b"sealed-out")?;
assert!(
matches!(read_frame(&mut peer)?, Frame::LinkMessage { stream_id: 1, subject, sealed, .. } if subject == "orders" && sealed == b"sealed-out")
);
write_frame(
&mut peer,
&Frame::LinkMessage {
flags: 0,
stream_id: 1,
subject: "orders.created".to_owned(),
sealed: b"sealed-in".to_vec(),
},
)?;
assert_eq!(
carrier.receive()?,
CarrierReceive::Sealed {
subject: "orders.created".to_owned(),
sealed: b"sealed-in".to_vec()
}
);
peer.shutdown(std::net::Shutdown::Write)?;
assert_eq!(
carrier.receive()?,
CarrierReceive::End {
reason: "wire_eof".to_owned()
}
);
assert!(matches!(
carrier.send("orders", b"after-end"),
Err(CarrierError::Ended)
));
let (mut carrier, mut peer) = wire_pair()?;
write_frame(
&mut peer,
&Frame::DisconnectReason {
flags: 0,
reason: liminal::protocol::DisconnectReason::GrantRevoked,
},
)?;
assert_eq!(
carrier.receive()?,
CarrierReceive::End {
reason: "grant_revoked".to_owned()
}
);
assert_eq!(
carrier.receive()?,
CarrierReceive::End {
reason: "grant_revoked".to_owned()
}
);
Ok(())
}
#[test]
fn a_wire_carrier_refuses_an_unplaced_inbound_subject() -> TestResult {
let (mut carrier, mut peer) = wire_pair()?;
write_frame(
&mut peer,
&Frame::LinkMessage {
flags: 0,
stream_id: 1,
subject: "orders.*".to_owned(),
sealed: b"sealed-in".to_vec(),
},
)?;
assert!(matches!(carrier.receive(), Err(CarrierError::Subject(_))));
Ok(())
}
#[test]
fn a_wire_carrier_refuses_a_wrong_stream_and_an_oversize_body_before_reading_it() -> TestResult {
use std::io::Write;
let (mut carrier, mut peer) = wire_pair()?;
write_frame(
&mut peer,
&Frame::LinkMessage {
flags: 0,
stream_id: 2,
subject: "orders".to_owned(),
sealed: b"sealed".to_vec(),
},
)?;
assert!(matches!(
carrier.receive(),
Err(CarrierError::UnexpectedFrame)
));
peer.write_all(&[0x1f, 0, 0, 0, 0, 1, 255, 255, 255, 255])?;
assert!(matches!(
carrier.receive(),
Err(CarrierError::TooLarge { .. })
));
Ok(())
}