//! Middle-level API for RPC connections //! //! This module focuses around the `RpcConn` type, which supports sending RPC requests //! and matching them with their responses. use std::{ io::{self, BufReader}, path::PathBuf, sync::{Arc, Mutex}, }; use crate::{ llconn, msgs::{ request::InvalidRequestError, response::{ResponseKind, RpcError, ValidatedResponse}, AnyRequestId, ObjectId, }, util::define_from_for_arc, }; mod auth; mod connimpl; use crate::util::Utf8CString; pub use connimpl::RpcConn; /// A handle to an open request. /// /// These handles are created with [`RpcConn::execute_with_handle`]. #[derive(educe::Educe)] #[educe(Debug)] pub struct RequestHandle { /// The underlying `Receiver` that we'll use to get updates for this request /// /// It's wrapped in a `Mutex` to prevent concurrent calls to `Receiver::wait_on_message_for`. // // NOTE: As an alternative to using a Mutex here, we _could_ remove // the restriction from `wait_on_message_for` that says that only one thread // may be waiting on a given request ID at once. But that would introduce // complexity to the implementation, // and it's not clear that the benefit would be worth it. #[educe(Debug(ignore))] conn: Mutex>, /// The ID of this request. id: AnyRequestId, } // TODO RPC: Possibly abolish these types. // // I am keeping this for now because it makes it more clear that we can never reinterpret // a success as an update or similar. // // I am not at all pleased with these types; we should revise them. // // TODO RPC: Possibly, all of these should be reconstructed // from their serde_json::Values rather than forwarded verbatim. // (But why would we our json to be more canonical than arti's? See #1491.) // // DODGY TYPES BEGIN: TODO RPC /// A Success Response from Arti, indicating that a request was successful. /// /// This is the complete message, including `id` and `result` fields. // // Invariant: it is valid JSON and contains no NUL bytes or newlines. // TODO RPC: check that the newline invariant is enforced in constructors. #[derive(Clone, Debug, derive_more::AsRef, derive_more::Into)] #[as_ref(forward)] pub struct SuccessResponse(Utf8CString); /// An Update Response from Arti, with information about the progress of a request. /// /// This is the complete message, including `id` and `update` fields. // // Invariant: it is valid JSON and contains no NUL bytes or newlines. // TODO RPC: check that the newline invariant is enforced in constructors. // TODO RPC consider changing this to CString. #[derive(Clone, Debug, derive_more::AsRef, derive_more::Into)] #[as_ref(forward)] pub struct UpdateResponse(Utf8CString); /// A Error Response from Arti, indicating that an error occurred. /// /// (This is the complete message, including the `error` field. /// It also an `id` if it /// is in response to a request; but not if it is a fatal protocol error.) // // Invariant: Does not contain a NUL. (Safe to convert to CString.) // // Invariant: This field MUST encode a response whose body is an RPC error. // // Otherwise the `decode` method may panic. // // TODO RPC: check that the newline invariant is enforced in constructors. #[derive(Clone, Debug, derive_more::AsRef, derive_more::Into)] #[as_ref(forward)] // TODO: If we keep this, it should implement Error. pub struct ErrorResponse(Utf8CString); impl ErrorResponse { /// Construct an ErrorResponse from the Error reply. /// /// This not a From impl because we want it to be crate-internal. pub(crate) fn from_validated_string(s: Utf8CString) -> Self { ErrorResponse(s) } /// Try to interpret this response as an [`RpcError`]. pub fn decode(&self) -> RpcError { crate::msgs::response::try_decode_response_as_err(self.0.as_ref()) .expect("Could not decode response that was already decoded as an error?") .expect("Could not extract error from response that was already decoded as an error?") } } impl std::fmt::Display for ErrorResponse { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let e = self.decode(); write!(f, "Peer said {:?}", e.message()) } } /// A final response -- that is, the last one that we expect to receive for a request. /// type FinalResponse = Result; /// Any of the three types of Arti responses. #[derive(Clone, Debug)] #[allow(clippy::exhaustive_structs)] pub enum AnyResponse { /// The request has succeeded; no more response will be given. Success(SuccessResponse), /// The request has failed; no more response will be given. Error(ErrorResponse), /// An incremental update; more messages may arrive. Update(UpdateResponse), } // TODO RPC: DODGY TYPES END. /// Information about how to construct a connection to an Arti instance. pub struct RpcConnBuilder { /// A path to a unix domain socket at which Arti is listening. // TODO RPC: Right now this is the only kind of supported way to connect. unix_socket: PathBuf, // todo RPC: include selector for how to connect. // // TODO RPC: Possibly kill off the builder entirely. } // TODO: For FFI purposes, define a slightly higher level API that // tries to do this all at once, possibly decoding a "connect string" // and some optional secret stuff? impl RpcConnBuilder { /// Create a Builder from a connect string. /// /// (Right now the only supported string type is "unix:" followed by a path.) // // TODO RPC: Should this take an OsString? // // TODO RPC: Specify the actual metaformat that we want to use here. // Possibly turn this into a K=V sequence ... or possibly, just // turn it into a JSON object. pub fn from_connect_string(s: &str) -> Result { let (kind, location) = s .split_once(':') .ok_or(BuilderError::InvalidConnectString)?; if kind == "unix" { Ok(Self::new_unix_socket(location)) } else { Err(BuilderError::InvalidConnectString) } } /// Create a Builder to connect to a unix socket at a given path. /// /// Note that this function may succeed even in environments where /// unix sockets are not supported. On these environments, /// the `connect` attempt will later fail with `SchemeNotSupported`. pub fn new_unix_socket(addr: impl Into) -> Self { Self { unix_socket: addr.into(), } } /// Try to connect to an Arti process as specified by this Builder. pub fn connect(&self) -> Result { #[cfg(not(unix))] { return Err(ConnectError::SchemeNotSupported); } #[cfg(unix)] { let sock = std::os::unix::net::UnixStream::connect(&self.unix_socket) .map_err(|e| ConnectError::CannotConnect(Arc::new(e)))?; let sock_dup = sock .try_clone() .map_err(|e| ConnectError::CannotConnect(Arc::new(e)))?; let mut conn = RpcConn::new( llconn::Reader::new(Box::new(BufReader::new(sock))), llconn::Writer::new(Box::new(sock_dup)), ); let session_id = conn.authenticate_inherent("inherent:unix_path")?; conn.session = Some(session_id); Ok(conn) } } } impl AnyResponse { /// Convert `v` into `AnyResponse`. fn from_validated(v: ValidatedResponse) -> Self { // TODO RPC, Perhaps unify AnyResponse with ValidatedResponse, once we are sure what // AnyResponse should look like. match v.meta.kind { ResponseKind::Error => AnyResponse::Error(ErrorResponse::from_validated_string(v.msg)), ResponseKind::Success => AnyResponse::Success(SuccessResponse(v.msg)), ResponseKind::Update => AnyResponse::Update(UpdateResponse(v.msg)), } } /// Consume this `AnyResponse`, and return its internal string. #[cfg(feature = "ffi")] pub(crate) fn into_string(self) -> Utf8CString { match self { AnyResponse::Success(m) => m.into(), AnyResponse::Error(m) => m.into(), AnyResponse::Update(m) => m.into(), } } } impl RpcConn { /// Return the ObjectId for the negotiated Session. /// /// Nearly all RPC methods require a Session, or some other object /// accessed via the session. /// /// (This function will only return None if no authentication has been performed. /// TODO RPC: It is not currently possible to make an unauthenticated connection.) pub fn session(&self) -> Option<&ObjectId> { self.session.as_ref() } /// Run a command, and wait for success or failure. /// /// Note that this function will return `Err(.)` only if sending the command or getting a /// response failed. If the command was sent successfully, and Arti reported an error in response, /// this function returns `Ok(Err(.))`. /// /// Note that the command does not need to include an `id` field. If you omit it, /// one will be generated. pub fn execute(&self, cmd: &str) -> Result { let hnd = self.execute_with_handle(cmd)?; hnd.wait() } /// Cancel a request by ID. pub fn cancel(&self, _id: &AnyRequestId) -> Result<(), ProtoError> { todo!() } /// Like `execute`, but don't wait. This lets the caller see the /// request ID and maybe cancel it. pub fn execute_with_handle(&self, cmd: &str) -> Result { self.send_request(cmd) } /// As execute(), but run update_cb for every update we receive. pub fn execute_with_updates( &self, cmd: &str, mut update_cb: F, ) -> Result where F: FnMut(UpdateResponse) + Send + Sync, { let hnd = self.execute_with_handle(cmd)?; loop { match hnd.wait_with_updates()? { AnyResponse::Success(s) => return Ok(Ok(s)), AnyResponse::Error(e) => return Ok(Err(e)), AnyResponse::Update(u) => update_cb(u), } } } // TODO RPC: shutdown() on the socket on Drop. } impl RequestHandle { /// Return the ID of this request, to help cancelling it. pub fn id(&self) -> &AnyRequestId { &self.id } /// Wait for success or failure, and return what happened. /// /// (Ignores any update messages that are received.) /// /// Note that this function will return `Err(.)` only if sending the command or getting a /// response failed. If the command was sent successfully, and Arti reported an error in response, /// this function returns `Ok(Err(.))`. pub fn wait(self) -> Result { loop { match self.wait_with_updates()? { AnyResponse::Success(s) => return Ok(Ok(s)), AnyResponse::Error(e) => return Ok(Err(e)), AnyResponse::Update(_) => {} } } } /// Wait for the next success, failure, or update from this handle. /// /// Note that this function will return `Err(.)` only if sending the command or getting a /// response failed. If the command was sent successfully, and Arti reported an error in response, /// this function returns `Ok(AnyResponse::Error(.))`. /// /// You may call this method on the same `RequestHandle` from multiple threads. /// If you do so, those calls will receive responses (or errors) in an unspecified order. /// /// If this function returns Success or Error, then you shouldn't call it again. /// All future calls to this function will fail with `CmdError::RequestCancelled`. /// (TODO RPC: Maybe rename that error.) pub fn wait_with_updates(&self) -> Result { let conn = self.conn.lock().expect("Poisoned lock"); let validated = conn.wait_on_message_for(&self.id)?; Ok(AnyResponse::from_validated(validated)) } // TODO RPC: Cancel on drop. // TODO RPC: way to drop without cancelling. // TODO RPC: Sketch out how we would want to do this in an async world, // or with poll } /// An error (or other condition) that has caused an RPC connection to shut down. #[derive(Clone, Debug, thiserror::Error)] #[non_exhaustive] pub enum ShutdownError { /// Io error occurred while reading. #[error("Unable to read response: {0}")] Read(#[source] Arc), /// Io error occurred while writing. #[error("Unable to write request: {0}")] Write(#[source] Arc), /// Something was wrong with Arti's responses; this is a protocol violation. #[error("Arti sent a message that didn't conform to the RPC protocol: {0}")] ProtocolViolated(String), /// Arti has told us that we violated the protocol somehow. #[error("Arti reported a fatal error: {0:?}")] ProtocolViolationReport(ErrorResponse), /// The underlying connection closed. /// /// This probably means that Arti has shut down. #[error("Connection closed")] ConnectionClosed, } impl From for ShutdownError { fn from(value: crate::msgs::response::DecodeResponseError) -> Self { use crate::msgs::response::DecodeResponseError::*; use ShutdownError as E; match value { JsonProtocolViolation(e) => E::ProtocolViolated(e.to_string()), ProtocolViolation(s) => E::ProtocolViolated(s.to_string()), Fatal(rpc_err) => E::ProtocolViolationReport(rpc_err), } } } /// An error that has occurred while launching an RPC command. #[derive(Clone, Debug, thiserror::Error)] #[non_exhaustive] pub enum ProtoError { /// The RPC connection failed, or was closed by the other side. #[error("RPC connection is shut down: {0}")] Shutdown(#[from] ShutdownError), /// There was a problem in the request we tried to send. #[error("Invalid request: {0}")] InvalidRequest(#[from] InvalidRequestError), /// We tried to send a request with an ID that was already pending. #[error("Request ID already in use.")] RequestIdInUse, /// We tried to wait for a request that had already been cancelled. // // TODO RPC: Possibly this should be impossible. Revisit when I implement // cancellation here. #[error("Request already cancelled.")] RequestCancelled, /// We tried to wait for the same request more than once. /// /// (This should be impossible.) #[error("Internal error: waiting on the same request more than once at a time.")] DuplicateWait, /// We got an internal error while trying to encode an RPC request. /// /// (This should be impossible.) #[error("Internal error while encoding request: {0}")] CouldNotEncode(#[source] Arc), } /// An error while trying to connect to the Arti process. #[derive(Clone, Debug, thiserror::Error)] #[non_exhaustive] pub enum ConnectError { /// We specified a prefix to our connect string, but we don't /// have run-time support for it. #[error("Selected connection scheme was not supported in this build")] SchemeNotSupported, /// IO error while connecting to Arti. #[error("Unable to make a connection: {0}")] CannotConnect(#[source] Arc), /// One of our authentication messages was rejected. #[error("Arti rejected our authentication: {0:?}")] AuthenticationRejected(ErrorResponse), /// We couldn't decode one of the responses we got. #[error("Message not in expected format: {0:?}")] BadMessage(#[source] Arc), /// A protocol error occurred during negotiations. #[error("Error while negotiating with Arti: {0}")] ProtoError(#[from] ProtoError), } define_from_for_arc!(serde_json::Error => ConnectError [BadMessage]); /// An error occurred while trying to construct or manipulate a #[derive(Clone, Debug, thiserror::Error)] #[non_exhaustive] pub enum BuilderError { /// We couldn't decode a provided connect string. #[error("Invalid connect string.")] InvalidConnectString, } #[cfg(test)] mod test { // @@ begin test lint list maintained by maint/add_warning @@ #![allow(clippy::bool_assert_comparison)] #![allow(clippy::clone_on_copy)] #![allow(clippy::dbg_macro)] #![allow(clippy::mixed_attributes_style)] #![allow(clippy::print_stderr)] #![allow(clippy::print_stdout)] #![allow(clippy::single_char_pattern)] #![allow(clippy::unwrap_used)] #![allow(clippy::unchecked_duration_subtraction)] #![allow(clippy::useless_vec)] #![allow(clippy::needless_pass_by_value)] //! use std::{sync::atomic::AtomicUsize, thread, time::Duration}; use io::{BufRead as _, Write as _}; use rand::{seq::SliceRandom as _, Rng as _, SeedableRng as _}; use tor_basic_utils::{test_rng::testing_rng, RngExt as _}; use crate::msgs::request::{JsonMap, Request, ValidatedRequest}; use super::*; /// helper: Return a dummy RpcConn, along with a socketpair for it to talk to. fn dummy_connected() -> (RpcConn, socketpair::SocketpairStream) { let (s1, s2) = socketpair::socketpair_stream().unwrap(); let s1_w = s1.try_clone().unwrap(); let s1_r = io::BufReader::new(s1); let conn = RpcConn::new(llconn::Reader::new(s1_r), llconn::Writer::new(s1_w)); (conn, s2) } fn write_val(w: &mut impl io::Write, v: &serde_json::Value) { let mut enc = serde_json::to_string(v).unwrap(); enc.push('\n'); w.write_all(enc.as_bytes()).unwrap(); } #[test] fn simple() { let (conn, sock) = dummy_connected(); let user_thread = thread::spawn(move || { let response1 = conn .execute(r#"{"obj":"fred","method":"arti:x-frob","params":{}}"#) .unwrap(); (response1, conn) }); let fake_arti_thread = thread::spawn(move || { let mut sock = BufReader::new(sock); let mut s = String::new(); let _len = sock.read_line(&mut s).unwrap(); let request = ValidatedRequest::from_string_strict(s.as_ref()).unwrap(); let response = serde_json::json!({ "id": request.id().clone(), "result": { "xyz" : 3 } }); write_val(sock.get_mut(), &response); sock // prevent close }); let _sock = fake_arti_thread.join().unwrap(); let (response, _conn) = user_thread.join().unwrap(); let map = response.unwrap().deserialize_as::().unwrap(); assert_eq!(map.get("xyz"), Some(&serde_json::Value::Number(3.into()))); } #[test] fn complex() { use std::sync::atomic::Ordering::SeqCst; let n_threads = 16; let n_commands_per_thread = 4096; let n_commands_total = n_threads * n_commands_per_thread; let n_completed = Arc::new(AtomicUsize::new(0)); let (conn, sock) = dummy_connected(); let conn = Arc::new(conn); let mut user_threads = Vec::new(); let mut rng = testing_rng(); // ------- // User threads: Make a bunch of requests. for th_idx in 0..n_threads { let conn = Arc::clone(&conn); let n_completed = Arc::clone(&n_completed); let mut rng = rand_chacha::ChaCha12Rng::from_seed(rng.gen()); let th = thread::spawn(move || { for cmd_idx in 0..n_commands_per_thread { // We are spawning a bunch of worker threads, each of which will run a number of // commands in sequence. Each command will be a request that gets optional // updates, and an error or a success. // We will double-check that each request gets the response it asked for. let s = format!("{}:{}", th_idx, cmd_idx); let want_updates: bool = rng.gen(); let want_failure: bool = rng.gen(); let req = serde_json::json!({ "obj":"fred", "method":"arti:x-echo", "meta": { "updates": want_updates, }, "params": { "val": &s, "fail": want_failure, }, }); let req = serde_json::to_string(&req).unwrap(); // Wait for a final response, processing updates if we asked for them. let mut n_updates = 0; let outcome = conn .execute_with_updates(&req, |_update| { n_updates += 1; }) .unwrap(); assert_eq!(n_updates > 0, want_updates); // See if we liked the final response. if want_failure { let e = outcome.unwrap_err().decode(); assert_eq!(e.message(), "You asked me to fail"); assert_eq!(i32::from(e.code()), 33); assert_eq!( e.kinds_iter().collect::>(), vec!["Example".to_string()] ); } else { let success = outcome.unwrap(); let map = success.deserialize_as::().unwrap(); assert_eq!(map.get("echo"), Some(&serde_json::Value::String(s))); } n_completed.fetch_add(1, SeqCst); if rng.gen::() < 0.02 { thread::sleep(Duration::from_millis(3)); } } }); user_threads.push(th); } #[derive(serde::Deserialize, Debug)] struct Echo { val: String, fail: bool, } // ----- // Worker thread: handles user requests. let worker_rng = rand_chacha::ChaCha12Rng::from_seed(rng.gen()); let worker_thread = thread::spawn(move || { let mut rng = worker_rng; let mut sock = BufReader::new(sock); let mut pending: Vec> = Vec::new(); let mut n_received = 0; // How many requests do we buffer before we shuffle them and answer them out-of-order? let scramble_factor = 7; // After receiving how many requests do we stop shuffling requests? // // (Our shuffling algorithm can deadlock us otherwise.) let scramble_threshold = n_commands_total - (n_commands_per_thread + 1) * scramble_factor; 'outer: loop { let flush_pending_at = if n_received >= scramble_threshold { 1 } else { scramble_factor }; // Queue a handful of requests in "pending" while pending.len() < flush_pending_at { let mut buf = String::new(); if sock.read_line(&mut buf).unwrap() == 0 { break 'outer; } n_received += 1; let req: Request = serde_json::from_str(&buf).unwrap(); pending.push(req); } // Handle the requests in "pending" in random order. let mut handling = std::mem::take(&mut pending); handling.shuffle(&mut rng); for req in handling { if req.meta.unwrap_or_default().updates { let n_updates = rng.gen_range_checked(1..4).unwrap(); for _ in 0..n_updates { let up = serde_json::json!({ "id": req.id.clone(), "update": { "hello": req.params.val.clone(), } }); write_val(sock.get_mut(), &up); } } let response = if req.params.fail { serde_json::json!({ "id": req.id.clone(), "error": { "message": "You asked me to fail", "code": 33, "kinds": ["Example"], "data": req.params.val }, }) } else { serde_json::json!({ "id": req.id.clone(), "result": { "echo": req.params.val } }) }; write_val(sock.get_mut(), &response); } } }); drop(conn); for t in user_threads { t.join().unwrap(); } worker_thread.join().unwrap(); assert_eq!(n_completed.load(SeqCst), n_commands_total); } #[test] fn arti_socket_closed() { // Here we send a bunch of requests and then close the socket without answering them. // // Every request should get a ProtoError::Shutdown. let n_threads = 16; let (conn, sock) = dummy_connected(); let conn = Arc::new(conn); let mut user_threads = Vec::new(); for _ in 0..n_threads { let conn = Arc::clone(&conn); let th = thread::spawn(move || { // We are spawning a bunch of worker threads, each of which will run a number of // We will double-check that each request gets the response it asked for. let req = serde_json::json!({ "obj":"fred", "method":"arti:x-echo", "params":{} }); let req = serde_json::to_string(&req).unwrap(); let outcome = conn.execute(&req); if !matches!( &outcome, Err(ProtoError::Shutdown(ShutdownError::Write(_))) | Err(ProtoError::Shutdown(ShutdownError::Read(_))), ) { dbg!(&outcome); } assert!(matches!( outcome, Err(ProtoError::Shutdown(ShutdownError::Write(_))) | Err(ProtoError::Shutdown(ShutdownError::Read(_))) | Err(ProtoError::Shutdown(ShutdownError::ConnectionClosed)) )); }); user_threads.push(th); } drop(sock); for t in user_threads { t.join().unwrap(); } } /// Send a bunch of requests and then send back a single reply. /// /// That reply should cause every request to get closed. fn proto_err_with_msg(msg: &str, outcome_ok: F) where F: Fn(ProtoError) -> bool, { let n_threads = 16; let (conn, mut sock) = dummy_connected(); let conn = Arc::new(conn); let mut user_threads = Vec::new(); for _ in 0..n_threads { let conn = Arc::clone(&conn); let th = thread::spawn(move || { // We are spawning a bunch of worker threads, each of which will run a number of // We will double-check that each request gets the response it asked for. let req = serde_json::json!({ "obj":"fred", "method":"arti:x-echo", "params":{} }); let req = serde_json::to_string(&req).unwrap(); conn.execute(&req) }); user_threads.push(th); } sock.write_all(msg.as_bytes()).unwrap(); for t in user_threads { let outcome = t.join().unwrap(); assert!(outcome_ok(outcome.unwrap_err())); } } #[test] fn syntax_error() { proto_err_with_msg("this is not json\n", |outcome| { matches!( outcome, ProtoError::Shutdown(ShutdownError::ProtocolViolated(_)) ) }); } #[test] fn fatal_error() { let j = serde_json::json!({ "error":{ "message": "This test is doomed", "code": 413, "kinds": ["Example"], "data": {} }, }); let mut s = serde_json::to_string(&j).unwrap(); s.push('\n'); proto_err_with_msg(&s, |outcome| { matches!( outcome, ProtoError::Shutdown(ShutdownError::ProtocolViolationReport(_)) ) }); } }