use crate::serve::protocol::{HelloAgreement, PROTOCOL, RECEIPT_CAPABILITY};
use crate::serve::receipt_fragments::{Assembler, VerifiedReceipt, minimum_cap};
use std::io::{self, Read, Write};
use std::num::{NonZeroU32, NonZeroU64};
use std::os::unix::net::UnixStream;
use std::path::{Path, PathBuf};
use std::process::ExitCode;
use std::sync::{Arc, mpsc};
use std::time::Duration;
use serde::{Deserialize, Deserializer, Serialize};
use serde_json::value::RawValue;
use crate::serve::{cli::Config, frame};
pub mod query;
pub mod write;
const USAGE: &str = "\
usage: haematite call (--socket <path> | --config <serve.json>) [--timeout-ms N] <method> [<params> | -]
<params> is one JSON value; - reads it from standard input; absent, the operation has no params.
--timeout-ms bounds connecting, writing and reading; the default is 30000.
Exit 0: answered, the result on standard output. Exit 1: could not ask; nothing was sent.
Exit 2: refused by the service, its code and message on standard error. Exit 3: the outcome
is not known to the caller; the service may have acted.";
const DEFAULT_TIMEOUT_MS: NonZeroU64 = NonZeroU64::MIN.saturating_add(29_999);
const HELLO_BOUND: NonZeroU32 = NonZeroU32::MIN.saturating_add(65_535);
const ID: &str = "haem-call";
pub const MAY_HAVE_APPLIED: &[&str] = &[
"durability_unknown",
"committed_unpublished",
"reference_missing",
"discard_abort_failed",
"catalogue_unavailable",
"storage_unavailable",
"response_limit",
];
pub use crate::serve::protocol::NOT_APPLIED_REFUSALS as KNOWN_REFUSALS;
#[derive(Debug)]
pub enum Ended {
Usage(String),
NotAsked(String),
Unknown(String),
Received {
why: String,
receipt: Arc<VerifiedReceipt>,
},
}
impl Ended {
#[must_use]
pub fn why(&self) -> &str {
match self {
Self::Usage(why)
| Self::NotAsked(why)
| Self::Unknown(why)
| Self::Received { why, .. } => why,
}
}
}
impl std::fmt::Display for Ended {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let kind = match self {
Self::Usage(_) => "usage",
Self::NotAsked(_) => "not asked",
Self::Unknown(_) => "outcome unknown",
Self::Received { .. } => "complete receipt received",
};
write!(formatter, "{kind}: {}", self.why())
}
}
impl std::error::Error for Ended {}
#[derive(Debug)]
pub enum Reply {
Answered(Box<RawValue>),
Refused(Refusal),
}
#[derive(Debug, Deserialize)]
pub struct Refusal {
pub code: String,
pub message: String,
#[serde(default)]
pub path: Option<String>,
#[serde(skip)]
pub verified: Option<Arc<VerifiedReceipt>>,
}
pub struct Client {
stream: UnixStream,
hello: Box<RawValue>,
max_frame: NonZeroU32,
max_page: NonZeroU32,
}
#[derive(Deserialize)]
struct Limits {
protocol: String,
capabilities: Vec<String>,
max_frame: NonZeroU32,
max_page: NonZeroU32,
}
impl Client {
pub fn connect(socket: &Path, timeout: Duration) -> Result<Self, Ended> {
let mut stream = connect(socket, timeout)?;
let unset =
|error: io::Error| Ended::NotAsked(format!("the timeout could not be set: {error}"));
stream.set_read_timeout(Some(timeout)).map_err(unset)?;
stream.set_write_timeout(Some(timeout)).map_err(unset)?;
let agreement = serde_json::value::to_raw_value(&HelloAgreement::client())
.map_err(|error| Ended::NotAsked(format!("receipt_hello_encode_failed: {error}")))?;
let hello = match exchange(&mut stream, HELLO_BOUND, "hello", Some(&agreement)) {
Ok(Reply::Answered(hello)) => hello,
Ok(Reply::Refused(refusal)) => {
return Err(Ended::NotAsked(format!(
"the service refused hello: {}: {}",
refusal.code, refusal.message
)));
}
Err(Ended::Received { why, receipt }) => {
return Err(Ended::Received {
why: format!("hello failed: {why}"),
receipt,
});
}
Err(Ended::Unknown(why) | Ended::NotAsked(why) | Ended::Usage(why)) => {
return Err(Ended::NotAsked(format!("hello failed: {why}")));
}
};
let limits: Limits = serde_json::from_str(hello.get()).map_err(|error| {
Ended::NotAsked(format!(
"the service's hello did not give its limits: {error}"
))
})?;
if limits.protocol != PROTOCOL {
return Err(Ended::NotAsked(format!(
"receipt_protocol_mismatch: expected {PROTOCOL}, received {}",
limits.protocol
)));
}
if !limits
.capabilities
.iter()
.any(|capability| capability == RECEIPT_CAPABILITY)
{
return Err(Ended::NotAsked(format!(
"typed_receipt_capability_missing: required {RECEIPT_CAPABILITY}"
)));
}
let minimum = minimum_cap().map_err(|error| Ended::NotAsked(error.to_string()))?;
if u64::from(limits.max_frame.get())
< u64::try_from(minimum).map_err(|error| Ended::NotAsked(error.to_string()))?
{
return Err(Ended::NotAsked(format!(
"receipt_frame_too_small: offered {}, minimum {minimum}",
limits.max_frame
)));
}
Ok(Self {
stream,
hello,
max_frame: limits.max_frame,
max_page: limits.max_page,
})
}
#[must_use]
pub fn hello(&self) -> &RawValue {
&self.hello
}
#[must_use]
pub const fn max_frame(&self) -> NonZeroU32 {
self.max_frame
}
#[must_use]
pub const fn max_page(&self) -> NonZeroU32 {
self.max_page
}
pub fn call(&mut self, method: &str, params: Option<&RawValue>) -> Result<Reply, Ended> {
let reply = exchange(&mut self.stream, self.max_frame, method, params)?;
let Reply::Refused(refusal) = reply else {
return Ok(reply);
};
if method != "mutation.apply" || KNOWN_REFUSALS.contains(&refusal.code.as_str()) {
return Ok(Reply::Refused(refusal));
}
let why = if MAY_HAVE_APPLIED.contains(&refusal.code.as_str()) {
"the change may have been applied"
} else {
"this client does not know that code, so whether the change was applied is not known"
};
let why = format!(
"the service answered {}: {}; {why}",
refusal.code, refusal.message
);
match refusal.verified {
Some(receipt) => Err(Ended::Received { why, receipt }),
None => Err(Ended::Unknown(why)),
}
}
pub fn next_reply(&mut self) -> Result<Reply, Ended> {
receive(&mut self.stream, self.max_frame)
}
pub(crate) fn follow_socket(&self) -> io::Result<UnixStream> {
self.stream.try_clone()
}
pub fn read_within(&self, timeout: Option<Duration>) -> Result<(), Ended> {
self.stream
.set_read_timeout(timeout)
.map_err(|error| Ended::Unknown(format!("the read bound could not be set: {error}")))
}
}
#[must_use]
pub fn run(args: impl Iterator<Item = String>) -> ExitCode {
finish(call(args))
}
fn finish(outcome: Result<Reply, Ended>) -> ExitCode {
match outcome {
Ok(Reply::Answered(result)) => {
let mut output = io::stdout().lock();
match writeln!(output, "{}", result.get()).and_then(|()| output.flush()) {
Ok(()) => ExitCode::SUCCESS,
Err(error) => {
eprintln!(
"haematite: answer received, but the result could not be written to stdout: {error}"
);
ExitCode::from(3)
}
}
}
Ok(Reply::Refused(refusal)) => {
if let Err(error) = print_refusal(&refusal) {
eprintln!("haematite: receipt_output_failed: {error}");
}
ExitCode::from(2)
}
Err(ended) => {
if let Err(error) = print_ended(&ended) {
eprintln!("haematite: receipt_output_failed: {error}");
}
ExitCode::from(exit_code(&ended))
}
}
}
#[must_use]
pub const fn exit_code(ended: &Ended) -> u8 {
match ended {
Ended::Usage(_) | Ended::NotAsked(_) => 1,
Ended::Unknown(_) | Ended::Received { .. } => 3,
}
}
#[must_use]
pub fn refused(refusal: &Refusal) -> String {
let at = refusal
.path
.as_deref()
.map_or_else(String::new, |path| format!(" at {path}"));
let (code, message) = (&refusal.code, &refusal.message);
format!("haematite: refused {code}: {message}{at}")
}
#[must_use]
pub fn words(ended: &Ended) -> String {
match ended {
Ended::Usage(why) => format!("haematite: {why}"),
Ended::NotAsked(why) => format!("haematite: could not ask: {why}"),
Ended::Unknown(why) => format!("haematite: the outcome is not known: {why}"),
Ended::Received { why, .. } => format!("haematite: complete receipt received: {why}"),
}
}
pub(super) fn print_refusal(refusal: &Refusal) -> io::Result<()> {
let mut output = io::stderr().lock();
writeln!(output, "{}", refused(refusal))?;
if let Some(receipt) = &refusal.verified {
output.write_all(receipt.bytes())?;
output.write_all(b"\n")?;
}
output.flush()
}
pub(super) fn print_ended(ended: &Ended) -> io::Result<()> {
let mut output = io::stderr().lock();
writeln!(output, "{}", words(ended))?;
if let Ended::Received { receipt, .. } = ended {
output.write_all(receipt.bytes())?;
output.write_all(b"\n")?;
}
output.flush()
}
enum Target {
Socket(PathBuf),
Config(PathBuf),
}
#[derive(Default)]
struct Common {
target: Option<Target>,
timeout: Option<NonZeroU64>,
}
impl Common {
fn take(&mut self, arg: &str, args: &mut impl Iterator<Item = String>) -> Result<bool, String> {
match arg {
"--socket" | "--config" if self.target.is_none() => {
let path = PathBuf::from(args.next().ok_or_else(|| format!("{arg} needs a path"))?);
self.target = Some(if arg == "--socket" {
Target::Socket(path)
} else {
Target::Config(path)
});
Ok(true)
}
"--socket" | "--config" => Err("give exactly one of --socket and --config".to_owned()),
"--timeout-ms" if self.timeout.is_none() => {
let text = args
.next()
.ok_or("--timeout-ms needs a number of milliseconds")?;
let millis = text
.parse::<NonZeroU64>()
.map_err(|error| format!("--timeout-ms {text:?}: {error}"))?;
self.timeout = Some(millis);
Ok(true)
}
_ => Ok(false),
}
}
fn finish(self) -> Result<(Target, Duration), String> {
let target = self
.target
.ok_or("give exactly one of --socket and --config")?;
let millis = self.timeout.unwrap_or(DEFAULT_TIMEOUT_MS);
Ok((target, Duration::from_millis(millis.get())))
}
}
struct Call {
target: Target,
timeout: Duration,
method: String,
params: Option<Text>,
}
enum Text {
Given(String),
Stdin,
}
impl Text {
fn of(text: String) -> Self {
if text == "-" {
Self::Stdin
} else {
Self::Given(text)
}
}
}
fn call(args: impl Iterator<Item = String>) -> Result<Reply, Ended> {
let asked = parse(args).map_err(|why| Ended::Usage(format!("{why}\n{USAGE}")))?;
let params = asked
.params
.map(|text| one_value(text, "params"))
.transpose()?;
let socket = socket_of(&asked.target)?;
let mut client = Client::connect(&socket, asked.timeout)?;
if asked.method == "hello" && params.is_none() {
return Ok(Reply::Answered(client.hello().to_owned()));
}
client.call(&asked.method, params.as_deref())
}
fn parse(mut args: impl Iterator<Item = String>) -> Result<Call, String> {
let mut common = Common::default();
let mut rest = Vec::new();
while let Some(arg) = args.next() {
if common.take(&arg, &mut args)? {
continue;
}
if arg.starts_with("--") {
return Err(format!("unknown or repeated flag {arg:?}"));
}
rest.push(arg);
}
let (target, timeout) = common.finish()?;
let mut rest = rest.into_iter();
let method = rest.next().ok_or("name the method to call")?;
let params = rest.next().map(Text::of);
if let Some(extra) = rest.next() {
return Err(format!("unexpected argument {extra:?}"));
}
Ok(Call {
target,
timeout,
method,
params,
})
}
fn one_value(given: Text, what: &str) -> Result<Box<RawValue>, Ended> {
let text = match given {
Text::Given(text) => text,
Text::Stdin => {
let mut text = String::new();
io::stdin().read_to_string(&mut text).map_err(|error| {
Ended::NotAsked(format!(
"the {what} could not be read from standard input: {error}"
))
})?;
text
}
};
serde_json::from_str(&text)
.map_err(|error| Ended::NotAsked(format!("the {what} is not one JSON value: {error}")))
}
fn socket_of(target: &Target) -> Result<PathBuf, Ended> {
match target {
Target::Socket(path) => Ok(path.clone()),
Target::Config(path) => {
let named = path.display();
let bytes = std::fs::read(path).map_err(|error| {
Ended::NotAsked(format!(
"the service config {named} could not be read: {error}"
))
})?;
let config: Config = serde_json::from_slice(&bytes)
.map_err(|error| Ended::NotAsked(format!("the service config {named}: {error}")))?;
config
.endpoint(false)
.map(|endpoint| endpoint.socket)
.map_err(|error| Ended::NotAsked(format!("the service config {named}: {error}")))
}
}
}
fn connect(socket: &Path, timeout: Duration) -> Result<UnixStream, Ended> {
let (sender, receiver) = mpsc::channel();
let path = socket.to_path_buf();
std::thread::Builder::new()
.name("haem-call-connect".to_owned())
.spawn(move || {
let _unwaited = sender.send(UnixStream::connect(path));
})
.map_err(|error| Ended::NotAsked(format!("the connect could not start: {error}")))?;
match receiver.recv_timeout(timeout) {
Ok(Ok(stream)) => Ok(stream),
Ok(Err(error)) => Err(Ended::NotAsked(format!(
"could not connect to {}: {error}",
socket.display()
))),
Err(_) => Err(Ended::NotAsked(format!(
"could not connect to {} within {} ms",
socket.display(),
timeout.as_millis()
))),
}
}
#[derive(Serialize)]
struct Outgoing<'a> {
id: &'a str,
operation: Operation<'a>,
}
#[derive(Serialize)]
struct Operation<'a> {
method: &'a str,
#[serde(skip_serializing_if = "Option::is_none")]
params: Option<&'a RawValue>,
}
#[derive(Deserialize)]
struct Incoming {
id: String,
#[serde(default, deserialize_with = "present")]
result: Option<Box<RawValue>>,
#[serde(default)]
error: Option<Refusal>,
}
fn present<'de, D: Deserializer<'de>>(input: D) -> Result<Option<Box<RawValue>>, D::Error> {
Box::<RawValue>::deserialize(input).map(Some)
}
#[derive(Deserialize)]
struct Probe {
#[serde(default)]
receipt: Option<Box<RawValue>>,
}
fn exchange(
stream: &mut UnixStream,
limit: NonZeroU32,
method: &str,
params: Option<&RawValue>,
) -> Result<Reply, Ended> {
let operation = Operation { method, params };
let request = serde_json::to_vec(&Outgoing { id: ID, operation })
.map_err(|error| Ended::NotAsked(format!("the request could not be written: {error}")))?;
if !u32::try_from(request.len()).is_ok_and(|length| length <= limit.get()) {
return Err(Ended::NotAsked(format!(
"request_too_large: the request is {} bytes and the service reads at most {limit}",
request.len()
)));
}
frame::write(stream, &request, limit).map_err(|error| lost("sending the request", &error))?;
receive(stream, limit)
}
fn receive(stream: &mut UnixStream, limit: NonZeroU32) -> Result<Reply, Ended> {
let bytes = match frame::read(stream, limit) {
Ok(Some(bytes)) => bytes,
Ok(None) => {
return Err(Ended::Unknown(
"the service closed the connection before answering".to_owned(),
));
}
Err(error) => return Err(lost("waiting for the answer", &error)),
};
let probe: Probe = serde_json::from_slice(&bytes)
.map_err(|error| Ended::Unknown(format!("receipt_frame_decode_failed: {error}")))?;
if probe.receipt.is_none() {
return reply(&bytes);
}
let mut assembler =
Assembler::new(ID, limit.get()).map_err(|error| Ended::Unknown(error.to_string()))?;
let mut body = bytes;
loop {
if let Some(complete) = assembler
.accept(&body)
.map_err(|error| Ended::Unknown(error.to_string()))?
{
let complete = Arc::new(complete);
let refusal = complete.refusal().ok_or_else(|| Ended::Received {
why: "receipt_refusal_missing".to_owned(),
receipt: Arc::clone(&complete),
})?;
return Ok(Reply::Refused(Refusal {
code: refusal.code.clone(),
message: refusal.message.clone(),
path: refusal.path.clone(),
verified: Some(complete),
}));
}
body = match frame::read(stream, limit) {
Ok(Some(body)) => body,
Ok(None) => return Err(Ended::Unknown("receipt_connection_closed".to_owned())),
Err(error) => {
return Err(Ended::Unknown(format!(
"receipt_connection_failed: {error}"
)));
}
};
}
}
fn lost(doing: &str, error: &io::Error) -> Ended {
match error.kind() {
io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut => {
Ended::Unknown(format!("timed out {doing}: {error}"))
}
_ => Ended::Unknown(format!("the connection failed {doing}: {error}")),
}
}
fn reply(bytes: &[u8]) -> Result<Reply, Ended> {
let unreadable = |why: String| Ended::Unknown(format!("the answer could not be read: {why}"));
let incoming: Incoming =
serde_json::from_slice(bytes).map_err(|error| unreadable(error.to_string()))?;
if incoming.id != ID {
return Err(unreadable(format!(
"it answers {:?}, not {ID:?}",
incoming.id
)));
}
match (incoming.result, incoming.error) {
(Some(result), None) => Ok(Reply::Answered(result)),
(None, Some(refusal)) => Ok(Reply::Refused(refusal)),
_ => Err(unreadable(
"it carries neither or both of result and error".to_owned(),
)),
}
}
#[cfg(test)]
mod nonzero_tests {
use super::{Common, DEFAULT_TIMEOUT_MS, NonZeroU64, PathBuf, Target};
#[test]
fn nonzero_timeout_readback() -> Result<(), Box<dyn std::error::Error>> {
let cases = [
(None, DEFAULT_TIMEOUT_MS.get()),
(Some(1), 1),
(Some(37), 37),
(Some(u64::MAX), u64::MAX),
];
for (given, expected) in cases {
let timeout = given
.map(|value| NonZeroU64::new(value).ok_or("zero timeout"))
.transpose()?;
let common = Common {
target: Some(Target::Socket(PathBuf::from("unused.sock"))),
timeout: std::hint::black_box(timeout),
};
let (_, actual) = common.finish()?;
assert_eq!(actual.as_millis(), u128::from(expected));
}
Ok(())
}
}
#[cfg(test)]
mod json_text_refusal_tests {
use super::{KNOWN_REFUSALS, Refusal, Reply, finish, refused};
use std::process::ExitCode;
#[test]
fn json_text_invalid_is_a_named_client_refusal() {
let refusal = Refusal {
code: "json_text_invalid".to_owned(),
message: "value is not JSON".to_owned(),
path: Some("/payload".to_owned()),
verified: None,
};
assert!(KNOWN_REFUSALS.contains(&refusal.code.as_str()));
assert_eq!(
refused(&refusal),
"haematite: refused json_text_invalid: value is not JSON at /payload"
);
assert_eq!(finish(Ok(Reply::Refused(refusal))), ExitCode::from(2));
}
}
#[cfg(test)]
mod access_refusal_tests {
use super::KNOWN_REFUSALS;
#[test]
fn every_access_refusal_is_known_not_to_have_applied() {
for code in [
"route_unclassified",
"permission_refused",
"held_for_approval",
"pass_missing",
"lys_could_not_be_asked",
"signature_refused",
"pass_expired",
] {
assert!(KNOWN_REFUSALS.contains(&code), "{code} is not known");
}
}
}