use super::{OBSERVATION, Result};
use crate::agent_pass;
use agent_pass::base64url;
use liminal::protocol::{CausalContext, Frame, MessageEnvelope, ProtocolVersion, SchemaId};
use liminal_protocol::credential::WireCredential;
use liminal_server::auth_lys::{
GrantMode, LysPassConsumer,
lys_pass_adapter::{LiveAnswer, LiveLysAsk, LysPassAdapter, MachineLinkRights},
session::ConsumerAdmitter,
};
use liminal_server::config::ServerConfig;
use liminal_server::link::LinkHost;
use liminal_server::link_bridge::BridgeRegistry;
use liminal_server::link_handshake::LocalMachine;
use liminal_server::server::connection::LiminalConnectionServices;
use liminal_server::server::runtime::link_test_support::{self, Configured};
use liminal_server::server::{ConnectionSupervisor, ServerListener};
use liminal_server::services::production::ServiceBindings;
use lys_core::keys::Ed25519Identity;
use serde_json::{Value, json};
use std::collections::BTreeMap;
use std::net::{SocketAddr, TcpStream};
use std::sync::Arc;
use zeroize::Zeroizing;
#[path = "wire.rs"]
pub mod wire;
const ISSUER: &str = "https://carrier-live.example";
const HOLDERS: [&str; 4] = ["exporter", "importer", "relay", "broker"];
#[derive(Debug)]
struct Authority {
refusals: std::sync::atomic::AtomicUsize,
}
impl LiveLysAsk for Authority {
fn ask(
&self,
_pass: &str,
holder: &str,
target: &lys_pass::Target,
_now: u64,
) -> std::result::Result<LiveAnswer, lys_pass::Error> {
Ok(
if holder == "exporter"
&& target.kind == "liminal.link"
&& matches!(target.id.as_str(), "orders-link" | "second-link")
&& target.action == "link_export"
{
if self
.refusals
.fetch_update(
std::sync::atomic::Ordering::AcqRel,
std::sync::atomic::Ordering::Acquire,
|remaining| remaining.checked_sub(1),
)
.is_ok()
{
LiveAnswer::Refused {
refusal: "fixture_open_refused".to_owned(),
}
} else {
LiveAnswer::Outright {
grant: "open-grant".to_owned(),
}
}
} else {
LiveAnswer::Held {
mode: GrantMode::ByDraft,
grant: "unexpected-open".to_owned(),
draft: "held".to_owned(),
}
},
)
}
}
fn pass(
issuer: &Ed25519Identity,
holder: &str,
devices: &[String; 4],
) -> Result<Zeroizing<String>> {
let subjects: Vec<String> = [
"orders.created",
"console.topology",
"net.broker.register",
"net.broker.lookup",
"net.broker.punch",
]
.into_iter()
.map(str::to_owned)
.chain(
devices
.iter()
.flat_map(|device| [format!("net.punch.{device}"), format!("net.relay.{device}")]),
)
.collect();
let mut rights: Vec<_> = subjects.iter().map(|subject| json!({"resource":{"kind":"liminal.subject","id":subject},"actions":["publish","subscribe","call"],"mode":"outright","grant":format!("{holder}-{subject}")})).collect();
for link in ["orders-link", "second-link"] {
rights.push(json!({"resource":{"kind":"liminal.link","id":link},
"actions":["link_import","link_export"],"mode":"outright","grant":format!("{holder}-{link}")}));
}
let signing = agent_pass::Issuer {
url: ISSUER,
key_id: "carrier-live",
identity: issuer,
};
agent_pass::mint(holder, &rights, &signing)
}
pub struct Server {
pub config: ServerConfig,
pub services: Arc<LiminalConnectionServices>,
pub host: Arc<LinkHost>,
pub supervisor: ConnectionSupervisor,
pub registry: BridgeRegistry,
pub bindings: Option<Arc<ServiceBindings>>,
pub links: Option<Configured>,
pub(super) listener: Option<ServerListener>,
admitter: Arc<ConsumerAdmitter<LysPassAdapter>>,
}
impl Server {
pub fn address(&self) -> Result<SocketAddr> {
Ok(self
.listener
.as_ref()
.ok_or("listener stopped")?
.local_addr())
}
pub fn machine(&mut self) -> Result<&mut liminal_server::config::types::MachineConfig> {
Ok(self
.config
.auth
.as_mut()
.and_then(|auth| auth.lys.as_mut())
.and_then(|lys| lys.machine.as_mut())
.ok_or("machine absent")?)
}
pub fn start_links(&mut self) -> Result {
self.links = Some(link_test_support::start(
&self.config,
&self.services,
self.host.clone(),
self.admitter.clone(),
&self.supervisor,
&self.registry,
)?);
Ok(())
}
pub fn connect(&self) -> Result<TcpStream> {
let pass_path = &self
.config
.auth
.as_ref()
.and_then(|auth| auth.lys.as_ref())
.and_then(|lys| lys.machine.as_ref())
.ok_or("machine absent")?
.pass_path;
self.connect_as(pass_path)
}
pub fn connect_as(&self, pass_path: &std::path::Path) -> Result<TcpStream> {
let token = Zeroizing::new(std::fs::read_to_string(pass_path)?);
let mut socket = TcpStream::connect(self.address()?)?;
socket.set_read_timeout(Some(OBSERVATION))?;
wire::write_frame(
&mut socket,
&Frame::Connect {
flags: 0,
min_version: ProtocolVersion::new(1, 0),
max_version: ProtocolVersion::new(1, 1),
auth_token: WireCredential::lys_record(token.as_bytes()),
},
)?;
assert!(matches!(
wire::read_frame(&mut socket)?,
Frame::ConnectAck { .. }
));
Ok(socket)
}
pub fn subscribe(&self, subject: &str) -> Result<TcpStream> {
let mut socket = self.connect()?;
wire::write_frame(
&mut socket,
&Frame::Subscribe {
flags: 0,
stream_id: 1,
channel: subject.to_owned(),
accepted_schemas: Vec::new(),
max_in_flight: 32,
},
)?;
assert!(matches!(
wire::read_frame(&mut socket)?,
Frame::SubscribeAck { .. }
));
Ok(socket)
}
pub fn publish(&self, number: u64) -> Result {
let mut socket = self.connect()?;
wire::write_frame(
&mut socket,
&Frame::Publish {
flags: 0,
stream_id: 1,
channel: "orders.created".to_owned(),
envelope: MessageEnvelope::new(
SchemaId::new([0; 32]),
CausalContext::independent(),
json!({"n":number}).to_string().into_bytes(),
),
idempotency_key: None,
},
)?;
assert!(matches!(
wire::read_frame(&mut socket)?,
Frame::PublishAck { .. }
));
Ok(())
}
pub fn topology(&self) -> Result<Value> {
let mut subscriber = self.subscribe("console.topology")?;
self.services.console().pump(self.services.as_ref())?;
let Frame::Deliver { envelope, .. } = wire::read_frame(&mut subscriber)? else {
return Err("no topology delivery".into());
};
Ok(serde_json::from_slice(&envelope.payload)?)
}
pub fn stop(&mut self) -> Result {
if let Some(links) = self.links.take() {
links.finish()?;
}
if let Some(listener) = self.listener.take() {
listener.shutdown()?;
}
self.supervisor.shutdown();
self.services.conversation_supervisor().shutdown();
self.services.channel_cluster().supervisor().shutdown();
Ok(())
}
}
impl Drop for Server {
fn drop(&mut self) {
if let Err(error) = self.stop() {
eprintln!("carrier fixture shutdown failed: {error}");
}
}
}
pub struct Fixture {
pub servers: Vec<Server>,
pub devices: [String; 4],
pub echo: SocketAddr,
runtime: Option<tokio::runtime::Runtime>,
root: tempfile::TempDir,
}
impl Fixture {
pub fn new() -> Result<Self> {
Self::with_open_refusals(0)
}
pub fn with_open_refusals(refusals: usize) -> Result<Self> {
let root = tempfile::tempdir()?;
let issuer = Ed25519Identity::from_seed(&Zeroizing::new([9; 32]));
let devices = [1, 2, 3, 4].map(|seed| {
hex::encode(Ed25519Identity::from_seed(&Zeroizing::new([seed; 32])).public_key_bytes())
});
let tokens = [
pass(&issuer, HOLDERS[0], &devices)?,
pass(&issuer, HOLDERS[1], &devices)?,
pass(&issuer, HOLDERS[2], &devices)?,
pass(&issuer, HOLDERS[3], &devices)?,
];
let keys=json!({"keys":[{"kty":"OKP","crv":"Ed25519","kid":"carrier-live","x":base64url(&issuer.public_key_bytes()),"alg":"EdDSA","use":"sig"}]}).to_string();
let keys_path = root.path().join("issuer.json");
std::fs::write(&keys_path, &keys)?;
let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(1)
.enable_all()
.build()?;
let adapter = Arc::new(
LysPassAdapter::from_keys_json(&keys, ISSUER.to_owned())?.with_live(Arc::new(
Authority {
refusals: std::sync::atomic::AtomicUsize::new(refusals),
},
)),
);
let admitter = Arc::new(ConsumerAdmitter::new(adapter.clone()));
let mut fixture = Self {
servers: Vec::new(),
devices,
echo: SocketAddr::from(([127, 0, 0, 1], 0)),
runtime: Some(runtime),
root,
};
for (index, token) in tokens.iter().enumerate() {
fixture
.servers
.push(fixture.boot_server(index, token, &keys_path, &adapter, &admitter)?);
}
fixture.echo = fixture.servers[3]
.bindings
.as_ref()
.and_then(|bindings| bindings.echo_address())
.ok_or("broker echo address absent")?;
Ok(fixture)
}
fn server_config(
&self,
index: usize,
token: &str,
keys_path: &std::path::Path,
) -> Result<ServerConfig> {
let holder = HOLDERS[index];
let directory = self.root.path().join(holder);
std::fs::create_dir(&directory)?;
let identity_path = directory.join("machine.key");
let pass_path = directory.join("machine.pass");
std::fs::write(&identity_path, hex::encode([u8::try_from(index + 1)?; 32]))?;
std::fs::write(&pass_path, token.as_bytes())?;
let machines: serde_json::Map<_, _> = HOLDERS
.iter()
.zip(&self.devices)
.map(|(name, key)| ((*name).to_owned(), json!(key)))
.collect();
let mut channels = vec![
json!({"name":"orders.created","durable":true}),
json!({"name":"console.topology","durable":false}),
];
channels.extend(
[
"net.broker.register",
"net.broker.lookup",
"net.broker.punch",
]
.into_iter()
.map(|name| json!({"name":name,"durable":false})),
);
channels.extend(
self.devices
.iter()
.map(|device| json!({"name":format!("net.punch.{device}"),"durable":false})),
);
let config: ServerConfig = serde_json::from_value(
json!({"listen_address":"127.0.0.1:0","health_listen_address":"127.0.0.1:0","persistence_path":directory,"drain_timeout_ms":0,"channels":channels,"routing_rules":[],"auth":{"token":"fixture-bearer","lys":{"issuer":ISSUER,"network_id":"network-a","keys_path":keys_path,"relay":{"enabled":index==2},"broker":if index==3 {json!({"udp_listen":"127.0.0.1:0"})}else{Value::Null},"machine":{"identity_path":identity_path,"pass_path":pass_path,"machines":machines}}}}),
)?;
Ok(config)
}
fn boot_server(
&self,
index: usize,
token: &str,
keys_path: &std::path::Path,
adapter: &Arc<LysPassAdapter>,
admitter: &Arc<ConsumerAdmitter<LysPassAdapter>>,
) -> Result<Server> {
let holder = HOLDERS[index];
let config = self.server_config(index, token, keys_path)?;
let services = Arc::new(LiminalConnectionServices::from_config(&config)?);
let machine = config
.auth
.as_ref()
.and_then(|auth| auth.lys.as_ref())
.and_then(|lys| lys.machine.as_ref())
.ok_or("machine absent")?;
let verified = adapter.verify_offline(token.as_bytes(), "liminal", 0)?;
let peers: BTreeMap<_, _> = HOLDERS
.iter()
.zip(&self.devices)
.filter(|(peer, _)| **peer != holder)
.map(|(peer, key)| {
Ok((
(*peer).to_owned(),
liminal_server::link_handshake::MachineIdentity::new(
hex::decode(key)?
.try_into()
.map_err(|_| "machine key length")?,
),
))
})
.collect::<Result<_>>()?;
let host = Arc::new(LinkHost::new(
LocalMachine::new(Ed25519Identity::from_seed(&Zeroizing::new(
[u8::try_from(index + 1)?; 32],
))),
Box::new(MachineLinkRights::new(adapter, verified)),
peers,
));
assert_eq!(machine.links.len(), 0);
let bindings = if index >= 2 {
Some(Arc::new(
ServiceBindings::configure(
config
.auth
.as_ref()
.and_then(|auth| auth.lys.as_ref())
.ok_or("lys absent")?,
services.clone(),
self.runtime
.as_ref()
.ok_or("runtime absent")?
.handle()
.clone(),
1,
)?
.ok_or("bindings absent")?,
))
} else {
None
};
let supervisor = link_test_support::supervisor(
&config,
services.clone(),
host.clone(),
admitter.clone(),
bindings.clone(),
)?;
let listener = ServerListener::bind(&config, supervisor.clone())?;
Ok(Server {
config,
services,
host,
supervisor,
registry: BridgeRegistry::default(),
bindings,
links: None,
listener: Some(listener),
admitter: admitter.clone(),
})
}
pub fn configure(&mut self, address: Option<String>) -> Result {
let broker = self.servers[3].address()?;
let relay = self.servers[2].address()?;
let machine = self.servers[0].machine()?;
machine
.links
.push(liminal_server::config::types::OutboundLinkConfig {
link_id: "orders-link".to_owned(),
peer_holder: "importer".to_owned(),
peer_address: address,
exports: vec!["orders".to_owned()],
});
machine.carrier_choice.broker_address = Some(broker.to_string());
machine.carrier_choice.broker_echo = Some(self.echo);
machine.carrier_choice.relay_address = Some(relay.to_string());
machine.carrier_choice.relay_holder = Some("relay".to_owned());
Ok(())
}
}
impl Drop for Fixture {
fn drop(&mut self) {
for server in &mut self.servers {
if let Err(error) = server.stop() {
eprintln!("carrier cleanup failed: {error}");
}
}
if let Some(runtime) = self.runtime.take() {
runtime.shutdown_background();
}
}
}