diff --git a/.changeset/validator-cookie-read-again-when-refused.md b/.changeset/validator-cookie-read-again-when-refused.md new file mode 100644 index 000000000..c857c7412 --- /dev/null +++ b/.changeset/validator-cookie-read-again-when-refused.md @@ -0,0 +1,15 @@ +--- +default: minor +--- + +# zingod-source, zingod: validator cookie read again when refused; a local validator's default cookie found + +- A cookie (`cookie_path`) is read on the first request and again after the validator answers + `401` or `403`: a validator restart's new cookie is used from the request after the refusal + (was 401 until zingod restarted). A cookie not yet written at boot is a retried request + failure, never a failed start; a cookie that exists and cannot be read fails the start. +- A `[[trusted_validators]]` entry with no credentials and a loopback `jsonrpc_address` sends no + credentials until the validator answers `401` or `403`, then the validator's default cookie, + `/zebra/.cookie` or `/zakura/.cookie`, whichever was written last + (neither = no auth; to pin one, set `cookie_path`). A validator at any other address is never + sent a local cookie. diff --git a/docs/example_configs/zebrad.toml b/docs/example_configs/zebrad.toml index 14e9456c9..22cada052 100644 --- a/docs/example_configs/zebrad.toml +++ b/docs/example_configs/zebrad.toml @@ -7,7 +7,8 @@ network = "Testnet" [rpc] # zingod's [[trusted_validators]] jsonrpc_address listen_addr = "127.0.0.1:18232" -# true → zebrad writes /.cookie; point zingod's [[trusted_validators]] cookie_path at it +# true → zebrad writes /.cookie on each start; zingod reads it again when refused (default +# cookie_dir, same user, loopback: found unset; else zingod's cookie_path) enable_cookie_auth = false cookie_dir = "" diff --git a/docs/example_configs/zingod.toml b/docs/example_configs/zingod.toml index c011e1473..38eafdb85 100644 --- a/docs/example_configs/zingod.toml +++ b/docs/example_configs/zingod.toml @@ -27,7 +27,8 @@ network = "testnet" # - down at boot or later = retried, never fatal [[trusted_validators]] jsonrpc_address = "127.0.0.1:18232" -# zebrad enable_cookie_auth = true +# No credentials + a loopback address = no auth until the validator refuses, then its default +# cookie: /zebra/.cookie or /zakura/.cookie (both = the one written last) # cookie_path = "/.cookie" # basic auth # user = "xxxxxx" diff --git a/packages/zingod-source/src/endpoint.rs b/packages/zingod-source/src/endpoint.rs index 049c1fb92..f4f77c165 100644 --- a/packages/zingod-source/src/endpoint.rs +++ b/packages/zingod-source/src/endpoint.rs @@ -3,9 +3,11 @@ //! //! - no probe: reachability = the caller's retry loop (validator down at boot ≠ fatal) +use std::io::Read as _; +use std::net::IpAddr; use std::num::NonZeroU32; -use std::path::PathBuf; -use std::time::Duration; +use std::path::{Path, PathBuf}; +use std::time::{Duration, SystemTime}; use crate::ResponseSizeCap; @@ -17,6 +19,9 @@ pub enum EndpointError { #[error("cannot read validator cookie {path}: {source}")] Cookie { path: String, source: std::io::Error }, + #[error("validator cookie {path} is empty")] + EmptyCookie { path: String }, + #[error("cannot read indexer TLS {role} {path}: {source}")] TlsFile { role: &'static str, path: String, source: std::io::Error }, } @@ -35,27 +40,137 @@ pub struct Endpoint { pub enum Auth { None, Cookie(PathBuf), + DefaultCookie(Vec), + Basic { user: String, password: String }, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum Credentials { + NoAuth, Basic { user: String, password: String }, } +impl Credentials { + pub(crate) fn basic(user: &str, password: &str) -> Self { + Self::Basic { user: user.to_owned(), password: password.to_owned() } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum Held { + Unread, + Known(Credentials), +} + +struct Cookie { + modified: Option, + contents: String, +} + +type Reading = (PathBuf, std::io::Result); + impl Auth { - /// Basic-auth pair sent (`None` = no auth) - /// - /// - `__cookie__:` prefix stripped if present (older validators, some packagers omit it) - pub(crate) fn credentials(&self) -> Result, EndpointError> { + /// - reads each cookie file on the blocking pool + pub(crate) async fn credentials(&self) -> Result { + let paths = self.cookie_paths().to_vec(); + if paths.is_empty() { + return self.sent(Vec::new()); + } + let read = tokio::task::spawn_blocking({ + let paths = paths.clone(); + move || read_cookies(&paths) + }); + let unread = |stopped: tokio::task::JoinError| { + let reading = |path| (path, Err(std::io::Error::other(stopped.to_string()))); + paths.into_iter().map(reading).collect() + }; + self.sent(read.await.unwrap_or_else(unread)) + } + + /// - reads each cookie file on the calling thread + pub(crate) fn startable(&self) -> Result<(), EndpointError> { + match self.before_challenge() { + Held::Known(_) => Ok(()), + Held::Unread => start_verdict(self.sent(read_cookies(self.cookie_paths()))), + } + } + + pub(crate) fn before_challenge(&self) -> Held { + match self { + Self::DefaultCookie(_) => Held::Known(Credentials::NoAuth), + fixed_or_explicit => fixed_or_explicit.fixed().map_or(Held::Unread, Held::Known), + } + } + + fn fixed(&self) -> Option { + match self { + Self::None => Some(Credentials::NoAuth), + Self::Basic { user, password } => Some(Credentials::basic(user, password)), + Self::Cookie(_) | Self::DefaultCookie(_) => None, + } + } + + fn cookie_paths(&self) -> &[PathBuf] { match self { - Self::None => Ok(None), - Self::Basic { user, password } => Ok(Some((user.clone(), password.clone()))), - Self::Cookie(path) => { - let contents = std::fs::read_to_string(path).map_err(|source| { - EndpointError::Cookie { path: path.display().to_string(), source } - })?; - let token = contents.trim(); - let token = token.strip_prefix("__cookie__:").unwrap_or(token); - Ok(Some(("__cookie__".to_string(), token.to_string()))) - } + Self::None | Self::Basic { .. } => &[], + Self::Cookie(path) => std::slice::from_ref(path), + Self::DefaultCookie(candidates) => candidates, } } + + fn sent(&self, readings: Vec) -> Result { + if let Some(fixed) = self.fixed() { + return Ok(fixed); + } + let absent_is_no_auth = matches!(self, Self::DefaultCookie(_)); + let present = readings + .into_iter() + .filter(|(_, cookie)| !(absent_is_no_auth && cookie.as_ref().is_err_and(absent))) + .map(|(path, cookie)| match cookie { + Ok(cookie) => Ok((path, cookie)), + Err(source) => { + Err(EndpointError::Cookie { path: path.display().to_string(), source }) + } + }) + .collect::, _>>()?; + let newest = present.into_iter().rev().max_by_key(|(_, cookie)| cookie.modified); + newest.map_or(Ok(Credentials::NoAuth), |(path, cookie)| basic_auth(&path, &cookie)) + } +} + +fn absent(error: &std::io::Error) -> bool { + error.kind() == std::io::ErrorKind::NotFound +} + +fn basic_auth(path: &Path, cookie: &Cookie) -> Result { + let token = cookie.contents.trim(); + let token = token.strip_prefix("__cookie__:").unwrap_or(token); + if token.is_empty() { + return Err(EndpointError::EmptyCookie { path: path.display().to_string() }); + } + Ok(Credentials::basic("__cookie__", token)) +} + +fn start_verdict(read: Result) -> Result<(), EndpointError> { + match read { + Err(EndpointError::Cookie { source, .. }) if absent(&source) => Ok(()), + Err(EndpointError::EmptyCookie { .. }) => Ok(()), + settled => settled.map(drop), + } +} + +/// - reads each of `paths` +fn read_cookies(paths: &[PathBuf]) -> Vec { + paths.iter().map(|path| (path.clone(), read_cookie(path))).collect() +} + +/// - reads `path` +fn read_cookie(path: &Path) -> std::io::Result { + let mut file = std::fs::File::open(path)?; + let modified = file.metadata()?.modified().ok(); + let mut contents = String::new(); + file.read_to_string(&mut contents)?; + Ok(Cookie { modified, contents }) } /// - `read` = silence, never total duration (multi-MB block over a slow link must finish) @@ -92,6 +207,10 @@ impl Default for LinkLimits { /// `http://{address}`, hostname kept, not pre-resolved (reqwest resolves per connection → /// follows a validator whose IP changes, e.g. a restarted pod) pub(crate) fn validator_url(address: &str) -> Result { + Ok(parsed(address)?.as_str().trim_end_matches('/').to_string()) +} + +fn parsed(address: &str) -> Result { let invalid = |reason: &str| EndpointError::Address { address: address.to_string(), reason: reason.to_string(), @@ -105,36 +224,176 @@ pub(crate) fn validator_url(address: &str) -> Result { if url.path() != "/" || url.query().is_some() { return Err(invalid("unexpected path or query")); } - Ok(url.as_str().trim_end_matches('/').to_string()) + Ok(url) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Reach { + Loopback, + Private, + Public, + Named, +} + +impl Reach { + pub fn of_ip(ip: IpAddr) -> Self { + match ip.to_canonical() { + ip if ip.is_loopback() => Self::Loopback, + IpAddr::V4(ip) if ip.is_private() => Self::Private, + IpAddr::V6(ip) if ip.is_unique_local() => Self::Private, + _ => Self::Public, + } + } + + pub fn of_validator(address: &str) -> Result { + let url = parsed(address)?; + let host = url.host_str().unwrap_or_default(); + let host = host.trim_start_matches('[').trim_end_matches(']'); + let named = || match host.trim_end_matches('.') { + "localhost" => Self::Loopback, + _ => Self::Named, + }; + Ok(host.parse().map_or_else(|_| named(), Self::of_ip)) + } } #[cfg(test)] mod tests { use super::*; - use std::io::Write as _; - /// User always `__cookie__`, token trimmed with or without its prefix (packagers write both) + fn written(contents: &str, age_secs: u64) -> std::io::Result { + let modified = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000 - age_secs); + Ok(Cookie { modified: Some(modified), contents: contents.to_owned() }) + } + + fn failed(kind: std::io::ErrorKind) -> std::io::Result { + Err(std::io::Error::from(kind)) + } + + fn basic() -> Auth { + Auth::Basic { user: "user".to_owned(), password: "pass".to_owned() } + } + + fn outcome(sent: Result) -> String { + match sent { + Ok(Credentials::NoAuth) => "no auth".to_owned(), + Ok(Credentials::Basic { user, password }) => format!("{user}:{password}"), + Err(EndpointError::Cookie { path, .. }) => format!("unreadable {path}"), + Err(EndpointError::EmptyCookie { path }) => format!("empty {path}"), + Err(other) => format!("{other}"), + } + } + #[test] - fn cookie_credentials_are_the_cookie_user_and_its_trimmed_token() { - for contents in ["__cookie__:sekrit", " sekrit \n"] { - let mut file = tempfile::NamedTempFile::new().expect("temp file"); - write!(file, "{contents}").expect("write cookie"); + fn credentials_sent_follow_the_cookies_read() { + use std::io::ErrorKind::{NotFound, PermissionDenied}; + let explicit = Auth::Cookie("zebra".into()); + let default = Auth::DefaultCookie(vec!["zebra".into(), "zakura".into()]); + #[rustfmt::skip] + let cases = [ + ("explicit, prefixed", &explicit, vec![written("__cookie__:one", 0)], "__cookie__:one"), + ("explicit, bare and untrimmed", &explicit, vec![written(" one \n", 0)], "__cookie__:one"), + ("explicit, absent", &explicit, vec![failed(NotFound)], "unreadable zebra"), + ("explicit, empty", &explicit, vec![written("", 0)], "empty zebra"), + ("default, neither", &default, vec![failed(NotFound), failed(NotFound)], "no auth"), + ("default, zebra alone", &default, vec![written("one", 0), failed(NotFound)], "__cookie__:one"), + ("default, zakura newer", &default, vec![written("one", 9), written("two", 0)], "__cookie__:two"), + ("default, same age: the first", &default, vec![written("one", 5), written("two", 5)], "__cookie__:one"), + ("default, one unreadable", &default, vec![written("one", 0), failed(PermissionDenied)], "unreadable zakura"), + ("default, the newest empty", &default, vec![written("one", 9), written("", 0)], "empty zakura"), + ]; + for (case, auth, cookies, expected) in cases { + let readings = auth.cookie_paths().iter().cloned().zip(cookies).collect(); + assert_eq!(outcome(auth.sent(readings)), expected, "{case}"); + } + } - let auth = - Auth::Cookie(file.path().to_owned()).credentials().map_err(|e| e.to_string()); + #[test] + fn only_an_explicit_cookie_is_unknown_before_the_first_request() { + let cases = [ + ("none: no auth", Auth::None, Held::Known(Credentials::NoAuth)), + ("basic: the pair", basic(), Held::Known(Credentials::basic("user", "pass"))), + ("explicit cookie: to be read", Auth::Cookie("zebra".into()), Held::Unread), + ( + "default cookie: no auth until refused", + Auth::DefaultCookie(vec![]), + Held::Known(Credentials::NoAuth), + ), + ]; + for (case, auth, expected) in cases { + assert_eq!(auth.before_challenge(), expected, "{case}"); + } + } - let token = Some(("__cookie__".to_owned(), "sekrit".to_owned())); - assert_eq!(auth, Ok(token), "cookie file {contents:?}"); + #[test] + fn a_validator_address_reaches_what_its_parsed_host_names() { + #[rustfmt::skip] + let cases = [ + ("127.0.0.1:8232", Reach::Loopback), + ("127.1:8232", Reach::Loopback), + ("user@127.0.0.1:8232", Reach::Loopback), + ("[::1]:8232", Reach::Loopback), + ("[::ffff:127.0.0.1]:8232", Reach::Loopback), + ("localhost:8232", Reach::Loopback), + ("LOCALHOST.:8232", Reach::Loopback), + ("10.0.0.5:8232", Reach::Private), + ("192.168.1.9:8232", Reach::Private), + ("[fd00::1]:8232", Reach::Private), + ("8.8.8.8:8232", Reach::Public), + ("[2001:4860::8888]:8232", Reach::Public), + ("zebra-eu:8232", Reach::Named), + ("localhost.example:8232", Reach::Named), + ]; + for (address, expected) in cases { + let reach = Reach::of_validator(address).map_err(|error| error.to_string()); + assert_eq!(reach, Ok(expected), "{address}"); } + let unparsed = Reach::of_validator("zebrad"); + assert!(matches!(unparsed, Err(EndpointError::Address { .. })), "no port: {unparsed:?}"); } #[test] - fn an_unreadable_cookie_is_refused_naming_it() { - let missing = Auth::Cookie(PathBuf::from("/nonexistent/cookie")).credentials(); + fn only_a_cookie_that_can_never_be_read_refuses_the_start() { + use std::io::ErrorKind::{IsADirectory, NotFound, PermissionDenied}; + let unreadable = |kind| { + let source = std::io::Error::from(kind); + Err(EndpointError::Cookie { path: "zebra".to_owned(), source }) + }; + let cases = [ + ("no auth", Ok(Credentials::NoAuth), true), + ("a cookie", Ok(Credentials::basic("__cookie__", "one")), true), + ("not yet written", unreadable(NotFound), true), + ( + "caught mid-write", + Err(EndpointError::EmptyCookie { path: "zebra".to_owned() }), + true, + ), + ("another user's", unreadable(PermissionDenied), false), + ("a directory", unreadable(IsADirectory), false), + ]; + for (case, read, starts) in cases { + assert_eq!(start_verdict(read).is_ok(), starts, "{case}"); + } + } + + #[tokio::test] + async fn the_newer_default_cookie_on_disk_is_the_one_sent() { + let dir = tempfile::tempdir().expect("the OS temp directory is writable"); + let [zebra, zakura] = ["zebra.cookie", "zakura.cookie"].map(|name| dir.path().join(name)); + let default = Auth::DefaultCookie(vec![zebra.clone(), zakura.clone()]); + let started = |cookie: &Path, token: &str, at_secs: u64| { + std::fs::write(cookie, token)?; + let at = SystemTime::UNIX_EPOCH + Duration::from_secs(at_secs); + std::fs::File::options().write(true).open(cookie)?.set_modified(at) + }; - let named = - matches!(&missing, Err(EndpointError::Cookie { path, .. }) if path.ends_with("cookie")); - assert!(named, "/nonexistent/cookie: {missing:?}"); + assert_eq!(outcome(default.credentials().await), "no auth", "neither started"); + started(&zebra, "__cookie__:one", 100).expect("the temp directory is writable"); + assert_eq!(outcome(default.credentials().await), "__cookie__:one", "zebrad started"); + started(&zakura, "__cookie__:two", 200).expect("the temp directory is writable"); + assert_eq!(outcome(default.credentials().await), "__cookie__:two", "zakurad started later"); + started(&zebra, "__cookie__:three", 300).expect("the temp directory is writable"); + assert_eq!(outcome(default.credentials().await), "__cookie__:three", "zebrad restarted"); } #[test] diff --git a/packages/zingod-source/src/legacy_rpc.rs b/packages/zingod-source/src/legacy_rpc.rs index 222b777f2..7d4c5050f 100644 --- a/packages/zingod-source/src/legacy_rpc.rs +++ b/packages/zingod-source/src/legacy_rpc.rs @@ -28,10 +28,11 @@ impl ZcashLegacyRpcClient { /// Validator at `endpoint`, unprobed (down at boot = the first call's failure + the caller's /// retry, never a boot failure) pub fn new(endpoint: &Endpoint) -> Result { + endpoint.auth.startable()?; let rpc = RpcClient::new(RpcClientConfig { url: validator_url(&endpoint.address)?, name: endpoint.address.clone(), - auth: endpoint.auth.credentials()?, + auth: endpoint.auth.clone(), timeouts: endpoint.timeouts, limits: endpoint.limits, }); @@ -395,11 +396,39 @@ fn raw_batches(listed: &[MempoolListed], budget: u64) -> Vec<&[MempoolListed]> { #[cfg(test)] mod tests { use super::*; + use crate::Auth; fn rpc(code: i64, message: &str) -> RpcError { RpcError::Rpc { code, message: message.to_string() } } + #[test] + fn only_a_cookie_that_can_never_be_read_fails_the_start() { + let dir = tempfile::tempdir().expect("the OS temp directory is writable"); + let unwritten = dir.path().join("unwritten"); + let a_directory = dir.path().to_owned(); + let started = |auth: Auth| { + let endpoint = Endpoint { + address: "127.0.0.1:8232".to_owned(), + auth, + timeouts: crate::Timeouts::default(), + limits: crate::LinkLimits::default(), + }; + ZcashLegacyRpcClient::new(&endpoint).map(drop) + }; + let refused = + |start: &Result<(), EndpointError>| matches!(start, Err(EndpointError::Cookie { .. })); + + let cases = [ + ("not yet written", Auth::Cookie(unwritten), false), + ("a directory", Auth::Cookie(a_directory), true), + ]; + for (case, auth, fails) in cases { + let start = started(auth); + assert_eq!(refused(&start), fails, "{case}: {start:?}"); + } + } + /// Batches keep listing order, cover it exactly, close at 100 calls or before passing the /// budget (8 MiB here), an entry over it alone #[test] diff --git a/packages/zingod-source/src/lib.rs b/packages/zingod-source/src/lib.rs index 5dd4d78c4..7f489ebb6 100644 --- a/packages/zingod-source/src/lib.rs +++ b/packages/zingod-source/src/lib.rs @@ -33,7 +33,7 @@ pub use decode::{ decode_block, decode_prepared, decode_transaction, mined_transaction, prepare_transaction, DecodeError, Prepared, }; -pub use endpoint::{Auth, Endpoint, EndpointError, LinkLimits, Timeouts}; +pub use endpoint::{Auth, Endpoint, EndpointError, LinkLimits, Reach, Timeouts}; pub use error::{FailureMode, NonDomainError, QueryError}; pub use indexer_client::ZebraIndexerGrpcClient; pub use indexer_endpoint::ZebraIndexerTls; diff --git a/packages/zingod-source/src/rpc/client.rs b/packages/zingod-source/src/rpc/client.rs index 82c5cfd6c..c5a96f045 100644 --- a/packages/zingod-source/src/rpc/client.rs +++ b/packages/zingod-source/src/rpc/client.rs @@ -5,6 +5,7 @@ use std::num::NonZeroUsize; use std::sync::atomic::{AtomicI64, Ordering}; +use std::sync::{Mutex, PoisonError}; use serde::de::DeserializeOwned; use serde_json::Value; @@ -12,8 +13,9 @@ use zingod_primitives::protocol::MAX_BLOCK_BYTES; use super::envelope; use super::error::RpcError; +use crate::endpoint::{Credentials, Held}; use crate::metrics; -use crate::{LinkLimits, Timeouts}; +use crate::{Auth, LinkLimits, Timeouts}; /// JSON-RPC framing around one result (id, version, keys) pub(crate) const REPLY_ENVELOPE_BYTES: usize = 1024; @@ -60,7 +62,7 @@ pub(crate) struct Call { pub(crate) struct RpcClientConfig { pub(crate) url: String, pub(crate) name: String, - pub(crate) auth: Option<(String, String)>, + pub(crate) auth: Auth, pub(crate) timeouts: Timeouts, pub(crate) limits: LinkLimits, } @@ -77,7 +79,8 @@ enum Wire { Http { client: reqwest::Client, url: String, - auth: Option<(String, String)>, + auth: Auth, + held: Mutex, }, /// In-process server: the same bytes, no socket (paused-clock tests) #[cfg(any(test, feature = "testing"))] @@ -104,7 +107,8 @@ impl RpcClient { ) .build() .expect("reqwest without TLS features + no fallible builder option = infallible build"); - let wire = Wire::Http { client, url: config.url, auth: config.auth }; + let held = Mutex::new(config.auth.before_challenge()); + let wire = Wire::Http { client, url: config.url, auth: config.auth, held }; Self::over(config.name, wire, limits) } @@ -161,21 +165,35 @@ impl RpcClient { } async fn send_http(&self, body: &Value) -> Result, RpcError> { - let (client, url, auth) = match &self.wire { - Wire::Http { client, url, auth } => (client, url, auth), + let (client, url, auth, held) = match &self.wire { + Wire::Http { client, url, auth, held } => (client, url, auth, held), #[cfg(any(test, feature = "testing"))] Wire::Served(serve) => { return self.read_served(serve(body.to_string().into_bytes()).await?) } }; + let lock = || held.lock().unwrap_or_else(PoisonError::into_inner); + let held_now = lock().clone(); + let credentials = match held_now { + Held::Known(credentials) => credentials, + Held::Unread => { + let read = auth.credentials().await?; + *lock() = Held::Known(read.clone()); + read + } + }; let mut request = client.post(url).header("Content-Type", "application/json").body(body.to_string()); - if let Some((user, pass)) = auth { - request = request.basic_auth(user, Some(pass)); + if let Credentials::Basic { user, password } = &credentials { + request = request.basic_auth(user, Some(password)); } let response = request.send().await?; let status = response.status().as_u16(); + { + let mut held = lock(); + *held = held_after(std::mem::replace(&mut *held, Held::Unread), &credentials, status); + } if !(200..300).contains(&status) { return Err(RpcError::Status(status)); } @@ -217,12 +235,22 @@ impl RpcClient { } } +fn held_after(held: Held, sent: &Credentials, status: u16) -> Held { + match held { + Held::Known(refused) if matches!(status, 401 | 403) && refused == *sent => Held::Unread, + kept => kept, + } +} + fn rpc_failure(settled: &Result) -> Option { settled.as_ref().err().map(RpcError::failure_mode) } #[cfg(test)] mod tests { + use std::path::Path; + use std::sync::atomic::AtomicUsize; + use std::sync::Arc; use std::time::Duration; use serde_json::json; @@ -240,51 +268,190 @@ mod tests { async fn validator_answering( answer: fn(&Value) -> Value, ) -> (String, JoinHandle<(Value, bool)>) { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind"); - let url = format!("http://{}", listener.local_addr().expect("bound")); + let (listener, url) = loopback().await; let exchange = tokio::spawn(async move { let (mut stream, _) = listener.accept().await.expect("accept"); - let mut request = Vec::new(); - let body = loop { - let mut chunk = [0u8; 4096]; - let n = stream.read(&mut chunk).await.expect("read"); - request.extend_from_slice(&chunk[..n]); - let text = String::from_utf8_lossy(&request); - let Some((head, body)) = text.split_once("\r\n\r\n") else { continue }; - let length: usize = head - .lines() - .find_map(|line| { - line.to_ascii_lowercase().strip_prefix("content-length: ")?.parse().ok() - }) - .expect("content-length"); - if body.len() == length { - break serde_json::from_str::(body).expect("json body"); - } - }; - let reply = answer(&body).to_string(); - let response = format!( - "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{reply}", - reply.len() - ); - stream.write_all(response.as_bytes()).await.expect("write"); + let (_, body) = request(&mut stream).await; + let reply = answered(&answer(&body)); + stream + .write_all(reply.as_bytes()) + .await + .expect("the client waits for its reply before it closes"); let resent = tokio::time::timeout(RESEND_WINDOW, listener.accept()).await.is_ok(); (body, resent) }); (url, exchange) } + /// - binds a loopback listener on a free port + async fn loopback() -> (tokio::net::TcpListener, String) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("loopback has a free ephemeral port"); + let address = listener.local_addr().expect("a bound listener has a local address"); + (listener, format!("http://{address}")) + } + + fn answered(reply: &Value) -> String { + let reply = reply.to_string(); + format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{reply}", + reply.len() + ) + } + + /// - reads `stream` until one request's head and whole body arrived + async fn request(stream: &mut tokio::net::TcpStream) -> (String, Value) { + let mut request = Vec::new(); + loop { + let mut chunk = [0u8; 4096]; + let n = stream + .read(&mut chunk) + .await + .expect("the client holds its connection open until it is answered"); + request.extend_from_slice(&chunk[..n]); + let text = String::from_utf8_lossy(&request); + let Some((head, body)) = text.split_once("\r\n\r\n") else { continue }; + let length: usize = head + .lines() + .find_map(|line| { + line.to_ascii_lowercase().strip_prefix("content-length: ")?.parse().ok() + }) + .expect("reqwest announces a sized body's content-length"); + if body.len() == length { + let body = serde_json::from_str(body).expect("the client sends a JSON body"); + return (head.to_owned(), body); + } + } + } + fn client(name: &str, url: String) -> RpcClient { + client_with(name, url, Auth::None) + } + + fn client_with(name: &str, url: String, auth: Auth) -> RpcClient { let limits = LinkLimits::default(); let config = RpcClientConfig { url, name: name.to_owned(), - auth: None, + auth, timeouts: Timeouts::default(), limits, }; RpcClient::new(config) } + #[test] + fn held_credentials_are_dropped_only_when_the_validator_refuses_them() { + let token = |token: &str| Credentials::basic("__cookie__", token); + let known = |token: &str| Held::Known(Credentials::basic("__cookie__", token)); + let bare = Held::Known(Credentials::NoAuth); + #[rustfmt::skip] + let cases = [ + ("answered", known("new"), token("new"), 200, known("new")), + ("server error", known("new"), token("new"), 503, known("new")), + ("refused", known("old"), token("old"), 401, Held::Unread), + ("forbidden", known("old"), token("old"), 403, Held::Unread), + ("no auth refused", bare, Credentials::NoAuth, 401, Held::Unread), + ("refused, already read again", known("new"), token("old"), 401, known("new")), + ("refused, already dropped", Held::Unread, token("old"), 401, Held::Unread), + ]; + for (case, held, sent, status, expected) in cases { + assert_eq!(held_after(held, &sent, status), expected, "{case}"); + } + } + + fn cookie_dir() -> tempfile::TempDir { + tempfile::tempdir().expect("the OS temp directory is writable") + } + + /// - writes `contents` to `cookie` + fn start(cookie: &Path, contents: &str) { + std::fs::write(cookie, contents).expect("the temp directory is writable"); + } + + async fn ping(client: &RpcClient) -> Result { + client.call("ping", vec![]).await.map_err(|error| error.to_string()) + } + + fn refused() -> Result { + Err(RpcError::Status(401).to_string()) + } + + fn basic(token: &str) -> String { + let request = reqwest::Client::new() + .get("http://validator") + .basic_auth("__cookie__", Some(token)) + .build() + .expect("a literal URL with a basic-auth header builds a request"); + let header = request.headers().get("authorization").and_then(|value| value.to_str().ok()); + header.expect("basic_auth sets an ASCII authorization header").to_owned() + } + + /// - spawns a loopback listener that answers every connection until the runtime stops + async fn validator_demanding( + tokens: [Option<&'static str>; 2], + starts: Arc, + ) -> String { + let (listener, url) = loopback().await; + tokio::spawn(async move { + while let Ok((mut stream, _)) = listener.accept().await { + let (head, body) = request(&mut stream).await; + let sent = head.lines().find_map(|line| { + let (name, value) = line.split_once(": ")?; + name.eq_ignore_ascii_case("authorization").then(|| value.to_owned()) + }); + let demanded = tokens[starts.load(Ordering::Relaxed)].map(basic); + let response = if demanded.is_none() || demanded == sent { + let sent = sent.as_deref().unwrap_or("none"); + answered(&json!({ "id": body["id"], "result": sent })) + } else { + "HTTP/1.1 401 Unauthorized\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + .to_owned() + }; + if stream.write_all(response.as_bytes()).await.is_err() { + break; + } + } + }); + url + } + + #[tokio::test] + async fn a_cookie_is_read_once_then_again_only_after_the_validator_refuses_it() { + let dir = cookie_dir(); + let cookie = dir.path().join("cookie"); + let starts = Arc::new(AtomicUsize::new(0)); + let url = validator_demanding([Some("first"), Some("second")], starts.clone()).await; + let client = client_with("restarting", url, Auth::Cookie(cookie.clone())); + + start(&cookie, "__cookie__:first"); + assert_eq!(ping(&client).await, Ok(basic("first")), "first request: cookie read"); + std::fs::remove_file(&cookie).expect("the cookie written above is still there"); + assert_eq!(ping(&client).await, Ok(basic("first")), "cookie file gone: not read again"); + + starts.store(1, Ordering::Relaxed); + start(&cookie, "__cookie__:second"); + assert_eq!(ping(&client).await, refused(), "restart: old cookie"); + assert_eq!(ping(&client).await, Ok(basic("second")), "after the refusal: read again"); + } + + #[tokio::test] + async fn a_default_cookie_is_sent_only_after_a_bare_request_is_refused() { + let dir = cookie_dir(); + let candidates = ["zebra.cookie", "zakura.cookie"].map(|name| dir.path().join(name)); + start(&candidates[1], "__cookie__:local"); + let starts = Arc::new(AtomicUsize::new(0)); + let url = validator_demanding([None, Some("local")], starts.clone()).await; + let client = client_with("challenging", url, Auth::DefaultCookie(candidates.to_vec())); + + assert_eq!(ping(&client).await, Ok("none".to_owned()), "nothing demanded: no cookie"); + + starts.store(1, Ordering::Relaxed); + assert_eq!(ping(&client).await, refused(), "demanded: the bare request"); + assert_eq!(ping(&client).await, Ok(basic("local")), "after the refusal: the cookie"); + } + /// Batch replies reversed by the validator: outcomes still in call order, `-1` busy = that /// item's own refusal (never re-sent here: the balancer's retry) #[tokio::test] diff --git a/packages/zingod-source/src/rpc/error.rs b/packages/zingod-source/src/rpc/error.rs index fa81a905a..7f787e0e1 100644 --- a/packages/zingod-source/src/rpc/error.rs +++ b/packages/zingod-source/src/rpc/error.rs @@ -15,6 +15,8 @@ pub enum RpcError { ResponseBodyTooLarge { max: usize }, #[error("batch reply does not answer each call exactly once")] BatchMismatch, + #[error("credentials: {0}")] + Credentials(#[from] crate::EndpointError), } impl RpcError { @@ -24,7 +26,7 @@ impl RpcError { match self { RpcError::Http(inner) if inner.is_timeout() => FailureMode::Timeout, RpcError::Http(_) => FailureMode::Connection, - RpcError::Status(401 | 403) => FailureMode::Auth, + RpcError::Status(401 | 403) | RpcError::Credentials(_) => FailureMode::Auth, RpcError::Status(code) => FailureMode::HttpStatus(*code), RpcError::Rpc { code, .. } => FailureMode::RpcError(*code), // oversized body = `Parse`, not retryable (same request → same oversized reply) diff --git a/packages/zingod-source/usage.md b/packages/zingod-source/usage.md index e0d18c90b..f62f21974 100644 --- a/packages/zingod-source/usage.md +++ b/packages/zingod-source/usage.md @@ -30,10 +30,18 @@ let grpc = ZebraIndexerGrpcClient::new(legacy, indexer_address, tls.as_ref(), Ti ``` - `new` never contacts the validator: an unreachable validator is the caller's retry, never a - construction failure. `EndpointError` = a bad address or TLS config, an unreadable cookie or - TLS file. + construction failure. `EndpointError` = a bad address or TLS config, an unreadable TLS file, or + a cookie that exists and cannot be read (a cookie not yet written is not a failure). - `Auth`: `Cookie` sends basic auth as `__cookie__` with the file's token, `Basic` the pair, - `None` no `Authorization` header. + `None` no `Authorization` header. `DefaultCookie(candidates)` sends no header until the + validator refuses a request without one, then the most recently written candidate file present + (none = no auth, same age = the earlier candidate). +- A cookie is read on the first request, off the executor, and held. A `401` or `403` drops the + held credentials, so the next request reads the file again (a validator rewrites its cookie on + every start). An unreadable or empty cookie fails that request as `FailureMode::Auth` and is + read again on the next one. +- `Reach` = where an address leads: `of_validator` reads the host `validator_url` parses, + `of_ip` an IP. `Loopback` says where the socket connects, never who answers. - One attempt per call. Which validator, how many requests in flight, retries and hedges are the balancer's (`zingod-traffic`): this crate never re-sends, and a batch item's refusal (zebrad's `-1` work queue full included) is that item's own outcome. diff --git a/packages/zingod/src/config/mod.rs b/packages/zingod/src/config/mod.rs index c2fa6a11e..35eb3ff30 100644 --- a/packages/zingod/src/config/mod.rs +++ b/packages/zingod/src/config/mod.rs @@ -25,7 +25,9 @@ use zingod_index_transparent_address::DEFAULT_MAX_ADDRESS_ROWS; use zingod_persistence::{manifest_fits, LsmConfig}; use zingod_primitives::protocol::MAX_BLOCK_REORG_HEIGHT; use zingod_primitives::serde_secs; -use zingod_source::{Auth, Endpoint, LinkLimits, ResponseSizeCap, Timeouts, ZebraIndexerTls}; +use zingod_source::{ + Auth, Endpoint, LinkLimits, Reach, ResponseSizeCap, Timeouts, ZebraIndexerTls, +}; use zingod_sync::IndexKind; use zingod_traffic::{Limits, TrustedValidatorId}; @@ -141,7 +143,10 @@ impl TrustedValidatorTable { priority, } = self; let auth = match (cookie_path, user, password) { - (None, None, None) => Auth::None, + (None, None, None) => match Reach::of_validator(&address)? { + Reach::Loopback => Auth::DefaultCookie(paths::default_validator_cookies()), + Reach::Private | Reach::Public | Reach::Named => Auth::None, + }, (Some(path), None, None) => Auth::Cookie(path), (None, Some(user), Some(password)) => Auth::Basic { user, password }, (Some(_), _, _) => { @@ -680,7 +685,7 @@ impl DaemonConfig { }; // - Public bind publishes chain tip, sync progress, request volumes, RSS // - Warn, not reject (read-only telemetry, containers bind 0.0.0.0 by norm) - if !is_private_listen_addr(&endpoint) { + if Reach::of_ip(endpoint.ip()) == Reach::Public { tracing::warn!( %endpoint, "metrics.listen_address binds a non-private address; /metrics is \ @@ -702,13 +707,6 @@ fn resolved_path( path.or_else(default).ok_or_else(missing) } -fn is_private_listen_addr(addr: &SocketAddr) -> bool { - match addr.ip() { - std::net::IpAddr::V4(ipv4) => ipv4.is_private() || ipv4.is_loopback(), - std::net::IpAddr::V6(ipv6) => ipv6.is_unique_local() || ipv6.is_loopback(), - } -} - /// Defaults as TOML, behind [`GENERATED_CONFIG_HEADER`] (`Err` = defaults that don't resolve: a /// path with no home dir default) fn generate_default_config() -> Result { @@ -1241,8 +1239,32 @@ key_path = "/etc/zingod/indexer-client-key.pem" (Limits::new(8).expect("8 ≥ MIN_CONNECTIONS"), Some(indexer), 1), "eu's balancer limits, indexer, tier" ); - let local = local.resolve().expect("no credentials"); - assert_eq!((local.endpoint.auth, local.indexer), (Auth::None, None), "local: neither"); + let local = local.resolve().expect("an entry with only an address resolves"); + let defaults = Auth::DefaultCookie(paths::default_validator_cookies()); + assert_eq!( + (local.endpoint.auth, local.indexer), + (defaults.clone(), None), + "local: default cookies" + ); + #[rustfmt::skip] + let reaches = [ + ("localhost:8232", true), + ("[::1]:8232", true), + ("127.1:8232", true), + ("10.0.0.5:8232", false), + ("8.8.8.8:8232", false), + ("zebra-eu:8232", false), + ]; + for (address, loopback) in reaches { + let bare = TrustedValidatorTable { + jsonrpc_address: address.to_owned(), + ..TrustedValidatorTable::default() + }; + let auth = + bare.resolve().expect("an entry with only an address resolves").endpoint.auth; + let expected = if loopback { defaults.clone() } else { Auth::None }; + assert_eq!(auth, expected, "{address}: a default cookie on loopback alone"); + } let cookie = |user: Option<&str>, password: Option<&str>| TrustedValidatorTable { cookie_path: Some("/zebra/.cookie".into()), diff --git a/packages/zingod/src/config/paths.rs b/packages/zingod/src/config/paths.rs index 9bb134e9a..a91b0d630 100644 --- a/packages/zingod/src/config/paths.rs +++ b/packages/zingod/src/config/paths.rs @@ -23,6 +23,11 @@ pub(crate) fn default_state() -> Option { Some(dirs::data_dir()?.join("zingod/state")) } +pub(crate) fn default_validator_cookies() -> Vec { + let cookies = |cache: PathBuf| ["zebra", "zakura"].map(|dir| cache.join(dir).join(".cookie")); + dirs::cache_dir().map(cookies).map(Vec::from).unwrap_or_default() +} + /// `/zingod/peers`: zebra-network's peer address cache pub(crate) fn default_peer_cache() -> Option { Some(dirs::cache_dir()?.join("zingod/peers"))