diff --git a/crates/capture-authority/src/lib.rs b/crates/capture-authority/src/lib.rs index c1cacabcb..ac26698ff 100644 --- a/crates/capture-authority/src/lib.rs +++ b/crates/capture-authority/src/lib.rs @@ -9,10 +9,12 @@ use std::{ mod application; mod committed; mod created; +mod node; pub use created::{ CaptureBinding, CreatedApplyResult, CreatedEvent, CreatedOperation, CreatedResult, CreatedVersion, DecideCreated, MAX_OBJECT_KEY_LENGTH, ObjectIdentity, PreparedIdentity, }; +pub use node::{CaptureNode, CaptureNodeError}; #[cfg(test)] mod application_tests; mod truncate; diff --git a/crates/capture-authority/src/node.rs b/crates/capture-authority/src/node.rs new file mode 100644 index 000000000..741bed255 --- /dev/null +++ b/crates/capture-authority/src/node.rs @@ -0,0 +1,371 @@ +//! Validated capture peer metadata and its independent CNOD v1 codec. +use serde::{Deserialize, Deserializer, Serialize, Serializer, ser::SerializeStruct}; + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct CaptureNode { + rpc_uri: String, +} + +#[derive(Debug, thiserror::Error)] +pub enum CaptureNodeError { + #[error("invalid RPC URI")] + InvalidRpcUri, + #[error("invalid RPC URI")] + InvalidRpcUriParse(#[source] url::ParseError), + #[error("RPC URI too long")] + RpcUriTooLong, + #[error("invalid node magic")] + InvalidNodeMagic, + #[error("unsupported node version")] + UnsupportedNodeVersion, + #[error("invalid node length")] + InvalidNodeLength, + #[error("truncated node record")] + TruncatedNodeRecord, + #[error("invalid node UTF-8")] + InvalidNodeUtf8(#[source] std::str::Utf8Error), + #[error("trailing node bytes")] + TrailingNodeBytes, +} +impl CaptureNodeError { + pub fn code(&self) -> &'static str { + match self { + Self::InvalidRpcUri | Self::InvalidRpcUriParse(_) => "invalid_rpc_uri", + Self::RpcUriTooLong => "rpc_uri_too_long", + Self::InvalidNodeMagic => "invalid_node_magic", + Self::UnsupportedNodeVersion => "unsupported_node_version", + Self::InvalidNodeLength => "invalid_node_length", + Self::TruncatedNodeRecord => "truncated_node_record", + Self::InvalidNodeUtf8(_) => "invalid_node_utf8", + Self::TrailingNodeBytes => "trailing_node_bytes", + } + } +} +impl CaptureNode { + pub fn new(rpc_uri: String) -> Result { + let node = Self { rpc_uri }; + node.validate()?; + Ok(node) + } + + pub fn rpc_uri(&self) -> &str { + &self.rpc_uri + } + + pub fn validate(&self) -> Result<(), CaptureNodeError> { + let uri = &self.rpc_uri; + if uri.len() > 4096 { + return Err(CaptureNodeError::RpcUriTooLong); + } + if uri.is_empty() || uri.chars().any(|c| c.is_whitespace() || c.is_control() || c == '\\') { + return Err(CaptureNodeError::InvalidRpcUri); + } + let rest = uri + .strip_prefix("http://") + .or_else(|| uri.strip_prefix("https://")) + .ok_or(CaptureNodeError::InvalidRpcUri)?; + if uri.contains('?') || uri.contains('#') { + return Err(CaptureNodeError::InvalidRpcUri); + } + let (authority, path) = rest.split_once('/').map_or((rest, ""), |(a, p)| (a, p)); + // An empty suffix after the first slash is the only allowed path. + if !path.is_empty() || authority.contains('@') { + return Err(CaptureNodeError::InvalidRpcUri); + } + let (host, port) = if authority.starts_with('[') { + let end = authority.find(']').ok_or(CaptureNodeError::InvalidRpcUri)?; + let port = authority + .get(end + 1..) + .and_then(|suffix| suffix.strip_prefix(':')) + .ok_or(CaptureNodeError::InvalidRpcUri)?; + (&authority[..=end], port) + } else { + authority.split_once(':').ok_or(CaptureNodeError::InvalidRpcUri)? + }; + if host.is_empty() || port.is_empty() || !port.bytes().all(|b| b.is_ascii_digit()) { + return Err(CaptureNodeError::InvalidRpcUri); + } + let port = port.parse::().map_err(|_| CaptureNodeError::InvalidRpcUri)?; + if port == 0 { + return Err(CaptureNodeError::InvalidRpcUri); + } + let parsed_url = url::Url::parse(uri).map_err(CaptureNodeError::InvalidRpcUriParse)?; + if parsed_url.host().is_none() { + return Err(CaptureNodeError::InvalidRpcUri); + } + Ok(()) + } + + pub fn encode(&self) -> Result, CaptureNodeError> { + self.validate()?; + let length = u32::try_from(self.rpc_uri.len()).map_err(|_| CaptureNodeError::InvalidNodeLength)?; + let mut bytes = b"CNOD\x01".to_vec(); + bytes.extend_from_slice(&length.to_be_bytes()); + bytes.extend_from_slice(self.rpc_uri.as_bytes()); + Ok(bytes) + } + + pub fn decode(bytes: &[u8]) -> Result { + if bytes.len() < 9 { + return Err(CaptureNodeError::TruncatedNodeRecord); + } + if &bytes[..4] != b"CNOD" { + return Err(CaptureNodeError::InvalidNodeMagic); + } + if bytes[4] != 1 { + return Err(CaptureNodeError::UnsupportedNodeVersion); + } + let length = u32::from_be_bytes(bytes[5..9].try_into().map_err(|_| CaptureNodeError::InvalidNodeLength)?); + if !(1..=4096).contains(&length) { + return Err(CaptureNodeError::InvalidNodeLength); + } + let length = usize::try_from(length).map_err(|_| CaptureNodeError::InvalidNodeLength)?; + let end = 9_usize.checked_add(length).ok_or(CaptureNodeError::InvalidNodeLength)?; + if bytes.len() < end { + return Err(CaptureNodeError::TruncatedNodeRecord); + } + if bytes.len() != end { + return Err(CaptureNodeError::TrailingNodeBytes); + } + let uri = std::str::from_utf8(&bytes[9..end]).map_err(CaptureNodeError::InvalidNodeUtf8)?; + Self::new(uri.to_owned()) + } +} +impl Serialize for CaptureNode { + fn serialize(&self, serializer: S) -> Result { + self.validate().map_err(serde::ser::Error::custom)?; + let mut state = serializer.serialize_struct("CaptureNode", 1)?; + state.serialize_field("rpc_uri", &self.rpc_uri)?; + state.end() + } +} +impl<'de> Deserialize<'de> for CaptureNode { + fn deserialize>(deserializer: D) -> Result { + #[derive(Deserialize)] + #[serde(deny_unknown_fields)] + struct Fields { + rpc_uri: String, + } + let fields = Fields::deserialize(deserializer).map_err(|_| serde::de::Error::custom("invalid capture node fields"))?; + Self::new(fields.rpc_uri).map_err(serde::de::Error::custom) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::error::Error; + + fn record(payload: &[u8]) -> Vec { + let mut bytes = b"CNOD\x01".to_vec(); + bytes.extend_from_slice(&u32::try_from(payload.len()).expect("test payload length").to_be_bytes()); + bytes.extend_from_slice(payload); + bytes + } + fn reject_uri(uri: &str) { + let result = CaptureNode::new(uri.to_owned()); + assert!(result.is_err(), "URI should reject: {uri:?}"); + assert_eq!(result.expect_err("rejected URI").code(), "invalid_rpc_uri"); + } + fn reject_record(bytes: &[u8], code: &str) { + let result = CaptureNode::decode(bytes); + assert!(result.is_err(), "record should reject with {code}"); + assert_eq!(result.expect_err("rejected record").code(), code); + } + #[test] + fn actual_openraft_node_bound() { + fn require_node() {} + require_node::(); + assert_eq!(CaptureNode::default().rpc_uri(), ""); + } + #[test] + fn default_and_empty_rejected_at_every_boundary() { + let node = CaptureNode::default(); + assert_eq!(node.rpc_uri(), ""); + reject_uri(""); + assert_eq!(node.validate().expect_err("empty validation").code(), "invalid_rpc_uri"); + assert!(serde_json::to_string(&node).is_err()); + assert!(serde_json::from_str::(r#"{"rpc_uri":""}"#).is_err()); + assert_eq!(node.encode().expect_err("empty encoding").code(), "invalid_rpc_uri"); + reject_record(&record(b""), "invalid_node_length"); + } + #[test] + fn accepted_uris_preserve_exact_bytes() { + for uri in [ + "http://node:80", + "https://node:443", + "http://node:1", + "https://node:65535/", + "http://127.0.0.1:9000", + "https://[::1]:443/", + "http://[2001:db8::1]:1", + "http://NoDe:00080/", + ] { + let result = CaptureNode::new(uri.to_owned()); + assert!(result.is_ok(), "valid URI construction: {uri}"); + let node = result.expect("valid URI"); + assert_eq!(node.rpc_uri(), uri, "preserve explicit default port and raw URI"); + assert_eq!(node.encode().expect("encode URI"), record(uri.as_bytes())); + assert_eq!(CaptureNode::decode(&record(uri.as_bytes())).expect("decode URI"), node); + } + } + #[test] + fn golden_cnod_bytes() { + let node = CaptureNode { + rpc_uri: "http://node:80".to_owned(), + }; + let golden = b"CNOD\x01\x00\x00\x00\x0ehttp://node:80"; + assert_eq!(node.encode().ok().as_deref(), Some(golden.as_slice()), "golden CNOD bytes"); + } + #[test] + fn valid_cnod_decode() { + let golden = b"CNOD\x01\x00\x00\x00\x0ehttp://node:80"; + let decoded = CaptureNode::decode(golden); + assert!(decoded.is_ok(), "golden decode succeeds"); + assert_eq!(decoded.expect("golden decoded node").rpc_uri(), "http://node:80"); + } + #[test] + fn serde_shape_roundtrip_and_strict_fields() { + let node = CaptureNode::new("https://node:443/".to_owned()).expect("serde node"); + let json = r#"{"rpc_uri":"https://node:443/"}"#; + assert_eq!(serde_json::to_string(&node).expect("serialize node"), json); + assert_eq!(serde_json::from_str::(json).expect("deserialize node"), node); + for json in [ + r#"{}"#, + r#"{"rpc_uri":1}"#, + r#"{"rpc_uri":null}"#, + r#"{"rpc_uri":"http://node:1","extra":0}"#, + r#"{"rpc_uri":"http://node:1","rpc_uri":"http://node:2"}"#, + ] { + assert!(serde_json::from_str::(json).is_err(), "strict fields: {json}"); + } + } + #[test] + fn credentials_and_raw_uri_shapes_reject() { + for uri in [ + "http://user@node:1", + "http://user:password@node:1", + "http://@node:1", + "HTTP://node:1", + "Https://node:1", + "http://node", + "http://node:", + "http://node:0", + "http://node:-1", + "http://node:+1", + "http://node:65536", + "http://node:999999999999999999999", + "http://[::1:1", + "http://[::1]extra:1", + "http://[invalid]:1", + "http://::1:1", + "http://:1", + "/disk/path", + "file:///disk/path", + "http://node:1/./", + "http://node:1/../", + "http://node:1/%2e/", + "http://node:1/%2E%2e/", + "http://node:1/path", + "http://node:1\\", + " http://node:1", + "http://node:1\n", + "http://no\tde:1", + "http://node:1\u{00a0}", + "http://node:1\u{0085}", + "http://node:1\0", + ] { + reject_uri(uri); + } + } + #[test] + fn query_and_fragment_markers_reject() { + for uri in [ + "http://node:1?", + "http://node:1?x=y", + "http://node:1/?", + "http://node:1#", + "http://node:1#fragment", + "http://node:1/#", + ] { + reject_uri(uri); + } + } + #[test] + fn utf8_byte_length_boundaries() { + for length in [4095, 4096] { + let prefix = "http://node:"; + let uri = format!("{prefix}{}1", "0".repeat(length - prefix.len() - 1)); + assert_eq!(uri.len(), length); + assert!(url::Url::parse(&uri).is_ok(), "boundary fixture accepted by actual URL parser"); + let node = CaptureNode::new(uri.clone()).expect("boundary node"); + assert_eq!(node.rpc_uri(), uri); + assert_eq!( + CaptureNode::decode(&node.encode().expect("boundary encoding")).expect("boundary decoding"), + node + ); + } + let uri = format!("http://node:{}1", "0".repeat(4097 - "http://node:".len() - 1)); + assert_eq!(CaptureNode::new(uri).expect_err("oversize URI").code(), "rpc_uri_too_long"); + let unicode = format!("http://{}:1", "é".repeat(2044)); + assert!(unicode.chars().count() < 4096); + assert!(unicode.len() > 4096); + assert_eq!(CaptureNode::new(unicode).expect_err("UTF-8 byte limit").code(), "rpc_uri_too_long"); + } + #[test] + fn cnod_headers_and_lengths_reject_exact_codes() { + let valid = record(b"http://node:80"); + for end in 0..9 { + reject_record(&valid[..end], "truncated_node_record"); + } + let mut bad = valid.clone(); + bad[0] = b'X'; + reject_record(&bad, "invalid_node_magic"); + bad = valid.clone(); + bad[4] = 2; + reject_record(&bad, "unsupported_node_version"); + for length in [0_u32, 4097, u32::MAX] { + bad = b"CNOD\x01".to_vec(); + bad.extend_from_slice(&length.to_be_bytes()); + reject_record(&bad, "invalid_node_length"); + } + for end in 9..valid.len() { + reject_record(&valid[..end], "truncated_node_record"); + } + reject_record(&record(&[0xff]), "invalid_node_utf8"); + reject_record(&record(&[0xc3, 0x28]), "invalid_node_utf8"); + reject_record(&record(b"invalid URI"), "invalid_rpc_uri"); + } + #[test] + fn trailing_bytes_reject() { + for tail in [&b"x"[..], &b"xyz"[..]] { + let mut bytes = record(b"http://node:80"); + bytes.extend_from_slice(tail); + reject_record(&bytes, "trailing_node_bytes"); + } + } + #[test] + fn errors_and_sources_redact_input() { + let sentinel = "http://SECRET_USER:SECRET_PASSWORD@node:1"; + let error = CaptureNode::new(sentinel.to_owned()).expect_err("credentials reject"); + let mut current: &dyn Error = &error; + loop { + let rendered = format!("{current} {current:?}"); + assert!(!rendered.contains("SECRET")); + match current.source() { + Some(source) => current = source, + None => break, + } + } + let json = format!(r#"{{"rpc_uri":"{sentinel}"}}"#); + let error = serde_json::from_str::(&json).expect_err("serde credentials reject"); + assert!(!format!("{error} {error:?}").contains("SECRET")); + let json = format!(r#"{{"{sentinel}":0}}"#); + let error = serde_json::from_str::(&json).expect_err("unknown fields reject"); + assert!(!format!("{error} {error:?}").contains("SECRET")); + let error = CaptureNode::new("http://[bad]:1".to_owned()).expect_err("parser rejection"); + assert!(error.source().is_some(), "retain URL parser source"); + let error = CaptureNode::decode(&record(&[0xff])).expect_err("UTF-8 rejection"); + assert!(error.source().is_some(), "retain UTF-8 source"); + } +}