use super::*;
use lys_pass::drafts::request_digest;
use std::io::{BufRead, Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::{Arc, Mutex};
type TestResult = Result<(), Box<dyn std::error::Error>>;
pub const LYS_CONTRACT_COMMIT: &str = "819ca500d709c025e9e77a8afe722fe3f134bc37";
pub use super::waiting::tests::pull_limits as limits;
pub fn draft() -> Result<ApprovedDraft, lys_pass::Error> {
Ok(ApprovedDraft {
id: "draft-site".to_owned(),
app: "liminal".to_owned(),
grant: "grant-site".to_owned(),
target: lys_pass::Target::new("liminal.service", "site-a", "create_site")?,
request_digest: request_digest("create the site α"),
approved_by: vec!["fixture-owner".into()],
mode: lys_pass::Mode::ByDraft,
words: "create the site α".to_owned(),
})
}
pub struct Issuer {
pub api: String,
calls: Arc<Mutex<Vec<String>>>,
responses: Arc<Mutex<Vec<(u16, serde_json::Value)>>>,
address: std::net::SocketAddr,
stopping: Arc<std::sync::atomic::AtomicBool>,
check_pass_digest: Arc<Mutex<Option<String>>>,
thread: Option<std::thread::JoinHandle<Result<(), std::io::Error>>>,
}
fn check_body(first: &str, body: &[u8]) -> std::io::Result<()> {
if first.contains("/executed ") {
let value: serde_json::Value = serde_json::from_slice(body)?;
assert_eq!(value.as_object().map(serde_json::Map::len), Some(1));
assert_eq!(value["receipt_digest"].as_str().map(str::len), Some(64));
} else if first.contains("/refused-on-execution ") {
let value: serde_json::Value = serde_json::from_slice(body)?;
assert_eq!(value.as_object().map(serde_json::Map::len), Some(2));
if first.contains("/draft-site/") {
assert_eq!(value["refusal"], "target_retired");
assert_eq!(value["reason"], "The site was retired.");
} else {
assert_eq!(value["refusal"], "site_execution_refused");
assert_eq!(value["reason"], "service construction is not building");
}
} else if first.starts_with("POST /api/grants/check/batch ") {
let value: serde_json::Value = serde_json::from_slice(body)?;
assert_eq!(value["checks"][0]["subject"], "person-operator");
assert_eq!(value["checks"][0]["kind"], "liminal.network");
assert_eq!(value["checks"][0]["id"], "network-a");
assert_eq!(value["checks"][0]["action"], "pull_drafts");
} else if first.starts_with("POST /api/apps/liminal/placements ") {
let value: serde_json::Value = serde_json::from_slice(body)?;
assert!(
value["operation"]
.as_str()
.is_some_and(|operation| !operation.is_empty())
);
let parent = if value["child"]["kind"] == "liminal.subject"
&& value["parent"]["kind"] == "liminal.service"
{
"liminal.service"
} else {
"liminal.network"
};
assert_eq!(value["parent"]["kind"], parent);
assert_eq!(value.as_object().map(serde_json::Map::len), Some(3));
} else {
assert!(first.starts_with("GET /api/product-drafts?app=liminal&state=approved"));
}
Ok(())
}
impl Issuer {
pub fn new(
responses: Vec<(u16, serde_json::Value)>,
) -> Result<Self, Box<dyn std::error::Error>> {
assert_eq!(LYS_CONTRACT_COMMIT.len(), 40);
let listener = TcpListener::bind("127.0.0.1:0")?;
let address = listener.local_addr()?;
let calls = Arc::new(Mutex::new(Vec::new()));
let responses = Arc::new(Mutex::new(responses));
let stopping = Arc::new(std::sync::atomic::AtomicBool::new(false));
let check_pass_digest = Arc::new(Mutex::new(None));
let check_pass = Arc::clone(&check_pass_digest);
let seen = Arc::clone(&calls);
let answers = Arc::clone(&responses);
let stop = Arc::clone(&stopping);
let thread = std::thread::spawn(move || {
loop {
let (mut socket, _) = listener.accept()?;
if stop.load(std::sync::atomic::Ordering::Acquire) {
return Ok(());
}
let mut reader = std::io::BufReader::new(socket.try_clone()?);
let mut first = String::new();
reader.read_line(&mut first)?;
let mut length = 0;
let mut authorised = false;
loop {
let mut line = String::new();
reader.read_line(&mut line)?;
if line == "\r\n" {
break;
}
if line.is_empty() {
return Err(std::io::Error::other("fixture_request_ended"));
}
if let Some((name, value)) = line.split_once(':') {
if name.eq_ignore_ascii_case("content-length") {
length = value.trim().parse().map_err(std::io::Error::other)?;
}
if name.eq_ignore_ascii_case("authorization") {
authorised = if first.starts_with("POST /api/grants/check/batch ") {
value.trim().strip_prefix("Bearer ").is_some_and(|pass| {
check_pass.lock().is_ok_and(|expected| {
expected.as_ref() == Some(&request_digest(pass))
})
})
} else {
value.trim() == "Bearer connector-fixture"
};
}
}
}
assert!(authorised, "connector credential differs");
let mut body = vec![0; length];
reader.read_exact(&mut body)?;
check_body(&first, &body)?;
seen.lock()
.map_err(|_| std::io::Error::other("fixture_lock"))?
.push(first);
let (status, answer) = answers
.lock()
.map_err(|_| std::io::Error::other("fixture_lock"))?
.remove(0);
let bytes = serde_json::to_vec(&answer)?;
write!(
socket,
"HTTP/1.1 {status} fixture\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
bytes.len()
)?;
socket.write_all(&bytes)?;
}
});
Ok(Self {
api: format!("http://{address}/api/"),
calls,
responses,
address,
stopping,
check_pass_digest,
thread: Some(thread),
})
}
pub fn calls(&self) -> Result<usize, Box<dyn std::error::Error>> {
Ok(self.calls.lock().map_err(|_| "fixture lock")?.len())
}
pub fn permit_check(&self, pass: &str) -> TestResult {
*self.check_pass_digest.lock().map_err(|_| "fixture lock")? = Some(request_digest(pass));
Ok(())
}
pub fn finish(mut self) -> TestResult {
assert!(
self.responses
.lock()
.map_err(|_| "fixture lock")?
.is_empty()
);
self.stop()
}
fn stop(&mut self) -> TestResult {
self.stopping
.store(true, std::sync::atomic::Ordering::Release);
let Some(thread) = self.thread.take() else {
return Ok(());
};
drop(TcpStream::connect(self.address)?);
thread.join().map_err(|_| "fixture thread failed")??;
Ok(())
}
}
impl Drop for Issuer {
fn drop(&mut self) {
if let Err(error) = self.stop() {
eprintln!("fixture_shutdown_failed: {error}");
}
}
}
pub fn page(draft: &ApprovedDraft, next: Option<&str>) -> serde_json::Value {
serde_json::json!({"drafts":[draft], "next":next, "total":1})
}
pub fn closed(draft: &ApprovedDraft, state: &str) -> serde_json::Value {
serde_json::json!({"draft":draft.id,"state":state,"index":1,
"tree_size":2,"leaf_hash":"a".repeat(64)})
}
#[derive(serde::Serialize, serde::Deserialize, Default)]
struct TargetRecord {
prepared: Option<ApprovedDraft>,
transitions: Vec<ApprovalTransition>,
receipt: Option<Execution>,
acknowledged: bool,
deliveries: usize,
}
pub struct TargetStore {
path: std::path::PathBuf,
retired: bool,
approval_mode: crate::auth_lys::GrantMode,
approvers: Vec<String>,
}
impl TargetStore {
pub fn new(path: std::path::PathBuf) -> Self {
Self {
path,
retired: false,
approval_mode: crate::auth_lys::GrantMode::ByDraft,
approvers: vec!["person-approver".to_owned()],
}
}
fn read(&self) -> Result<TargetRecord, DraftError> {
match std::fs::read(&self.path) {
Ok(bytes) => serde_json::from_slice(&bytes).map_err(|_| DraftError::Store),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
Ok(TargetRecord::default())
}
Err(_) => Err(DraftError::Store),
}
}
fn write(&self, record: &TargetRecord) -> Result<(), DraftError> {
let bytes = serde_json::to_vec(record).map_err(|_| DraftError::Store)?;
let temporary = self.path.with_extension("new");
let mut file = std::fs::File::create(&temporary).map_err(|_| DraftError::Store)?;
file.write_all(&bytes)
.and_then(|()| file.sync_all())
.map_err(|_| DraftError::Store)?;
std::fs::rename(temporary, &self.path).map_err(|_| DraftError::Store)?;
let parent = self.path.parent().ok_or(DraftError::Store)?;
std::fs::File::open(parent)
.and_then(|dir| dir.sync_all())
.map_err(|_| DraftError::Store)
}
}
impl AtomicDraftStore for TargetStore {
fn receipt(&self, draft: &str) -> Result<Option<Execution>, DraftError> {
if draft != "draft-site" {
return Err(DraftError::ReceiptMismatch);
}
Ok(self.read()?.receipt)
}
fn approval(&self, draft: &ApprovedDraft) -> Result<ApprovalEvidence, DraftError> {
Ok(ApprovalEvidence {
draft: draft.id.clone(),
digest: draft.request_digest.clone(),
approved_by: self.approvers.clone(),
mode: self.approval_mode,
})
}
fn commit(
&mut self,
draft: &ApprovedDraft,
transition: &ApprovalTransition,
) -> Result<Execution, DraftError> {
let mut record = self.read()?;
if let Some(receipt) = record.receipt {
return Ok(receipt);
}
let receipt = if self.retired {
Execution::RefusedOnExecution {
request_digest: draft.request_digest.clone(),
refusal: "target_retired".to_owned(),
reason: "The site was retired.".to_owned(),
}
} else {
Execution::Executed {
request_digest: draft.request_digest.clone(),
receipt_digest: request_digest(
&serde_json::to_string(transition).map_err(|_| DraftError::Store)?,
),
}
};
record.transitions.push(transition.clone());
record.prepared = Some(draft.clone());
record.receipt = Some(receipt.clone());
self.write(&record)?;
Ok(receipt)
}
fn acknowledged(&self, draft: &str) -> Result<bool, DraftError> {
if draft != "draft-site" {
return Err(DraftError::ReceiptMismatch);
}
Ok(self.read()?.acknowledged)
}
fn acknowledge(&mut self, draft: &str, receipt: &Execution) -> Result<(), DraftError> {
if draft != "draft-site" {
return Err(DraftError::ReceiptMismatch);
}
let mut record = self.read()?;
if record.receipt.as_ref() != Some(receipt) {
return Err(DraftError::ReceiptMismatch);
}
record.acknowledged = true;
self.write(&record)
}
fn pending_acknowledgments(
&self,
limit: NonZeroUsize,
) -> Result<Vec<ApprovedDraft>, DraftError> {
let record = self.read()?;
let mut pending = if record.acknowledged {
vec![]
} else {
record.prepared.into_iter().collect::<Vec<_>>()
};
pending.truncate(limit.get());
Ok(pending)
}
fn deliver_outcome(
&mut self,
draft: &ApprovedDraft,
receipt: &Execution,
) -> Result<(), DraftError> {
let mut record = self.read()?;
if draft.id != "draft-site" || record.receipt.as_ref() != Some(receipt) {
return Err(DraftError::ReceiptMismatch);
}
if record.deliveries == 0 {
record.deliveries = 1;
self.write(&record)?;
}
Ok(())
}
}
#[tokio::test]
async fn explicit_pull_commits_once_and_replay_does_not_acknowledge_twice() -> TestResult {
let dir = tempfile::tempdir()?;
let draft = draft()?;
let issuer = Issuer::new(vec![
(200, page(&draft, None)),
(
200,
serde_json::json!({
"draft":draft.id,"state":"executed","index":1,"tree_size":2,"leaf_hash":"a".repeat(64)
}),
),
(200, page(&draft, None)),
])?;
let mut puller = DraftPuller::new(
&issuer.api,
TargetStore::new(dir.path().join("site.json")),
limits()?,
)?;
let first = puller
.pull("connector-fixture", PullSignal::Startup, 100)
.await?;
assert_eq!(
(first.executed, first.replayed, first.acknowledged),
(1, 0, 1)
);
let second = puller
.pull("connector-fixture", PullSignal::Console, 200)
.await?;
assert_eq!(
(second.executed, second.replayed, second.acknowledged),
(0, 1, 0)
);
assert_eq!(puller.executor.store.read()?.transitions.len(), 1);
assert_eq!(puller.executor.store.read()?.deliveries, 1);
assert_eq!(issuer.calls()?, 3);
issuer.finish()
}
#[tokio::test]
async fn retired_target_is_refused_once_and_receipt_survives_reopen() -> TestResult {
let dir = tempfile::tempdir()?;
let draft = draft()?;
let issuer = Issuer::new(vec![
(200, page(&draft, None)),
(200, closed(&draft, "refused_on_execution")),
(200, page(&draft, None)),
])?;
let path = dir.path().join("site.json");
let mut store = TargetStore::new(path.clone());
store.retired = true;
let mut puller = DraftPuller::new(&issuer.api, store, limits()?)?;
assert_eq!(
puller
.pull("connector-fixture", PullSignal::Console, 1)
.await?
.refused,
1
);
drop(puller);
let mut reopened = DraftPuller::new(&issuer.api, TargetStore::new(path), limits()?)?;
assert_eq!(
reopened
.pull("connector-fixture", PullSignal::Console, 2)
.await?
.replayed,
1
);
assert_eq!(reopened.executor.store.read()?.transitions.len(), 1);
assert_eq!(issuer.calls()?, 3);
issuer.finish()
}
#[tokio::test]
async fn failed_acknowledgment_preserves_receipt_for_an_explicit_retry() -> TestResult {
let dir = tempfile::tempdir()?;
let draft = draft()?;
let issuer = Issuer::new(vec![
(200, page(&draft, None)),
(
503,
serde_json::json!({"refusal":"fixture_issuer_unavailable",
"reason":"The fixture issuer is unavailable."}),
),
(200, closed(&draft, "executed")),
(200, serde_json::json!({"drafts":[],"next":null,"total":0})),
])?;
let mut puller = DraftPuller::new(
&issuer.api,
TargetStore::new(dir.path().join("site.json")),
limits()?,
)?;
assert!(
puller
.pull("connector-fixture", PullSignal::Startup, 1)
.await
.is_err()
);
assert_eq!(puller.executor.store.read()?.transitions.len(), 1);
assert_eq!(
puller
.pull("connector-fixture", PullSignal::Console, 2)
.await?
.replayed,
1
);
assert_eq!(puller.executor.store.read()?.transitions.len(), 1);
assert_eq!(issuer.calls()?, 4);
issuer.finish()
}
#[tokio::test]
async fn two_person_approval_requires_two_distinct_authenticated_people() -> TestResult {
for people in [vec!["person-one"], vec!["person-one", "person-one"]] {
let dir = tempfile::tempdir()?;
let draft = draft()?;
let issuer = Issuer::new(vec![(200, page(&draft, None))])?;
let mut store = TargetStore::new(dir.path().join("site.json"));
store.approval_mode = crate::auth_lys::GrantMode::ByTwo;
store.approvers = people.into_iter().map(str::to_owned).collect();
let mut puller = DraftPuller::new(&issuer.api, store, limits()?)?;
assert!(matches!(
puller
.pull("connector-fixture", PullSignal::Console, 1)
.await,
Err(DraftError::ApprovalUnavailable)
));
assert!(puller.executor.store.read()?.transitions.is_empty());
assert!(puller.executor.store.read()?.receipt.is_none());
assert_eq!(issuer.calls()?, 1);
issuer.finish()?;
}
Ok(())
}