1use crate::serve::protocol::{HelloAgreement, PROTOCOL, RECEIPT_CAPABILITY};
12use crate::serve::receipt_fragments::{Assembler, VerifiedReceipt, minimum_cap};
13use std::io::{self, Read, Write};
14use std::num::{NonZeroU32, NonZeroU64};
15use std::os::unix::net::UnixStream;
16use std::path::{Path, PathBuf};
17use std::process::ExitCode;
18use std::sync::{Arc, mpsc};
19use std::time::Duration;
20
21use serde::{Deserialize, Deserializer, Serialize};
22use serde_json::value::RawValue;
23
24use crate::serve::{cli::Config, frame};
25
26pub mod query;
27pub mod write;
28
29const USAGE: &str = "\
30usage: haematite call (--socket <path> | --config <serve.json>) [--timeout-ms N] <method> [<params> | -]
31<params> is one JSON value; - reads it from standard input; absent, the operation has no params.
32--timeout-ms bounds connecting, writing and reading; the default is 30000.
33Exit 0: answered, the result on standard output. Exit 1: could not ask; nothing was sent.
34Exit 2: refused by the service, its code and message on standard error. Exit 3: the outcome
35is not known to the caller; the service may have acted.";
36
37const DEFAULT_TIMEOUT_MS: NonZeroU64 = NonZeroU64::MIN.saturating_add(29_999);
38const HELLO_BOUND: NonZeroU32 = NonZeroU32::MIN.saturating_add(65_535);
40const ID: &str = "haem-call";
41
42pub const MAY_HAVE_APPLIED: &[&str] = &[
47 "durability_unknown",
51 "committed_unpublished",
54 "reference_missing",
58 "discard_abort_failed",
61 "catalogue_unavailable",
65 "storage_unavailable",
68 "response_limit",
71];
72
73pub use crate::serve::protocol::NOT_APPLIED_REFUSALS as KNOWN_REFUSALS;
75
76#[derive(Debug)]
78pub enum Ended {
79 Usage(String),
81 NotAsked(String),
83 Unknown(String),
86 Received {
91 why: String,
92 receipt: Arc<VerifiedReceipt>,
93 },
94}
95
96impl Ended {
97 #[must_use]
99 pub fn why(&self) -> &str {
100 match self {
101 Self::Usage(why)
102 | Self::NotAsked(why)
103 | Self::Unknown(why)
104 | Self::Received { why, .. } => why,
105 }
106 }
107}
108
109impl std::fmt::Display for Ended {
110 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
111 let kind = match self {
112 Self::Usage(_) => "usage",
113 Self::NotAsked(_) => "not asked",
114 Self::Unknown(_) => "outcome unknown",
115 Self::Received { .. } => "complete receipt received",
116 };
117 write!(formatter, "{kind}: {}", self.why())
118 }
119}
120
121impl std::error::Error for Ended {}
122
123#[derive(Debug)]
125pub enum Reply {
126 Answered(Box<RawValue>),
128 Refused(Refusal),
129}
130
131#[derive(Debug, Deserialize)]
133pub struct Refusal {
134 pub code: String,
135 pub message: String,
136 #[serde(default)]
137 pub path: Option<String>,
138 #[serde(skip)]
139 pub verified: Option<Arc<VerifiedReceipt>>,
140}
141
142pub struct Client {
144 stream: UnixStream,
145 hello: Box<RawValue>,
146 max_frame: NonZeroU32,
147 max_page: NonZeroU32,
148}
149
150#[derive(Deserialize)]
151struct Limits {
152 protocol: String,
153 capabilities: Vec<String>,
154 max_frame: NonZeroU32,
155 max_page: NonZeroU32,
156}
157
158impl Client {
159 pub fn connect(socket: &Path, timeout: Duration) -> Result<Self, Ended> {
166 let mut stream = connect(socket, timeout)?;
167 let unset =
168 |error: io::Error| Ended::NotAsked(format!("the timeout could not be set: {error}"));
169 stream.set_read_timeout(Some(timeout)).map_err(unset)?;
170 stream.set_write_timeout(Some(timeout)).map_err(unset)?;
171 let agreement = serde_json::value::to_raw_value(&HelloAgreement::client())
172 .map_err(|error| Ended::NotAsked(format!("receipt_hello_encode_failed: {error}")))?;
173 let hello = match exchange(&mut stream, HELLO_BOUND, "hello", Some(&agreement)) {
174 Ok(Reply::Answered(hello)) => hello,
175 Ok(Reply::Refused(refusal)) => {
176 return Err(Ended::NotAsked(format!(
177 "the service refused hello: {}: {}",
178 refusal.code, refusal.message
179 )));
180 }
181 Err(Ended::Received { why, receipt }) => {
182 return Err(Ended::Received {
183 why: format!("hello failed: {why}"),
184 receipt,
185 });
186 }
187 Err(Ended::Unknown(why) | Ended::NotAsked(why) | Ended::Usage(why)) => {
188 return Err(Ended::NotAsked(format!("hello failed: {why}")));
189 }
190 };
191 let limits: Limits = serde_json::from_str(hello.get()).map_err(|error| {
192 Ended::NotAsked(format!(
193 "the service's hello did not give its limits: {error}"
194 ))
195 })?;
196 if limits.protocol != PROTOCOL {
197 return Err(Ended::NotAsked(format!(
198 "receipt_protocol_mismatch: expected {PROTOCOL}, received {}",
199 limits.protocol
200 )));
201 }
202 if !limits
203 .capabilities
204 .iter()
205 .any(|capability| capability == RECEIPT_CAPABILITY)
206 {
207 return Err(Ended::NotAsked(format!(
208 "typed_receipt_capability_missing: required {RECEIPT_CAPABILITY}"
209 )));
210 }
211 let minimum = minimum_cap().map_err(|error| Ended::NotAsked(error.to_string()))?;
212 if u64::from(limits.max_frame.get())
213 < u64::try_from(minimum).map_err(|error| Ended::NotAsked(error.to_string()))?
214 {
215 return Err(Ended::NotAsked(format!(
216 "receipt_frame_too_small: offered {}, minimum {minimum}",
217 limits.max_frame
218 )));
219 }
220 Ok(Self {
221 stream,
222 hello,
223 max_frame: limits.max_frame,
224 max_page: limits.max_page,
225 })
226 }
227
228 #[must_use]
230 pub fn hello(&self) -> &RawValue {
231 &self.hello
232 }
233
234 #[must_use]
236 pub const fn max_frame(&self) -> NonZeroU32 {
237 self.max_frame
238 }
239
240 #[must_use]
242 pub const fn max_page(&self) -> NonZeroU32 {
243 self.max_page
244 }
245
246 pub fn call(&mut self, method: &str, params: Option<&RawValue>) -> Result<Reply, Ended> {
257 let reply = exchange(&mut self.stream, self.max_frame, method, params)?;
258 let Reply::Refused(refusal) = reply else {
259 return Ok(reply);
260 };
261 if method != "mutation.apply" || KNOWN_REFUSALS.contains(&refusal.code.as_str()) {
263 return Ok(Reply::Refused(refusal));
264 }
265 let why = if MAY_HAVE_APPLIED.contains(&refusal.code.as_str()) {
266 "the change may have been applied"
267 } else {
268 "this client does not know that code, so whether the change was applied is not known"
269 };
270 let why = format!(
271 "the service answered {}: {}; {why}",
272 refusal.code, refusal.message
273 );
274 match refusal.verified {
275 Some(receipt) => Err(Ended::Received { why, receipt }),
276 None => Err(Ended::Unknown(why)),
277 }
278 }
279
280 pub fn next_reply(&mut self) -> Result<Reply, Ended> {
289 receive(&mut self.stream, self.max_frame)
290 }
291
292 pub(crate) fn follow_socket(&self) -> io::Result<UnixStream> {
294 self.stream.try_clone()
295 }
296
297 pub fn read_within(&self, timeout: Option<Duration>) -> Result<(), Ended> {
303 self.stream
304 .set_read_timeout(timeout)
305 .map_err(|error| Ended::Unknown(format!("the read bound could not be set: {error}")))
306 }
307}
308
309#[must_use]
311pub fn run(args: impl Iterator<Item = String>) -> ExitCode {
312 finish(call(args))
313}
314
315fn finish(outcome: Result<Reply, Ended>) -> ExitCode {
318 match outcome {
319 Ok(Reply::Answered(result)) => {
320 let mut output = io::stdout().lock();
321 match writeln!(output, "{}", result.get()).and_then(|()| output.flush()) {
322 Ok(()) => ExitCode::SUCCESS,
323 Err(error) => {
324 eprintln!(
326 "haematite: answer received, but the result could not be written to stdout: {error}"
327 );
328 ExitCode::from(3)
329 }
330 }
331 }
332 Ok(Reply::Refused(refusal)) => {
333 if let Err(error) = print_refusal(&refusal) {
334 eprintln!("haematite: receipt_output_failed: {error}");
335 }
336 ExitCode::from(2)
337 }
338 Err(ended) => {
339 if let Err(error) = print_ended(&ended) {
340 eprintln!("haematite: receipt_output_failed: {error}");
341 }
342 ExitCode::from(exit_code(&ended))
343 }
344 }
345}
346
347#[must_use]
349pub const fn exit_code(ended: &Ended) -> u8 {
350 match ended {
351 Ended::Usage(_) | Ended::NotAsked(_) => 1,
352 Ended::Unknown(_) | Ended::Received { .. } => 3,
353 }
354}
355
356#[must_use]
358pub fn refused(refusal: &Refusal) -> String {
359 let at = refusal
360 .path
361 .as_deref()
362 .map_or_else(String::new, |path| format!(" at {path}"));
363 let (code, message) = (&refusal.code, &refusal.message);
364 format!("haematite: refused {code}: {message}{at}")
365}
366
367#[must_use]
369pub fn words(ended: &Ended) -> String {
370 match ended {
371 Ended::Usage(why) => format!("haematite: {why}"),
372 Ended::NotAsked(why) => format!("haematite: could not ask: {why}"),
373 Ended::Unknown(why) => format!("haematite: the outcome is not known: {why}"),
374 Ended::Received { why, .. } => format!("haematite: complete receipt received: {why}"),
375 }
376}
377
378pub(super) fn print_refusal(refusal: &Refusal) -> io::Result<()> {
379 let mut output = io::stderr().lock();
380 writeln!(output, "{}", refused(refusal))?;
381 if let Some(receipt) = &refusal.verified {
382 output.write_all(receipt.bytes())?;
383 output.write_all(b"\n")?;
384 }
385 output.flush()
386}
387pub(super) fn print_ended(ended: &Ended) -> io::Result<()> {
388 let mut output = io::stderr().lock();
389 writeln!(output, "{}", words(ended))?;
390 if let Ended::Received { receipt, .. } = ended {
391 output.write_all(receipt.bytes())?;
392 output.write_all(b"\n")?;
393 }
394 output.flush()
395}
396
397enum Target {
398 Socket(PathBuf),
399 Config(PathBuf),
400}
401
402#[derive(Default)]
405struct Common {
406 target: Option<Target>,
407 timeout: Option<NonZeroU64>,
408}
409
410impl Common {
411 fn take(&mut self, arg: &str, args: &mut impl Iterator<Item = String>) -> Result<bool, String> {
413 match arg {
414 "--socket" | "--config" if self.target.is_none() => {
415 let path = PathBuf::from(args.next().ok_or_else(|| format!("{arg} needs a path"))?);
416 self.target = Some(if arg == "--socket" {
417 Target::Socket(path)
418 } else {
419 Target::Config(path)
420 });
421 Ok(true)
422 }
423 "--socket" | "--config" => Err("give exactly one of --socket and --config".to_owned()),
424 "--timeout-ms" if self.timeout.is_none() => {
425 let text = args
426 .next()
427 .ok_or("--timeout-ms needs a number of milliseconds")?;
428 let millis = text
429 .parse::<NonZeroU64>()
430 .map_err(|error| format!("--timeout-ms {text:?}: {error}"))?;
431 self.timeout = Some(millis);
432 Ok(true)
433 }
434 _ => Ok(false),
435 }
436 }
437
438 fn finish(self) -> Result<(Target, Duration), String> {
440 let target = self
441 .target
442 .ok_or("give exactly one of --socket and --config")?;
443 let millis = self.timeout.unwrap_or(DEFAULT_TIMEOUT_MS);
444 Ok((target, Duration::from_millis(millis.get())))
445 }
446}
447
448struct Call {
449 target: Target,
450 timeout: Duration,
451 method: String,
452 params: Option<Text>,
454}
455
456enum Text {
458 Given(String),
459 Stdin,
460}
461
462impl Text {
463 fn of(text: String) -> Self {
464 if text == "-" {
465 Self::Stdin
466 } else {
467 Self::Given(text)
468 }
469 }
470}
471
472fn call(args: impl Iterator<Item = String>) -> Result<Reply, Ended> {
473 let asked = parse(args).map_err(|why| Ended::Usage(format!("{why}\n{USAGE}")))?;
474 let params = asked
475 .params
476 .map(|text| one_value(text, "params"))
477 .transpose()?;
478 let socket = socket_of(&asked.target)?;
479 let mut client = Client::connect(&socket, asked.timeout)?;
480 if asked.method == "hello" && params.is_none() {
481 return Ok(Reply::Answered(client.hello().to_owned()));
482 }
483 client.call(&asked.method, params.as_deref())
484}
485
486fn parse(mut args: impl Iterator<Item = String>) -> Result<Call, String> {
487 let mut common = Common::default();
488 let mut rest = Vec::new();
489 while let Some(arg) = args.next() {
490 if common.take(&arg, &mut args)? {
491 continue;
492 }
493 if arg.starts_with("--") {
494 return Err(format!("unknown or repeated flag {arg:?}"));
495 }
496 rest.push(arg);
497 }
498 let (target, timeout) = common.finish()?;
499 let mut rest = rest.into_iter();
500 let method = rest.next().ok_or("name the method to call")?;
501 let params = rest.next().map(Text::of);
502 if let Some(extra) = rest.next() {
503 return Err(format!("unexpected argument {extra:?}"));
504 }
505 Ok(Call {
506 target,
507 timeout,
508 method,
509 params,
510 })
511}
512
513fn one_value(given: Text, what: &str) -> Result<Box<RawValue>, Ended> {
515 let text = match given {
516 Text::Given(text) => text,
517 Text::Stdin => {
518 let mut text = String::new();
519 io::stdin().read_to_string(&mut text).map_err(|error| {
520 Ended::NotAsked(format!(
521 "the {what} could not be read from standard input: {error}"
522 ))
523 })?;
524 text
525 }
526 };
527 serde_json::from_str(&text)
528 .map_err(|error| Ended::NotAsked(format!("the {what} is not one JSON value: {error}")))
529}
530
531fn socket_of(target: &Target) -> Result<PathBuf, Ended> {
534 match target {
535 Target::Socket(path) => Ok(path.clone()),
536 Target::Config(path) => {
537 let named = path.display();
538 let bytes = std::fs::read(path).map_err(|error| {
539 Ended::NotAsked(format!(
540 "the service config {named} could not be read: {error}"
541 ))
542 })?;
543 let config: Config = serde_json::from_slice(&bytes)
544 .map_err(|error| Ended::NotAsked(format!("the service config {named}: {error}")))?;
545 config
546 .endpoint(false)
547 .map(|endpoint| endpoint.socket)
548 .map_err(|error| Ended::NotAsked(format!("the service config {named}: {error}")))
549 }
550 }
551}
552
553fn connect(socket: &Path, timeout: Duration) -> Result<UnixStream, Ended> {
556 let (sender, receiver) = mpsc::channel();
557 let path = socket.to_path_buf();
558 std::thread::Builder::new()
559 .name("haem-call-connect".to_owned())
560 .spawn(move || {
561 let _unwaited = sender.send(UnixStream::connect(path));
563 })
564 .map_err(|error| Ended::NotAsked(format!("the connect could not start: {error}")))?;
565 match receiver.recv_timeout(timeout) {
566 Ok(Ok(stream)) => Ok(stream),
567 Ok(Err(error)) => Err(Ended::NotAsked(format!(
568 "could not connect to {}: {error}",
569 socket.display()
570 ))),
571 Err(_) => Err(Ended::NotAsked(format!(
572 "could not connect to {} within {} ms",
573 socket.display(),
574 timeout.as_millis()
575 ))),
576 }
577}
578
579#[derive(Serialize)]
580struct Outgoing<'a> {
581 id: &'a str,
582 operation: Operation<'a>,
583}
584
585#[derive(Serialize)]
586struct Operation<'a> {
587 method: &'a str,
588 #[serde(skip_serializing_if = "Option::is_none")]
589 params: Option<&'a RawValue>,
590}
591
592#[derive(Deserialize)]
593struct Incoming {
594 id: String,
595 #[serde(default, deserialize_with = "present")]
597 result: Option<Box<RawValue>>,
598 #[serde(default)]
599 error: Option<Refusal>,
600}
601
602fn present<'de, D: Deserializer<'de>>(input: D) -> Result<Option<Box<RawValue>>, D::Error> {
603 Box::<RawValue>::deserialize(input).map(Some)
604}
605
606#[derive(Deserialize)]
608struct Probe {
609 #[serde(default)]
610 receipt: Option<Box<RawValue>>,
611}
612
613fn exchange(
614 stream: &mut UnixStream,
615 limit: NonZeroU32,
616 method: &str,
617 params: Option<&RawValue>,
618) -> Result<Reply, Ended> {
619 let operation = Operation { method, params };
620 let request = serde_json::to_vec(&Outgoing { id: ID, operation })
621 .map_err(|error| Ended::NotAsked(format!("the request could not be written: {error}")))?;
622 if !u32::try_from(request.len()).is_ok_and(|length| length <= limit.get()) {
623 return Err(Ended::NotAsked(format!(
624 "request_too_large: the request is {} bytes and the service reads at most {limit}",
625 request.len()
626 )));
627 }
628 frame::write(stream, &request, limit).map_err(|error| lost("sending the request", &error))?;
629 receive(stream, limit)
630}
631
632fn receive(stream: &mut UnixStream, limit: NonZeroU32) -> Result<Reply, Ended> {
634 let bytes = match frame::read(stream, limit) {
635 Ok(Some(bytes)) => bytes,
636 Ok(None) => {
637 return Err(Ended::Unknown(
638 "the service closed the connection before answering".to_owned(),
639 ));
640 }
641 Err(error) => return Err(lost("waiting for the answer", &error)),
642 };
643 let probe: Probe = serde_json::from_slice(&bytes)
644 .map_err(|error| Ended::Unknown(format!("receipt_frame_decode_failed: {error}")))?;
645 if probe.receipt.is_none() {
646 return reply(&bytes);
647 }
648 let mut assembler =
649 Assembler::new(ID, limit.get()).map_err(|error| Ended::Unknown(error.to_string()))?;
650 let mut body = bytes;
651 loop {
652 if let Some(complete) = assembler
653 .accept(&body)
654 .map_err(|error| Ended::Unknown(error.to_string()))?
655 {
656 let complete = Arc::new(complete);
657 let refusal = complete.refusal().ok_or_else(|| Ended::Received {
658 why: "receipt_refusal_missing".to_owned(),
659 receipt: Arc::clone(&complete),
660 })?;
661 return Ok(Reply::Refused(Refusal {
662 code: refusal.code.clone(),
663 message: refusal.message.clone(),
664 path: refusal.path.clone(),
665 verified: Some(complete),
666 }));
667 }
668 body = match frame::read(stream, limit) {
669 Ok(Some(body)) => body,
670 Ok(None) => return Err(Ended::Unknown("receipt_connection_closed".to_owned())),
671 Err(error) => {
672 return Err(Ended::Unknown(format!(
673 "receipt_connection_failed: {error}"
674 )));
675 }
676 };
677 }
678}
679
680fn lost(doing: &str, error: &io::Error) -> Ended {
681 match error.kind() {
682 io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut => {
683 Ended::Unknown(format!("timed out {doing}: {error}"))
684 }
685 _ => Ended::Unknown(format!("the connection failed {doing}: {error}")),
686 }
687}
688
689fn reply(bytes: &[u8]) -> Result<Reply, Ended> {
690 let unreadable = |why: String| Ended::Unknown(format!("the answer could not be read: {why}"));
691 let incoming: Incoming =
692 serde_json::from_slice(bytes).map_err(|error| unreadable(error.to_string()))?;
693 if incoming.id != ID {
694 return Err(unreadable(format!(
695 "it answers {:?}, not {ID:?}",
696 incoming.id
697 )));
698 }
699 match (incoming.result, incoming.error) {
700 (Some(result), None) => Ok(Reply::Answered(result)),
701 (None, Some(refusal)) => Ok(Reply::Refused(refusal)),
702 _ => Err(unreadable(
703 "it carries neither or both of result and error".to_owned(),
704 )),
705 }
706}
707
708#[cfg(test)]
709mod nonzero_tests {
710 use super::{Common, DEFAULT_TIMEOUT_MS, NonZeroU64, PathBuf, Target};
711
712 #[test]
713 fn nonzero_timeout_readback() -> Result<(), Box<dyn std::error::Error>> {
714 let cases = [
715 (None, DEFAULT_TIMEOUT_MS.get()),
716 (Some(1), 1),
717 (Some(37), 37),
718 (Some(u64::MAX), u64::MAX),
719 ];
720 for (given, expected) in cases {
721 let timeout = given
722 .map(|value| NonZeroU64::new(value).ok_or("zero timeout"))
723 .transpose()?;
724 let common = Common {
725 target: Some(Target::Socket(PathBuf::from("unused.sock"))),
726 timeout: std::hint::black_box(timeout),
727 };
728 let (_, actual) = common.finish()?;
729 assert_eq!(actual.as_millis(), u128::from(expected));
730 }
731 Ok(())
732 }
733}
734
735#[cfg(test)]
736mod json_text_refusal_tests {
737 use super::{KNOWN_REFUSALS, Refusal, Reply, finish, refused};
738 use std::process::ExitCode;
739
740 #[test]
741 fn json_text_invalid_is_a_named_client_refusal() {
742 let refusal = Refusal {
743 code: "json_text_invalid".to_owned(),
744 message: "value is not JSON".to_owned(),
745 path: Some("/payload".to_owned()),
746 verified: None,
747 };
748 assert!(KNOWN_REFUSALS.contains(&refusal.code.as_str()));
749 assert_eq!(
750 refused(&refusal),
751 "haematite: refused json_text_invalid: value is not JSON at /payload"
752 );
753 assert_eq!(finish(Ok(Reply::Refused(refusal))), ExitCode::from(2));
754 }
755}
756
757#[cfg(test)]
758mod access_refusal_tests {
759 use super::KNOWN_REFUSALS;
760
761 #[test]
764 fn every_access_refusal_is_known_not_to_have_applied() {
765 for code in [
766 "route_unclassified",
767 "permission_refused",
768 "held_for_approval",
769 "pass_missing",
770 "lys_could_not_be_asked",
771 "signature_refused",
772 "pass_expired",
773 ] {
774 assert!(KNOWN_REFUSALS.contains(&code), "{code} is not known");
775 }
776 }
777}