diff --git a/keetanetwork-client/src/client.rs b/keetanetwork-client/src/client.rs index 7884349..91dd368 100644 --- a/keetanetwork-client/src/client.rs +++ b/keetanetwork-client/src/client.rs @@ -30,7 +30,7 @@ use crate::model::{ AccountState, Acl, Certificate, ChainPage, ChainQuery, HistoryEntry, HistoryPage, HistoryQuery, LedgerChecksum, Representative, TokenBalance, TransmitOptions, }; -use crate::rep::{RepBook, RepPart, RepRecord, RepRef}; +use crate::rep::{is_safe_advertised_url, RepBook, RepPart, RepRecord, RepRef}; use crate::runtime::{Runtime, TaskHandle}; use crate::sync::{Mutex, RwLock}; use crate::transport::{LedgerSide, NodeTransport, TransportFactory}; @@ -43,6 +43,8 @@ use { crate::generated::Client as Transport, crate::model::RepStatus, crate::runtime::TokioRuntime, std::sync::OnceLock, }; +const DEFAULT_REP_REFRESH_TIMEOUT: Duration = Duration::from_secs(30); + /// Bookkeeping for the background representative-refresh task. #[derive(Debug, Default)] struct RefreshState { @@ -371,8 +373,8 @@ impl KeetaClient { refresh.handle = Some(handle); } - /// Refresh known representatives' voting weights from - /// `GET /representatives`, matching by account. + /// Refresh known representatives' voting weights from a strict majority of + /// configured peers, matching by account. async fn update_reps(&self) -> Result<(), ClientError> { let signature = self.inner.reps.sorted_keys().join(","); let ttl = Duration::from_millis(self.inner.config.reps_cache_ttl_ms); @@ -381,7 +383,9 @@ impl KeetaClient { Some(cached) => cached, None => { let entries = self.fetch_rep_entries().await?; - store_representatives(&self.inner.runtime, &signature, &entries); + if !entries.is_empty() { + store_representatives(&self.inner.runtime, &signature, &entries); + } entries } }; @@ -404,14 +408,39 @@ impl KeetaClient { Ok(()) } - /// Fetch the representative set as `(key, weight, api_url)` entries. + /// Fetch representative lists from every configured peer and retain only + /// values on which a strict majority of at least two peers agree. async fn fetch_rep_entries(&self) -> Result, ClientError> { - let representatives = self.representatives().await?; - let entries = representatives - .into_iter() - .map(|rep| (rep.account.to_string(), rep.weight.as_bigint().clone(), rep.api_url)) - .collect(); - Ok(entries) + let picks = self.snapshot_picks(); + if picks.is_empty() { + return Err(ClientError::NoRepresentatives); + } + let peer_count = picks.len(); + + let mut requests = FuturesUnordered::new(); + for pick in picks { + let client = self.clone(); + requests.push(async move { + client + .run_rep_refresh_call(pick.transport.representatives()) + .await + }); + } + + let mut responses = Vec::new(); + while let Some(result) = requests.next().await { + let Some(Ok(representatives)) = result else { + continue; + }; + responses.push( + representatives + .into_iter() + .map(|rep| (rep.account.to_string(), rep.weight.as_bigint().clone(), rep.api_url)) + .collect(), + ); + } + + Ok(consensus_rep_entries(responses, peer_count)) } /// Apply fetched representative entries to the shared state: refresh the @@ -432,7 +461,7 @@ impl KeetaClient { let Some(api) = api_url else { continue; }; - if !self.inner.reps.contains(key) { + if !self.inner.reps.contains(key) && is_safe_advertised_url(api) { self.inner .reps .add(RepRecord::new(key.clone(), api.clone(), weight.clone())); @@ -505,8 +534,29 @@ impl KeetaClient { return Some(future.await); } - let duration = Duration::from_millis(timeout_ms); - let timer = self.inner.runtime.sleep(duration); + self.run_call_with_timeout(future, Duration::from_millis(timeout_ms)) + .await + } + + /// Await a representative-refresh call with a finite deadline, even when + /// general requests have no configured timeout. + async fn run_rep_refresh_call( + &self, + future: impl Future>, + ) -> Option> { + let timeout = match self.inner.config.request_timeout_ms { + 0 => DEFAULT_REP_REFRESH_TIMEOUT, + timeout_ms => Duration::from_millis(timeout_ms), + }; + self.run_call_with_timeout(future, timeout).await + } + + async fn run_call_with_timeout( + &self, + future: impl Future>, + timeout: Duration, + ) -> Option> { + let timer = self.inner.runtime.sleep(timeout); pin_mut!(future); pin_mut!(timer); @@ -1925,6 +1975,53 @@ impl KeetaClient { /// advertised API URL (when the node provided one). type RepEntry = (String, BigInt, Option); +/// Resolve peer-advertised representative data by strict majority. A response +/// contributes at most one weight and URL per representative key. +fn consensus_rep_entries(responses: Vec>, peer_count: usize) -> Vec { + if peer_count < 2 { + return Vec::new(); + } + + let mut weight_votes: BTreeMap> = BTreeMap::new(); + let mut url_votes: BTreeMap> = BTreeMap::new(); + for response in responses { + let unique: BTreeMap)> = response + .into_iter() + .map(|(key, weight, api_url)| (key, (weight, api_url))) + .collect(); + for (key, (weight, api_url)) in unique { + *weight_votes + .entry(key.clone()) + .or_default() + .entry(weight) + .or_default() += 1; + if let Some(api_url) = api_url.filter(|url| is_safe_advertised_url(url)) { + *url_votes + .entry(key) + .or_default() + .entry(api_url) + .or_default() += 1; + } + } + } + + let majority = peer_count / 2; + weight_votes + .into_iter() + .filter_map(|(key, votes)| { + let weight = votes + .into_iter() + .find_map(|(weight, count)| (count > majority).then_some(weight))?; + let api_url = url_votes.remove(&key).and_then(|votes| { + votes + .into_iter() + .find_map(|(url, count)| (count > majority).then_some(url)) + }); + Some((key, weight, api_url)) + }) + .collect() +} + /// Process-shared representative cache, keyed by rep-set signature, so /// concurrent clients/clones over the same reps refresh from one fetch. The /// timestamp is the runtime's monotonic millisecond tick at storage time. @@ -2106,6 +2203,7 @@ mod tests { use keetanetwork_block::BlockHash; use keetanetwork_vote::{Fee, Fees, VoteBuilder}; + use crate::runtime::{BoxFuture, TokioRuntime}; use crate::transport::GeneratedTransport; const ISSUER_SEED: u8 = 0xA1; @@ -2117,6 +2215,58 @@ mod tests { KeetaClient::new("http://localhost").with_network(BigInt::from(TEST_NETWORK)) } + #[derive(Debug, Default)] + struct RecordingFactory { + urls: Mutex>, + } + + impl TransportFactory for RecordingFactory { + fn create(&self, url: &str) -> Arc { + self.urls.lock().push(url.to_owned()); + Arc::new(GeneratedTransport::new(url, reqwest::Client::new())) + } + } + + #[derive(Debug)] + struct ImmediateRuntime; + + #[derive(Debug)] + struct NoopTask; + + impl TaskHandle for NoopTask { + fn abort(&self) {} + } + + #[async_trait::async_trait] + impl Runtime for ImmediateRuntime { + async fn sleep(&self, _duration: Duration) {} + + fn spawn(&self, _future: BoxFuture) -> Box { + Box::new(NoopTask) + } + + fn now_millis(&self) -> u64 { + 0 + } + + fn unix_millis(&self) -> i64 { + 0 + } + } + + fn multi_rep_client(factory: Arc) -> KeetaClient { + KeetaClient::with_parts( + [ + RepPart { key: "a".to_owned(), url: "http://127.0.0.1:8001".to_owned(), weight: 60.into() }, + RepPart { key: "b".to_owned(), url: "http://127.0.0.1:8002".to_owned(), weight: 40.into() }, + ], + factory, + Arc::new(TokioRuntime), + ClientConfig::default(), + false, + ) + } + fn fee(amount: u64, pay_to: Option, token: Option) -> Fee { Fee { amount: Amount::from(amount), pay_to, token } } @@ -2401,4 +2551,68 @@ mod tests { assert_eq!(contacts[0].key, rep_a.to_string()); Ok(()) } + + #[test] + fn representative_weights_require_multi_peer_consensus() { + let honest = vec![ + ("a".to_owned(), 60.into(), Some("https://a.example.com".to_owned())), + ("b".to_owned(), 40.into(), Some("https://b.example.com".to_owned())), + ]; + let malicious = vec![ + ("a".to_owned(), 1.into(), Some("https://a.example.com".to_owned())), + ("b".to_owned(), 999_999.into(), Some("https://b.example.com".to_owned())), + ]; + + assert!(consensus_rep_entries(vec![malicious.clone()], 3).is_empty()); + let consensus = consensus_rep_entries(vec![honest.clone(), honest, malicious], 3); + let client = multi_rep_client(Arc::new(RecordingFactory::default())); + client.apply_reps(&consensus, false); + + let (snapshot, total) = client.inner.reps.snapshot_with_total(); + assert_eq!(total, BigInt::from(100)); + assert_eq!(client.inner.reps.pick().map(|pick| pick.key), Some("a".to_owned())); + let malicious_rep_weight = snapshot + .iter() + .find(|rep| rep.key == "b") + .map(|rep| rep.weight.clone()) + .expect("rep b"); + assert_eq!(malicious_rep_weight, BigInt::from(40)); + assert!(!meets_quorum(&malicious_rep_weight, &total, 0.5)); + } + + #[test] + fn representative_consensus_does_not_shrink_when_peers_timeout() { + let colluding = vec![("a".to_owned(), 999_999.into(), None)]; + assert!(consensus_rep_entries(vec![colluding.clone(), colluding], 4).is_empty()); + } + + #[test] + fn representative_refresh_bounds_calls_when_request_timeout_is_unset() { + let factory = Arc::new(RecordingFactory::default()); + let client = KeetaClient::with_parts( + [RepPart { key: "a".to_owned(), url: "http://127.0.0.1:8001".to_owned(), weight: 1.into() }], + factory, + Arc::new(ImmediateRuntime), + ClientConfig::default(), + false, + ); + let pending = core::future::pending::>(); + + assert!(matches!(resolve(client.run_rep_refresh_call(pending)), Some(None))); + } + + #[test] + fn discovery_does_not_create_transport_for_unsafe_advertised_url() { + let factory = Arc::new(RecordingFactory::default()); + let client = multi_rep_client(Arc::clone(&factory)); + assert_eq!(factory.urls.lock().len(), 2); + + client.apply_reps( + &[("attacker".to_owned(), BigInt::from(1), Some("http://169.254.169.254/latest/meta-data".to_owned()))], + true, + ); + + assert_eq!(factory.urls.lock().len(), 2); + assert!(!client.inner.reps.contains("attacker")); + } } diff --git a/keetanetwork-client/src/rep.rs b/keetanetwork-client/src/rep.rs index b1e34a9..e3f99f8 100644 --- a/keetanetwork-client/src/rep.rs +++ b/keetanetwork-client/src/rep.rs @@ -6,6 +6,8 @@ use alloc::collections::BTreeMap; use alloc::string::String; use alloc::vec::Vec; +use core::net::{IpAddr, Ipv4Addr}; +use core::str::FromStr; use num_bigint::BigInt; @@ -31,6 +33,113 @@ impl RepRecord { } } +/// Whether an untrusted, peer-advertised API URL is safe to use for +/// representative discovery. Configured URLs do not pass through this check. +pub(crate) fn is_safe_advertised_url(url: &str) -> bool { + let Some(rest) = url + .strip_prefix("http://") + .or_else(|| url.strip_prefix("https://")) + else { + return false; + }; + let authority = rest.split(['/', '?', '#']).next().unwrap_or_default(); + if authority.is_empty() || authority.contains('@') { + return false; + } + + let host = if let Some(bracketed) = authority.strip_prefix('[') { + let Some((host, suffix)) = bracketed.split_once(']') else { + return false; + }; + if !valid_port_suffix(suffix) { + return false; + } + host + } else { + let mut parts = authority.split(':'); + let host = parts.next().unwrap_or_default(); + if let Some(port) = parts.next() { + if parts.next().is_some() || !valid_port(port) { + return false; + } + } + host + }; + + if host.is_empty() + || !host + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'-' | b':')) + { + return false; + } + + let normalized = host.trim_end_matches('.').to_ascii_lowercase(); + if normalized == "localhost" + || normalized.ends_with(".localhost") + || normalized == "metadata.google.internal" + || normalized.ends_with(".metadata.google.internal") + { + return false; + } + + match IpAddr::from_str(&normalized) { + Ok(ip) => is_public_ip(ip), + // Reject non-canonical numeric hosts such as `2130706433` and + // `0x7f000001`, which URL clients may normalize to loopback. + Err(_) + if normalized + .bytes() + .all(|byte| byte.is_ascii_digit() || byte == b'.') => + { + false + } + Err(_) if normalized.starts_with("0x") => false, + Err(_) => true, + } +} + +fn valid_port_suffix(suffix: &str) -> bool { + suffix.is_empty() || suffix.strip_prefix(':').is_some_and(valid_port) +} + +fn valid_port(port: &str) -> bool { + !port.is_empty() && port.parse::().is_ok() +} + +fn is_public_ip(ip: IpAddr) -> bool { + match ip { + IpAddr::V4(ip) => { + let octets = ip.octets(); + !(octets[0] == 0 + || is_shared_address(ip) + || ip == Ipv4Addr::new(168, 63, 129, 16) + || ip.is_private() + || ip.is_loopback() + || ip.is_link_local() + || ip.is_unspecified() + || ip.is_broadcast() + || ip.is_multicast()) + } + IpAddr::V6(ip) => { + if let Some(embedded) = ip.to_ipv4() { + return is_public_ip(IpAddr::V4(embedded)); + } + let first = ip.segments()[0]; + !(ip.is_loopback() + || ip.is_unspecified() + || ip.is_multicast() + || first & 0xfe00 == 0xfc00 + || first & 0xffc0 == 0xfe80) + } + } +} + +fn is_shared_address(ip: Ipv4Addr) -> bool { + let octets = ip.octets(); + octets[0] == 100 && octets[1] & 0xc0 == 0x40 +} + /// A cloned selection target: the representative's key and weight. The live /// transport is bound by `key` in the std client. #[derive(Clone, Debug)] @@ -372,4 +481,45 @@ mod tests { assert!(snapshot.iter().any(|rep| rep.key == "a")); assert_eq!(total, BigInt::from(1)); } + + #[test] + fn advertised_urls_reject_local_and_non_http_targets() { + for url in [ + "file:///etc/passwd", + "gopher://example.com", + "http://localhost/admin", + "http://service.localhost/admin", + "http://0.1.2.3/admin", + "http://127.0.0.1/admin", + "http://2130706433/admin", + "http://0x7f000001/admin", + "http://10.0.0.1/admin", + "http://100.64.0.1/admin", + "http://100.100.100.200/latest/meta-data", + "http://100.127.255.254/admin", + "http://168.63.129.16/machine?comp=goalstate", + "http://169.254.169.254/latest/meta-data", + "http://[::1]/admin", + "http://[::127.0.0.1]/admin", + "http://[::100.100.100.200]/latest/meta-data", + "http://[fe80::1]/admin", + "http://[fc00::1]/admin", + "http://metadata.google.internal/computeMetadata/v1", + "http://user@example.com", + ] { + assert!(!is_safe_advertised_url(url), "{url}"); + } + } + + #[test] + fn advertised_urls_allow_public_http_endpoints() { + for url in [ + "https://rep.example.com/api", + "http://rep.example.com:8080/api", + "https://8.8.8.8/api", + "https://[2606:4700:4700::1111]/api", + ] { + assert!(is_safe_advertised_url(url), "{url}"); + } + } }