use std::collections::BTreeSet;
use std::sync::{Arc, Mutex};
use haematite::{Database, DatabaseConfig, EventStore, Retention};
use liminal::durability::{CountingStore, DurableStore, HaematiteStore, bridge::block_on};
use lys_core::keys::Ed25519Identity;
use lys_pass::drafts::{ApprovedDraft, request_digest};
use serde_json::json;
use super::config::AgentsConfig;
use super::records::{AgentAct, AgentEvent};
use super::registry::AgentRegistry;
use crate::auth_lys::lys_pass_adapter::LysPassAdapter;
use crate::auth_lys::session::{ConsumerAdmitter, LysAdmitter, LysSession};
use crate::services::lifecycle::Placer;
pub(super) type TestResult = Result<(), Box<dyn std::error::Error>>;
pub(super) struct Fixture {
pub counts: Arc<CountingStore<HaematiteStore>>,
directory: tempfile::TempDir,
}
impl Fixture {
pub fn new() -> Result<Self, Box<dyn std::error::Error>> {
let directory = crate::test_fixture::fixture_copy("agt002-empty-store", |path| {
let database = Database::create(
DatabaseConfig {
data_dir: path.to_path_buf(),
shard_count: 1,
distributed: None,
executor_threads: None,
node_cache_budget: Some(haematite::NodeCacheBudget::Unlimited),
retention: Retention::reclaim(0),
},
store_options(),
)?;
let store = HaematiteStore::new(Arc::new(EventStore::new(database)));
block_on(store.flush())??;
drop(store);
Ok(())
})?;
Self::at(directory)
}
pub fn at(directory: tempfile::TempDir) -> Result<Self, Box<dyn std::error::Error>> {
let database = Database::open(directory.path(), store_options())?;
Ok(Self {
counts: Arc::new(CountingStore::new(HaematiteStore::new(Arc::new(
EventStore::new(database),
)))),
directory,
})
}
pub fn close(self) -> tempfile::TempDir {
let Self { counts, directory } = self;
drop(counts);
directory
}
pub fn registry(&self) -> Result<Arc<AgentRegistry>, super::error::AgentError> {
let store: Arc<dyn DurableStore> = self.counts.clone();
AgentRegistry::open(
store,
AgentsConfig::default(),
"fixture-network".into(),
Some(Arc::new(FixtureIssuer::default())),
None,
)
}
}
#[derive(Debug, Default)]
pub(super) struct FixtureIssuer(Mutex<BTreeSet<(String, String, String, String)>>);
impl Placer for FixtureIssuer {
fn place(
&self,
operation: &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 operation.is_empty()
|| !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.0
.lock()
.map_err(|_| "fixture_issuer_lock_poisoned")?
.insert((
child.0.into(),
child.1.into(),
parent.0.into(),
parent.1.into(),
));
Ok(())
}
}
pub(super) fn approved(
draft: &str,
name: &str,
holder: &str,
change: AgentEvent,
) -> Result<ApprovedDraft, Box<dyn std::error::Error>> {
let action = match &change {
AgentEvent::Create(_) => "create_agent",
AgentEvent::Start => "start",
AgentEvent::Stop => "stop",
AgentEvent::Retire => "retire",
_ => return Err("not a deliberate fixture act".into()),
};
let words = serde_json::to_string(&AgentAct {
name: name.into(),
holder: holder.into(),
owner: holder.into(),
at_ms: 1,
change,
})?;
Ok(ApprovedDraft {
id: draft.into(),
app: "liminal".into(),
grant: "fixture-approved-grant".into(),
target: lys_pass::Target::new("liminal.agent", name, action)?,
request_digest: request_digest(&words),
approved_by: vec!["fixture-owner".into()],
mode: lys_pass::Mode::ByDraft,
words,
})
}
pub(super) fn session(holder: &str) -> Result<Box<dyn LysSession>, Box<dyn std::error::Error>> {
let (keys, pass) = signed_pass(holder)?;
let adapter = LysPassAdapter::from_keys_json(&keys, "https://agent.fixture".into())?;
Ok(ConsumerAdmitter::new(Arc::new(adapter)).admit(pass.as_bytes(), now()?)?)
}
pub(super) fn signed_pass(holder: &str) -> Result<(String, String), Box<dyn std::error::Error>> {
let issuer = Ed25519Identity::from_seed(&zeroize::Zeroizing::new([44; 32]));
let keys = json!({"keys":[{"kty":"OKP","crv":"Ed25519","kid":"agent-fixture", "alg":"EdDSA","use":"sig",
"x":base64url(&issuer.public_key_bytes())}]}).to_string();
let now = now()?;
let mut rights = Vec::new();
for id in ["agents.a.inbox", "agents.b.inbox", "console.topology"] {
rights.push(
json!({"resource":{"kind":"liminal.subject","id":id}, "actions":["subscribe","publish"],
"mode":"outright", "grant":"fixture-grant"}),
);
}
for id in ["a", "b"] {
rights.push(
json!({"resource":{"kind":"liminal.agent","id":id}, "actions":["message","watch"],
"mode":"outright", "grant":"fixture-agent-grant"}),
);
}
let claims = json!({"iss":"https://agent.fixture","sub":holder,"aud":"liminal","iat":now,
"exp":now.checked_add(600).ok_or("fixture expiry overflow")?,"rights":rights,
"holder":{"id":holder,"kind":"agent","responsible":"fixture-owner"}});
let header = base64url(br#"{"alg":"EdDSA","typ":"JWT","kid":"agent-fixture"}"#);
let signed = format!("{header}.{}", base64url(&serde_json::to_vec(&claims)?));
Ok((
keys,
format!("{signed}.{}", base64url(&issuer.sign(signed.as_bytes()))),
))
}
pub(super) fn now() -> Result<u64, Box<dyn std::error::Error>> {
Ok(std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)?
.as_secs())
}
fn base64url(bytes: &[u8]) -> String {
const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
let mut encoded = 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);
encoded.push(char::from(ALPHABET[usize::from(a >> 2)]));
encoded.push(char::from(ALPHABET[usize::from((a & 3) << 4 | b >> 4)]));
if chunk.len() > 1 {
encoded.push(char::from(ALPHABET[usize::from((b & 15) << 2 | c >> 6)]));
}
if chunk.len() > 2 {
encoded.push(char::from(ALPHABET[usize::from(c & 63)]));
}
}
encoded
}
fn store_options() -> haematite::RuntimeOptions {
haematite::RuntimeOptions {
long_wait_interval: std::time::Duration::from_secs(5),
reclaim_start: haematite::ReclaimStart::AtOpen,
}
}