#[path = "../../liminal-server/src/test_persistence.rs"]
mod test_persistence;
use std::error::Error;
use std::fs::File;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::mpsc;
use std::thread::{self, JoinHandle};
use ed25519_dalek::{Signer, SigningKey};
use liminal::protocol::{Frame, ProtocolVersion, decode, encode, encoded_len};
use liminal_server::auth_pass::WirePassV1;
use std::time::{SystemTime, UNIX_EPOCH};
const PARTICIPANT: &str = r"
[participant]
wire_frame_limit = 65536
attach_receipt_ttl_ms = 60000
receipt_provenance_ttl_ms = 600000
live_receipt_server_report_threshold = 1024
max_live_attach_receipts_per_participant = 8
receipt_provenance_server_report_threshold = 4096
receipt_provenance_per_conversation_report_threshold = 256
max_receipt_provenance_per_participant = 64
max_retired_identity_slots_server = 1024
identity_slots = 4
observer_recovery_max_entries = 64
max_semantic_conversations_per_connection = 32
max_ordinary_record_entries = 1
max_ordinary_record_bytes = 131072
max_generated_marker_entries = 1
max_generated_marker_bytes = 4096
mandatory_transaction_bound_entries = 4
mandatory_transaction_bound_bytes = 16384
full_recovery_claim_entries = 4
full_recovery_claim_bytes = 16384
retained_capacity_entries = 2048
retained_capacity_bytes = 16777216
max_retained_record_rows = 1024
closure_episode_churn_limit = 1024
";
fn record_stage(stage: &str) -> Result<(), Box<dyn Error>> {
let time = SystemTime::now().duration_since(UNIX_EPOCH)?;
eprintln!("PROOF_STAGE stage={stage} unix_ns={}", time.as_nanos());
Ok(())
}
struct RunningServer {
child: Child,
reader: Option<JoinHandle<Result<(), String>>>,
log: PathBuf,
}
impl RunningServer {
fn start(root: &Path, verifying_key: &str) -> Result<(Self, SocketAddr), Box<dyn Error>> {
let wire_reservation = TcpListener::bind("127.0.0.1:0")?;
let health_reservation = TcpListener::bind("127.0.0.1:0")?;
let wire = wire_reservation.local_addr()?;
let health = health_reservation.local_addr()?;
let config = root.join("server.toml");
std::fs::write(
&config,
format!(
"listen_address = \"{wire}\"\nhealth_listen_address = \"{health}\"\n\
drain_timeout_ms = 5000\nrouting_rules = []\nchannels = []\n\
{PARTICIPANT}\n[auth]\ntoken = \"proof-bearer\"\n\
[auth.pass]\nregistry_verifying_key = \"{verifying_key}\"\n\
maximum_clock_skew_seconds = 0\n"
),
)?;
let log = root.join("server-stderr.log");
let mut log_writer = File::create(&log)?;
drop(wire_reservation);
drop(health_reservation);
record_stage("spawn_requested")?;
let child = Command::new(env!("CARGO_BIN_EXE_liminal"))
.arg("--config")
.arg(config)
.env("LIMINAL_PERSISTENCE_PATH", root)
.env(
"RUST_LOG",
"info,liminal_server::server::self_streams=debug",
)
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::piped())
.spawn()?;
record_stage("spawn_returned")?;
let mut server = Self {
child,
reader: None,
log,
};
let stderr = server.child.stderr.take().ok_or("server stderr absent")?;
let (ready, started) = mpsc::sync_channel(1);
server.reader = Some(thread::spawn(move || {
let mut ready = Some(ready);
for line in BufReader::new(stderr).lines() {
let line = line.map_err(|error| format!("server stderr read failed: {error}"))?;
writeln!(log_writer, "{line}")
.map_err(|error| format!("server stderr log failed: {error}"))?;
if line.contains("liminal server started") {
if let Some(ready) = ready.take() {
ready
.send(())
.map_err(|error| format!("startup receiver closed: {error}"))?;
}
}
}
if ready.is_some() {
return Err("server exited before its startup signal".to_owned());
}
Ok(())
}));
started.recv().map_err(|error| {
format!(
"ordinary server startup failed: {error}; stderr: {}",
std::fs::read_to_string(&server.log)
.unwrap_or_else(|read_error| format!("stderr log read failed: {read_error}"))
)
})?;
record_stage("startup_observed")?;
Ok((server, wire))
}
fn join_reader(&mut self) -> Result<(), Box<dyn Error>> {
if let Some(reader) = self.reader.take() {
reader
.join()
.map_err(|_| "server stderr reader panicked")??;
}
Ok(())
}
fn stop(mut self) -> Result<(), Box<dyn Error>> {
record_stage("stop_requested")?;
let signalled = Command::new("/bin/kill")
.arg("-TERM")
.arg(self.child.id().to_string())
.status()?;
if !signalled.success() {
return Err(format!("owned server termination refused: {signalled}").into());
}
let status = self.child.wait()?;
record_stage("child_reaped")?;
self.join_reader()?;
record_stage("reader_joined")?;
eprintln!("{}", std::fs::read_to_string(&self.log)?);
if !status.success() {
return Err(format!(
"ordinary server shutdown failed: {status}; stderr: {}",
std::fs::read_to_string(&self.log)?
)
.into());
}
let log = std::fs::read_to_string(&self.log)?;
assert!(
log.lines().any(|line| {
line.contains("shutdown sequence complete") && line.contains("clean=true")
}),
"ordinary server did not confirm clean shutdown: {log}"
);
Ok(())
}
}
impl Drop for RunningServer {
fn drop(&mut self) {
match self.child.try_wait() {
Ok(Some(_)) => {}
result => {
if let Err(error) = result {
eprintln!("server cleanup status failed: {error}");
}
if let Err(error) = self.child.kill() {
eprintln!("owned server cleanup kill failed: {error}");
}
if let Err(error) = self.child.wait() {
eprintln!("owned server cleanup wait failed: {error}");
}
}
}
if let Err(error) = self.join_reader() {
eprintln!("server cleanup reader failed: {error}");
}
}
}
struct Peer {
stream: TcpStream,
buffer: Vec<u8>,
}
impl Peer {
fn connect(address: SocketAddr, pass: Vec<u8>) -> Result<Self, Box<dyn Error>> {
record_stage("connect_requested")?;
let mut peer = Self {
stream: TcpStream::connect(address)?,
buffer: Vec::new(),
};
peer.stream.set_nodelay(true)?;
peer.send(&Frame::Connect {
flags: 0,
min_version: ProtocolVersion::new(1, 0),
max_version: ProtocolVersion::new(1, 0),
auth_token: pass,
})?;
assert!(matches!(peer.read()?, Frame::ConnectAck { .. }));
record_stage("connect_ack")?;
Ok(peer)
}
fn send(&mut self, frame: &Frame) -> Result<(), Box<dyn Error>> {
let mut bytes = vec![0; encoded_len(frame)?];
let written = encode(frame, &mut bytes)?;
self.stream
.write_all(bytes.get(..written).ok_or("invalid encoded frame length")?)?;
Ok(())
}
fn read(&mut self) -> Result<Frame, Box<dyn Error>> {
loop {
match decode(&self.buffer) {
Ok((frame, consumed)) => {
self.buffer.drain(..consumed);
return Ok(frame);
}
Err(
liminal::protocol::ProtocolError::IncompleteHeader { .. }
| liminal::protocol::ProtocolError::TruncatedPayload { .. },
) => {
let mut chunk = [0; 8192];
let count = self.stream.read(&mut chunk)?;
if count == 0 {
return Err("ordinary wire closed before its frame arrived".into());
}
if self.buffer.len() + count > 65_536 {
return Err("ordinary wire proof frame exceeds its bound".into());
}
self.buffer.extend_from_slice(&chunk[..count]);
}
Err(error) => return Err(error.into()),
}
}
}
}
fn signed_pass(key: &SigningKey, live: &str) -> Result<Vec<u8>, Box<dyn Error>> {
let now = SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs();
let mut pass = WirePassV1 {
participant: b"self-stream-access-proof".to_vec(),
public_key: key.verifying_key().to_bytes(),
conversations: Vec::new(),
live: live.to_owned(),
may_enroll: false,
issued_at: now.saturating_sub(60),
expires_at: now.checked_add(3600).ok_or("pass expiry overflow")?,
signature: [0; 64],
};
let unsigned = pass
.canonical_unsigned_bytes()
.map_err(|error| format!("unsigned pass encoding failed: {error:?}"))?;
pass.signature = key.sign(&unsigned).to_bytes();
pass.canonical_bytes()
.map_err(|error| format!("signed pass encoding failed: {error:?}").into())
}
fn subscribe(peer: &mut Peer, stream_id: u32, channel: &str) -> Result<Frame, Box<dyn Error>> {
record_stage(&format!("subscribe_requested_{stream_id}"))?;
peer.send(&Frame::Subscribe {
flags: 0,
stream_id,
channel: channel.to_owned(),
accepted_schemas: Vec::new(),
max_in_flight: 8,
})?;
let reply = peer.read()?;
record_stage(&format!("subscribe_reply_{stream_id}"))?;
Ok(reply)
}
#[test]
fn verified_live_prefix_denies_and_allows_both_self_streams() -> Result<(), Box<dyn Error>> {
record_stage("case_entered")?;
let key = SigningKey::from_bytes(&[42; 32]);
let root = test_persistence::directory()?;
let (server, address) =
RunningServer::start(root.path(), &hex::encode(key.verifying_key().to_bytes()))?;
let proof = (|| -> Result<(), Box<dyn Error>> {
let mut denied = Peer::connect(address, signed_pass(&key, "ordinary")?)?;
for (stream_id, channel) in [(1, "liminal.cost"), (2, "liminal.events")] {
match subscribe(&mut denied, stream_id, channel)? {
Frame::SubscribeError {
stream_id: actual,
reason_code,
message,
..
} => {
assert_eq!(actual, stream_id);
assert_eq!(reason_code, 0x0201);
assert_eq!(
message.as_deref(),
Some(format!("subscribe refused for channel {channel}").as_str())
);
}
frame => {
return Err(
format!("excluded live prefix admitted {channel}: {frame:?}").into(),
);
}
}
}
denied.send(&Frame::new_ping(0)?)?;
assert!(matches!(denied.read()?, Frame::Pong { .. }));
let mut allowed = Peer::connect(address, signed_pass(&key, "liminal.")?)?;
for (stream_id, channel) in [(3, "liminal.cost"), (4, "liminal.events")] {
match subscribe(&mut allowed, stream_id, channel)? {
Frame::SubscribeAck {
stream_id: actual,
subscription_id,
..
} => {
assert_eq!(actual, stream_id);
assert!(subscription_id > 0);
}
frame => {
return Err(format!("included live prefix refused {channel}: {frame:?}").into());
}
}
}
Ok(())
})();
record_stage("wire_checks_finished")?;
let stopped = server.stop();
proof?;
stopped?;
record_stage("server_stopped")?;
root.close()?;
record_stage("fixture_removed")?;
Ok(())
}