use super::{
config::Config,
http::{Request, Response},
route::{Site, answer},
signin::{Door, Issuer, SignedIn, Started},
};
use lys_pass::Error;
use serde_json::{Value, json};
use std::{
io::{BufRead, BufReader, Write},
os::unix::net::UnixStream,
sync::{Arc, mpsc},
time::Duration,
};
pub type Result<T = ()> = anyhow::Result<T>;
pub const ORIGIN: &str = "https://studio.example";
pub const PERSON: &str = "person-clinician";
#[path = "assistant_record_fixture.rs"]
pub mod records;
#[derive(Default)]
pub struct Lys {
pub decisions: std::sync::atomic::AtomicUsize,
pub after_decision: std::sync::Mutex<Option<Box<dyn FnOnce() + Send>>>,
pub person: Option<String>,
}
impl Issuer for Lys {
fn assistant_identity(
&self,
agent: &str,
pass: &str,
store: &str,
) -> std::result::Result<String, crate::serve::storage::Failure> {
let keys = keys().map_err(|_| {
crate::serve::storage::Failure::new("fixture_keys_invalid", "fixture keyset")
})?;
super::assistant_lys::identity(
&keys,
"https://lys.example",
agent,
pass,
store,
crate::serve::pass::now(),
)
}
fn assistant_draft(
&self,
pass: &str,
draft: &str,
_within: Duration,
) -> std::result::Result<Value, crate::serve::storage::Failure> {
assert_eq!(pass, "person-pass");
let (agent, store) = if draft == "draft-pharmacy" {
("agent-pharmacy", "pharmacy")
} else {
("agent-assistant", "wards")
};
Ok(
json!({"id":draft,"app":"haematite","holder":agent,"responsible":PERSON,"target":{"kind":"haematite.database","id":store,"action":"agent_permission"},"request_digest":"a".repeat(64)}),
)
}
fn assistant_decide(
&self,
pass: &str,
draft: &Value,
decision: &str,
_within: Duration,
) -> std::result::Result<Value, crate::serve::storage::Failure> {
assert_eq!(pass, "person-pass");
let draft = draft["id"]
.as_str()
.ok_or_else(|| crate::serve::storage::Failure::new("fixture_draft_invalid", "id"))?;
if matches!(draft, "draft-unavailable" | "draft-malformed") {
return Err(crate::serve::storage::Failure::new(
"lys_unavailable",
"Lys did not return a complete answer",
));
}
if draft == "draft-refused" {
return Err(crate::serve::storage::Failure::new(
"draft_not_pending",
"draft is not pending",
));
}
self.decisions
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
let action = self
.after_decision
.lock()
.map_err(|_| {
crate::serve::storage::Failure::new("fixture_poisoned", "decision action")
})?
.take();
if let Some(action) = action {
action();
}
Ok(json!({"draft":draft,"state":if decision=="approve" {"approved"}else{"refused"}}))
}
fn assistant_url(&self) -> std::result::Result<&str, crate::serve::storage::Failure> {
Ok("https://lys.example")
}
fn start(&self) -> std::result::Result<Started, Error> {
Ok(Started {
url: "https://lys.example/signin".into(),
state: "state".into(),
verifier: "verifier".into(),
browser: "browser".into(),
})
}
fn finish(
&self,
_: &str,
_: &str,
_: &str,
_: &Started,
) -> std::result::Result<SignedIn, Error> {
Ok(SignedIn {
token: "person-pass".into(),
holder: self.person.clone().unwrap_or_else(|| PERSON.into()),
expires_at: crate::serve::pass::now() + 3600,
refresh: Some("refresh".into()),
})
}
fn refresh(&self, _: &str) -> std::result::Result<SignedIn, Error> {
Err(Error::Expired)
}
}
pub struct Fixture {
listener: Option<super::assistant_socket::Listener>,
pub files: tempfile::TempDir,
pub door: Arc<Door<Lys>>,
pub site: Arc<Site>,
pub cookie: String,
pub config: Config,
}
impl Fixture {
pub fn new(mut bounds: Value) -> Result<Self> {
let files = tempfile::tempdir()?;
let maximum = bounds
.as_object_mut()
.ok_or_else(|| anyhow::anyhow!("bounds object"))?
.remove("door_max_connections")
.unwrap_or_else(|| json!(64));
let timeout = bounds
.as_object_mut()
.ok_or_else(|| anyhow::anyhow!("bounds"))?
.remove("door_timeout_ms")
.unwrap_or_else(|| json!(30_000));
let person = bounds
.as_object_mut()
.ok_or_else(|| anyhow::anyhow!("bounds"))?
.remove("session_person")
.and_then(|p| p.as_str().map(str::to_owned));
let mut assistant = json!({"socket":files.path().join("assistant.sock")});
for (key, value) in bounds
.as_object()
.ok_or_else(|| anyhow::anyhow!("bounds object"))?
{
assistant[key] = value.clone();
}
let config = Config::parse(&serde_json::to_vec(&json!({
"listen":"127.0.0.1:0", "origin":ORIGIN, "files":files.path(),"max_connections":maximum,"timeout_ms":timeout,
"stores":{"wards":files.path().join("store.sock"),"other":files.path().join("other.sock")},
"lys":{"issuer":"https://lys.example","client_id":"haematite","secret_file":files.path().join("secret"),"redirect":format!("{ORIGIN}/callback")},
"assistant":assistant
}))?).map_err(anyhow::Error::msg)?;
let door = Arc::new(Door::new(Lys {
person,
..Lys::default()
}));
let listener = super::assistant_socket::Listener::start(&door, &config)?;
let begun = door.begin(crate::serve::pass::now())?;
let cookie = door.callback("code", "state", &begun.browser, crate::serve::pass::now())?;
Ok(Self {
listener,
files,
door,
site: Arc::new(config.site()),
cookie,
config,
})
}
pub fn request(&self, method: &str, suffix: &str, body: &Value) -> Request {
let (path, query) = suffix.split_once('?').unwrap_or((suffix, ""));
Request {
method: method.into(),
path: format!("/api/assistant/wards{path}"),
query: query.into(),
headers: vec![
("origin".into(), ORIGIN.into()),
("cookie".into(), format!("haem_session={}", self.cookie)),
],
body: body.to_string().into_bytes(),
}
}
pub fn answer(&self, method: &str, suffix: &str, body: &Value) -> Response {
answer(
&self.door,
&self.site,
&self.request(method, suffix, body),
crate::serve::pass::now(),
)
}
pub fn seat(&self, pass: &str, store: &str) -> Result<Seat> {
let mut stream = UnixStream::connect(self.files.path().join("assistant.sock"))?;
stream.set_read_timeout(Some(Duration::from_secs(5)))?;
stream.set_write_timeout(Some(Duration::from_secs(5)))?;
let token = agent_token_at(pass, store)?;
let agent = if pass == "squatter" {
"agent-squatter"
} else {
"agent-assistant"
};
send(
&mut stream,
&json!({"hello":{"agent":agent,"pass":token,"store":store,"resumable":true,"harness":{"id":"codex","version":"1","protocol":2}}}),
)?;
let mut seat = Seat(BufReader::new(stream));
let welcome = seat.read()?;
anyhow::ensure!(welcome == json!({"welcome":{"person":PERSON}}), "{welcome}");
Ok(seat)
}
}
impl Drop for Fixture {
fn drop(&mut self) {
drop(self.listener.take());
}
}
fn keys() -> Result<lys_pass::KeySet> {
use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
let key = ed25519_dalek::SigningKey::from_bytes(&[7; 32]);
Ok(lys_pass::KeySet::from_json(&json!({"keys":[{"kty":"OKP","crv":"Ed25519","kid":"fixture","x":URL_SAFE_NO_PAD.encode(key.verifying_key().as_bytes())}]}).to_string())?)
}
pub fn agent_token(alias: &str) -> Result<String> {
agent_token_at(alias, "wards")
}
fn agent_token_at(alias: &str, store: &str) -> Result<String> {
use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
use ed25519_dalek::Signer;
let key = ed25519_dalek::SigningKey::from_bytes(&[7; 32]);
let now = crate::serve::pass::now();
let person = if alias == "other-person" {
"person-other"
} else {
PERSON
};
let agent = if alias == "squatter" {
"agent-squatter"
} else if alias == "wrong-agent" {
"agent-other"
} else {
"agent-assistant"
};
let claims = json!({"iss":"https://lys.example","aud":if alias=="wrong-audience" {"liminal"}else{"haematite"},"sub":agent,"iat":now,"exp":now+3600,"holder":{"id":agent,"kind":"agent","responsible":if alias=="unowned-agent" {Value::Null}else{json!(person)}},"rights":[{"resource":{"kind":"haematite.database","id":if alias=="squatter" {"pharmacy"}else{store}},"actions":["agent_permission"],"mode":"by_draft","grant":"grant-fixture"}]});
let bytes = format!(
"{}.{}",
URL_SAFE_NO_PAD.encode(br#"{"alg":"EdDSA","kid":"fixture","typ":"JWT"}"#),
URL_SAFE_NO_PAD.encode(serde_json::to_vec(&claims)?)
);
Ok(format!(
"{bytes}.{}",
URL_SAFE_NO_PAD.encode(key.sign(bytes.as_bytes()).to_bytes())
))
}
pub struct Seat(pub BufReader<UnixStream>);
impl Seat {
pub fn read(&mut self) -> Result<Value> {
let mut line = String::new();
self.0.read_line(&mut line)?;
Ok(serde_json::from_str(&line)?)
}
pub fn send(&mut self, value: &Value) -> Result {
send(self.0.get_mut(), value)
}
pub fn item(&mut self, seq: u64) -> Result {
self.send(&json!({"item":item(seq)}))
}
}
pub fn send(stream: &mut impl Write, value: &Value) -> Result {
writeln!(stream, "{value}")?;
stream.flush()?;
Ok(())
}
pub fn item(seq: u64) -> Value {
json!({"conversation":"conversation","seq":seq,"durable":true,"kind":"text","text":format!("item {seq}")})
}
pub fn written(response: Response) -> Result<String> {
let mut out = Vec::new();
response.write(&mut out)?;
Ok(String::from_utf8(out)?)
}
pub fn check(response: Response, status: u16, code: &str) -> Result {
let text = written(response)?;
anyhow::ensure!(
text.starts_with(&format!("HTTP/1.1 {status} ")) && text.contains(code),
"{text}"
);
Ok(())
}
pub struct Stream {
pub input: BufReader<std::net::TcpStream>,
worker: std::thread::JoinHandle<std::io::Result<()>>,
}
impl Stream {
pub fn open(response: Response) -> Result<Self> {
let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
let client = std::net::TcpStream::connect(listener.local_addr()?)?;
client.set_read_timeout(Some(Duration::from_secs(5)))?;
let (server, _) = listener.accept()?;
let worker = std::thread::spawn(move || response.write_to(&server));
let mut stream = Self {
input: BufReader::new(client),
worker,
};
let mut head = String::new();
loop {
let mut line = String::new();
anyhow::ensure!(
stream.input.read_line(&mut line)? > 0,
"stream ended: {head}"
);
head.push_str(&line);
if line == "\r\n" {
break;
}
}
anyhow::ensure!(head.starts_with("HTTP/1.1 200"), "{head}");
Ok(stream)
}
pub fn event(&mut self) -> Result<String> {
let mut event = String::new();
loop {
let mut line = String::new();
anyhow::ensure!(self.input.read_line(&mut line)? > 0, "stream ended");
event.push_str(&line);
if line == "\n" {
return Ok(event);
}
}
}
pub fn close(self) -> Result {
match self.input.get_ref().shutdown(std::net::Shutdown::Both) {
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotConnected => {}
Err(error) => return Err(error.into()),
}
self.worker
.join()
.map_err(|_| anyhow::anyhow!("stream panic"))??;
Ok(())
}
}
pub fn ack(mut seat: Seat) -> mpsc::Receiver<Value> {
let (send, receive) = mpsc::channel();
std::thread::spawn(move || -> Result {
let op = seat.read()?;
seat.send(&json!({"ack":{"id":op["op"]["id"]}}))?;
seat.item(1)?;
send.send(op)?;
Ok(())
});
receive
}
pub fn record_store(
f: &Fixture,
items: Vec<Value>,
during: impl FnOnce() -> Result + Send + 'static,
) -> Result<std::thread::JoinHandle<Result<Vec<Value>>>> {
let store = records::start(f, records::rows(items), move |_| during())?;
Ok(std::thread::spawn(move || store.finish()))
}