use liminal::channel::ChannelSupervisor;
use liminal::durability::DurableStore;
use lys_pass::drafts::{ApprovedDraft, DraftRequest, Execution, execute_once, request_digest};
use serde_json::json;
use std::collections::{BTreeMap, BTreeSet};
use std::sync::{Arc, Mutex};
use super::executor::{ApprovalReader, OutcomeDelivery, SiteDraftStore};
use super::{BuilderGrants, Contract, SiteAct, SiteChange, SiteError, SiteLimits, SiteSpec};
use crate::auth_lys::live::drafts::{ApprovalEvidence, DraftError};
use crate::services::lifecycle::{Lifecycle, Placer};
pub type TestResult = Result<(), Box<dyn std::error::Error>>;
#[derive(Default)]
pub struct Issuer {
pub grants: Mutex<BTreeMap<String, (String, String, String)>>,
placements: Mutex<BTreeSet<(String, String, String, String)>>,
deliveries: Mutex<BTreeMap<String, String>>,
}
impl std::fmt::Debug for Issuer {
fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
out.debug_struct("FixtureIssuer").finish_non_exhaustive()
}
}
impl Placer for Issuer {
fn place(&self, _: &str, child: (&str, &str), parent: (&str, &str)) -> Result<(), String> {
let schema: serde_json::Value = serde_json::from_str(include_str!(
"../../../../../docs/design/access/schema.json"
))
.map_err(|_| "fixture_schema_invalid")?;
if !schema["kinds"][child.0]["parents"]
.as_array()
.is_some_and(|parents| parents.iter().any(|kind| kind.as_str() == Some(parent.0)))
{
return Err("fixture_placement_refused".into());
}
self.placements
.lock()
.map_err(|_| "fixture_placement_poisoned")?
.insert((
child.0.into(),
child.1.into(),
parent.0.into(),
parent.1.into(),
));
Ok(())
}
}
impl super::effects::SiteEffects for Issuer {
fn agent_owner(&self, agent: &str) -> Result<String, SiteError> {
if !["builder", "second"].contains(&agent) {
return Err(SiteError::Invalid("fixture agent absent"));
}
Ok("person-owner".into())
}
fn bind(
&self,
draft: &str,
site: &str,
agent: &str,
_: u64,
) -> Result<BuilderGrants, SiteError> {
self.agent_owner(agent)?;
let mut grants = self.grants.lock().map_err(|_| SiteError::Store)?;
let ids = BuilderGrants {
plan: super::effects::operation(draft, "plan"),
serve: super::effects::operation(draft, "serve"),
};
grants.insert(ids.plan.clone(), (agent.into(), site.into(), "plan".into()));
grants.insert(
ids.serve.clone(),
(agent.into(), site.into(), "serve".into()),
);
drop(grants);
Ok(ids)
}
fn revoke(&self, _: &str, _: &str, grant: &str) -> Result<(), SiteError> {
self.grants
.lock()
.map_err(|_| SiteError::Store)?
.remove(grant);
Ok(())
}
}
impl ApprovalReader for Issuer {
fn read(&self, draft: &ApprovedDraft) -> Result<ApprovalEvidence, DraftError> {
Ok(ApprovalEvidence {
draft: draft.id.clone(),
digest: draft.request_digest.clone(),
approved_by: vec!["person-owner".into()],
mode: crate::auth_lys::GrantMode::ByDraft,
})
}
}
impl OutcomeDelivery for Issuer {
fn deliver(
&self,
holder: &str,
draft: &ApprovedDraft,
_: &Execution,
) -> Result<(), DraftError> {
let mut deliveries = self.deliveries.lock().map_err(|_| DraftError::Store)?;
if let Some(old) = deliveries.get(&draft.id) {
if old != holder {
return Err(DraftError::ReceiptMismatch);
}
}
deliveries.insert(draft.id.clone(), holder.into());
drop(deliveries);
Ok(())
}
}
pub fn limits() -> Result<SiteLimits, serde_json::Error> {
serde_json::from_value(
json!({"sites":8,"subjects":8,"schema_bytes":16384,"text_bytes":512,
"steps":32,"transitions":128,"drafts":128,"approvers":2,"snapshot_bytes":1_048_576,"request_bytes":65_536}),
)
}
pub fn spec() -> SiteSpec {
SiteSpec {
name: "echo".into(),
purpose: "Echo a validated request.".into(),
subjects: vec!["echo.request".into()],
contracts: BTreeMap::from([(
"echo.request".into(),
Contract {
input_schema_json: "{\"type\":\"object\"}".into(),
output_schema_json: "{\"type\":\"object\"}".into(),
},
)]),
}
}
pub struct Fixture {
pub directory: tempfile::TempDir,
pub lifecycle: Arc<Lifecycle>,
pub issuer: Arc<Issuer>,
pub scheduler: ChannelSupervisor,
pub counts: Arc<liminal::durability::CountingStore<liminal::durability::HaematiteStore>>,
}
impl Fixture {
pub fn new() -> Result<Self, Box<dyn std::error::Error>> {
Self::with_policy(None)
}
pub fn with_policy(
policy: Option<crate::services::lifecycle::templates::TemplatePolicy>,
) -> Result<Self, Box<dyn std::error::Error>> {
let directory = crate::test_fixture::fixture_copy("agt003-empty-store", |path| {
let store = open_store(&path.join("store"))?;
liminal::durability::bridge::block_on(store.flush())??;
Ok(())
})?;
let issuer = Arc::new(Issuer::default());
let scheduler = ChannelSupervisor::new()?;
let lifecycle = Lifecycle::open_bounded(
&[],
directory.path().join("services"),
"fixture-network".into(),
Some(issuer.clone()),
&limits()?,
)?;
let retained = crate::agents::retained::Retained::new(
crate::agents::config::AgentsConfig {
max_event_bytes: 65536,
..crate::agents::config::AgentsConfig::default()
},
scheduler.clone(),
"fixture",
);
let counts = open_store(&directory.path().join("store"))?;
lifecycle.configure_sites(limits()?, counts.clone(), policy, issuer.clone(), retained)?;
Ok(Self {
directory,
lifecycle,
issuer,
scheduler,
counts,
})
}
pub fn execute(
&self,
id: &str,
change: SiteChange,
) -> Result<Execution, Box<dyn std::error::Error>> {
let pass = pass("builder", "agent")?;
let act = SiteAct::prepare("echo".into(), change, &pass)?;
let words = serde_json::to_string(&act)?;
let request = DraftRequest {
operation: id.into(),
grant: "fixture-held-grant".into(),
target: lys_pass::Target::new(
act.change.kind(),
act.permission_id(self.lifecycle.network_id()),
act.change.action(),
)?,
request_digest: request_digest(&words),
words,
};
self.lifecycle.keep_held_site(
id,
&request,
&act,
&pass,
crate::auth_lys::GrantMode::ByDraft,
1,
)?;
let draft = ApprovedDraft {
id: id.into(),
app: "liminal".into(),
grant: request.grant,
target: request.target,
request_digest: request.request_digest,
approved_by: vec!["fixture-owner".into()],
mode: lys_pass::Mode::ByDraft,
words: request.words,
};
let mut store = SiteDraftStore::new(
self.lifecycle.clone(),
self.issuer.clone(),
self.issuer.clone(),
10,
);
Ok(execute_once(&draft, &mut store)?)
}
}
impl Drop for Fixture {
fn drop(&mut self) {
self.scheduler.shutdown();
}
}
pub fn pass(
holder: &str,
kind: &str,
) -> Result<lys_pass::VerifiedPass, Box<dyn std::error::Error>> {
let signer = lys_core::keys::Ed25519Identity::from_seed(&zeroize::Zeroizing::new([65; 32]));
let keys =
json!({"keys":[{"kty":"OKP","crv":"Ed25519","kid":"site-fixture","alg":"EdDSA","use":"sig",
"x":base64url(&signer.public_key_bytes())}]})
.to_string();
let claims = json!({"iss":"https://site.fixture","sub":holder,"aud":"liminal","iat":0,"exp":100,
"holder":{"id":holder,"kind":kind,"responsible":"person-owner"},"rights":[]});
let head = base64url(br#"{"alg":"EdDSA","typ":"JWT","kid":"site-fixture"}"#);
let message = format!("{head}.{}", base64url(&serde_json::to_vec(&claims)?));
let token = format!("{message}.{}", base64url(&signer.sign(message.as_bytes())));
Ok(lys_pass::VerifiedPass::verify(
&token,
&lys_pass::KeySet::from_json(&keys)?,
"https://site.fixture",
"liminal",
1,
)?)
}
fn base64url(bytes: &[u8]) -> String {
const DIGITS: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
let mut result = String::new();
for chunk in bytes.chunks(3) {
let a = chunk[0];
let b = chunk.get(1).copied().unwrap_or(0);
let c = chunk.get(2).copied().unwrap_or(0);
result.push(char::from(DIGITS[usize::from(a >> 2)]));
result.push(char::from(DIGITS[usize::from((a & 3) << 4 | b >> 4)]));
if chunk.len() > 1 {
result.push(char::from(DIGITS[usize::from((b & 15) << 2 | c >> 6)]));
}
if chunk.len() > 2 {
result.push(char::from(DIGITS[usize::from(c & 63)]));
}
}
result
}
pub fn open_store(
path: &std::path::Path,
) -> Result<
Arc<liminal::durability::CountingStore<liminal::durability::HaematiteStore>>,
Box<dyn std::error::Error>,
> {
let options = haematite::RuntimeOptions {
long_wait_interval: std::time::Duration::from_secs(5),
reclaim_start: haematite::ReclaimStart::AtOpen,
};
let database = if path.exists() {
haematite::Database::open(path, options)?
} else {
haematite::Database::create(
haematite::DatabaseConfig {
data_dir: path.into(),
shard_count: 1,
distributed: None,
executor_threads: None,
node_cache_budget: Some(haematite::NodeCacheBudget::Unlimited),
retention: haematite::Retention::reclaim(0),
},
options,
)?
};
Ok(Arc::new(liminal::durability::CountingStore::new(
liminal::durability::HaematiteStore::new(Arc::new(haematite::EventStore::new(database))),
)))
}