Parse v2 gossip.

Author: arik-so

lightning-rapid-gossip-sync/src/processing.rs

@@ -5,10 +5,8 @@ use core::sync::atomic::Ordering;
 use bitcoin::blockdata::constants::ChainHash;
 use bitcoin::secp256k1::PublicKey;
 
-use lightning::ln::msgs::{
-	DecodeError, ErrorAction, LightningError, UnsignedChannelUpdate,
-};
-use lightning::routing::gossip::NetworkGraph;
+use lightning::ln::msgs::{DecodeError, ErrorAction, LightningError, SocketAddress, UnsignedChannelUpdate, UnsignedNodeAnnouncement};
+use lightning::routing::gossip::{NetworkGraph, NodeAlias, NodeId};
 use lightning::util::logger::Logger;
 use lightning::{log_debug, log_warn, log_trace, log_given_level, log_gossip};
 use lightning::util::ser::{BigSize, Readable};
@@ -21,12 +19,14 @@ use std::time::{SystemTime, UNIX_EPOCH};
 
 #[cfg(all(not(feature = "std"), not(test)))]
 use alloc::{vec::Vec, borrow::ToOwned};
+use lightning::ln::features::NodeFeatures;
+use lightning::util::hash_tables::HashMap;
 
 /// The purpose of this prefix is to identify the serialization format, should other rapid gossip
 /// sync formats arise in the future.
 ///
 /// The fourth byte is the protocol version in case our format gets updated.
-const GOSSIP_PREFIX: [u8; 4] = [76, 68, 75, 1];
+const GOSSIP_PREFIX: [u8; 3] = [76, 68, 75];
 
 /// Maximum vector allocation capacity for distinct node IDs. This constraint is necessary to
 /// avoid malicious updates being able to trigger excessive memory allocation.
@@ -59,13 +59,22 @@ impl<NG: Deref<Target=NetworkGraph<L>>, L: Deref> RapidGossipSync<NG, L> where L
 		current_time_unix: Option<u64>
 	) -> Result<u32, GraphSyncError> {
 		log_trace!(self.logger, "Processing RGS data...");
-		let mut prefix = [0u8; 4];
-		read_cursor.read_exact(&mut prefix)?;
+		let mut protocol_prefix = [0u8; 3];
+		let mut version_prefix = [0u8; 1];
 
-		if prefix != GOSSIP_PREFIX {
+		read_cursor.read_exact(&mut protocol_prefix)?;
+		if protocol_prefix != GOSSIP_PREFIX {
 			return Err(DecodeError::UnknownVersion.into());
 		}
 
+		read_cursor.read_exact(&mut version_prefix)?;
+		let version = version_prefix[0];
+		if version != 1 && version != 2 {
+			return Err(DecodeError::UnknownVersion.into());
+		}
+
+		let parse_node_details = version == 2;
+
 		let chain_hash: ChainHash = Readable::read(read_cursor)?;
 		let ng_chain_hash = self.network_graph.get_chain_hash();
 		if chain_hash != ng_chain_hash {
@@ -88,14 +97,84 @@ impl<NG: Deref<Target=NetworkGraph<L>>, L: Deref> RapidGossipSync<NG, L> where L
 		// backdate the applied timestamp by a week
 		let backdated_timestamp = latest_seen_timestamp.saturating_sub(24 * 3600 * 7);
 
+		let mut default_node_features = Vec::new();
+		if parse_node_details {
+			let default_feature_count: u8 = Readable::read(read_cursor)?;
+			for _ in 0..default_feature_count {
+				let current_default_feature: NodeFeatures = Readable::read(read_cursor)?;
+				default_node_features.push(current_default_feature);
+			}
+		};
+
 		let node_id_count: u32 = Readable::read(read_cursor)?;
 		let mut node_ids: Vec<PublicKey> = Vec::with_capacity(core::cmp::min(
 			node_id_count,
 			MAX_INITIAL_NODE_ID_VECTOR_CAPACITY,
 		) as usize);
+		let mut node_modifications: HashMap<PublicKey, UnsignedNodeAnnouncement> = HashMap::default();
 		for _ in 0..node_id_count {
-			let current_node_id = Readable::read(read_cursor)?;
-			node_ids.push(current_node_id);
+			if parse_node_details {
+				let mut pubkey_bytes = [0u8; 33];
+				read_cursor.read_exact(&mut pubkey_bytes)?;
+				let node_detail_flag = pubkey_bytes[0];
+				let has_address_details = (node_detail_flag & (1 << 2)) > 0;
+				let feature_detail_marker = (node_detail_flag & (0b111 << 3)) >> 3;
+				let has_additional_data = (node_detail_flag & (1 << 7)) > 0;
+				let key_parity = node_detail_flag & 0b_0000_0011;
+				pubkey_bytes[0] = key_parity;
+				let current_node_id = PublicKey::from_slice(&pubkey_bytes).unwrap();
+				node_ids.push(current_node_id);
+
+				let mut unsigned_node_announcement = UnsignedNodeAnnouncement {
+					features: NodeFeatures::empty(),
+					timestamp: 0,
+					node_id: NodeId::from_pubkey(&current_node_id),
+					rgb: [0, 0, 0],
+					alias: NodeAlias([0u8; 32]),
+					addresses: Vec::new(),
+					excess_address_data: Vec::new(),
+					excess_data: Vec::new(),
+				};
+
+				if has_address_details {
+					let address_count: u8 = Readable::read(read_cursor)?;
+					let mut node_addresses: Vec<SocketAddress> = Vec::new();
+					for _ in 0..address_count {
+						let current_byte_count: u8 = Readable::read(read_cursor)?;
+						let mut address_bytes = vec![0u8; current_byte_count as usize];
+						read_cursor.read_exact(&mut address_bytes).unwrap();
+
+						let mut address_cursor = io::Cursor::new(&address_bytes);
+						if let Ok(current_address) = Readable::read(&mut address_cursor) {
+							node_addresses.push(current_address);
+						}
+					}
+					unsigned_node_announcement.addresses = node_addresses;
+				}
+
+				if feature_detail_marker > 0 {
+					if feature_detail_marker < 7 {
+						let feature_index = (feature_detail_marker - 1) as usize;
+						unsigned_node_announcement.features = default_node_features[feature_index].clone();
+					} else {
+						let node_features: NodeFeatures = Readable::read(read_cursor)?;
+						unsigned_node_announcement.features = node_features;
+					}
+				}
+
+				if feature_detail_marker > 0 || has_address_details {
+					node_modifications.insert(current_node_id, unsigned_node_announcement);
+				}
+
+				if has_additional_data {
+					// forwards compatibility
+					let _additional_data: Vec<u8> = Readable::read(read_cursor)?;
+				}
+
+			} else {
+				let current_node_id = Readable::read(read_cursor)?;
+				node_ids.push(current_node_id);
+			}
 		}
 
 		let network_graph = &self.network_graph;
@@ -114,6 +193,7 @@ impl<NG: Deref<Target=NetworkGraph<L>>, L: Deref> RapidGossipSync<NG, L> where L
 
 			let node_id_1_index: BigSize = Readable::read(read_cursor)?;
 			let node_id_2_index: BigSize = Readable::read(read_cursor)?;
+			let has_additional_data = (node_id_2_index.0 & (1 << 63)) > 0;
 
 			if max(node_id_1_index.0, node_id_2_index.0) >= node_id_count as u64 {
 				return Err(DecodeError::InvalidValue.into());
@@ -139,6 +219,18 @@ impl<NG: Deref<Target=NetworkGraph<L>>, L: Deref> RapidGossipSync<NG, L> where L
 					return Err(lightning_error.into());
 				}
 			}
+
+			if version >= 2 && has_additional_data {
+				// forwards compatibility
+				let _additional_data: Vec<u8> = Readable::read(read_cursor)?;
+			}
+
+			if let Some(modification) = node_modifications.remove(&node_id_1) {
+				network_graph.update_node_from_unsigned_announcement(&modification)?;
+			}
+			if let Some(modification) = node_modifications.remove(&node_id_2) {
+				network_graph.update_node_from_unsigned_announcement(&modification)?;
+			}
 		}
 
 		previous_scid = 0; // updates start at a new scid
@@ -157,6 +249,8 @@ impl<NG: Deref<Target=NetworkGraph<L>>, L: Deref> RapidGossipSync<NG, L> where L
 		let default_fee_proportional_millionths: u32 = Readable::read(&mut read_cursor)?;
 		let default_htlc_maximum_msat: u64 = Readable::read(&mut read_cursor)?;
 
+		let mut previous_channel_direction = None;
+
 		for _ in 0..update_count {
 			let scid_delta: BigSize = Readable::read(read_cursor)?;
 			let short_channel_id = previous_scid
@@ -166,6 +260,18 @@ impl<NG: Deref<Target=NetworkGraph<L>>, L: Deref> RapidGossipSync<NG, L> where L
 
 			let channel_flags: u8 = Readable::read(read_cursor)?;
 
+			if version >= 2 {
+				let direction = (channel_flags & 1) == 1;
+				let is_same_direction_update = Some(direction) == previous_channel_direction;
+				previous_channel_direction = Some(direction);
+
+				if scid_delta.0 == 0 && is_same_direction_update {
+					// this is additional data for forwards compatibility
+					let _additional_data: Vec<u8> = Readable::read(read_cursor)?;
+					continue;
+				}
+			}
+
 			// flags are always sent in full, and hence always need updating
 			let standard_channel_flags = channel_flags & 0b_0000_0011;
 
@@ -265,7 +371,7 @@ mod tests {
 	#[cfg(feature = "std")]
 	use lightning::ln::msgs::DecodeError;
 
-	use lightning::routing::gossip::NetworkGraph;
+	use lightning::routing::gossip::{NetworkGraph, NodeId};
 	use lightning::util::test_utils::TestLogger;
 
 	use crate::processing::STALE_RGS_UPDATE_AGE_LIMIT_SECS;
@@ -322,6 +428,42 @@ mod tests {
 		}
 	}
 
+	#[test]
+	fn node_data_update_succeeds_with_v2() {
+		let logger = TestLogger::new();
+		let network_graph = NetworkGraph::new(Network::Bitcoin, &logger);
+
+		let example_input = vec![
+			76, 68, 75, 2, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
+			79, 147, 30, 131, 101, 225, 90, 8, 156, 104, 214, 25, 0, 0, 0, 0, 0, 102, 87, 151, 80,
+			1, 0, 2, 23, 48, 0, 0, 0, 3, 15, 27, 132, 197, 86, 123, 18, 100, 64, 153, 93, 62, 213,
+			170, 186, 5, 101, 215, 30, 24, 52, 96, 72, 25, 255, 156, 23, 245, 233, 213, 221, 7, 143,
+			5, 14, 5, 10, 103, 111, 111, 103, 108, 101, 46, 99, 111, 109, 1, 187, 13, 3, 1, 2, 3, 4,
+			5, 6, 7, 8, 9, 10, 11, 12, 38, 4, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1,
+			1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 2, 3, 0, 4, 7, 1, 127, 0, 0, 1, 37, 163,
+			19, 2, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 5, 57, 2, 77, 75, 108, 209, 54,
+			16, 50, 202, 155, 210, 174, 185, 217, 0, 170, 77, 69, 217, 234, 216, 10, 201, 66, 51,
+			116, 196, 81, 167, 37, 77, 7, 102, 2, 83, 31, 230, 6, 129, 52, 80, 61, 39, 35, 19, 50,
+			39, 200, 103, 172, 143, 166, 200, 60, 83, 126, 154, 68, 195, 197, 189, 189, 203, 31,
+			227, 55, 0, 0, 0, 1, 0, 0, 1, 0, 1, 0, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+			6, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 1
+		];
+
+		let rapid_sync = RapidGossipSync::new(&network_graph, &logger);
+		let update_result = rapid_sync.update_network_graph_no_std(&example_input[..], None);
+		assert!(update_result.is_ok());
+
+		let read_only_graph = network_graph.read_only();
+		let nodes = read_only_graph.nodes();
+		assert_eq!(nodes.len(), 2);
+
+		let node_id = NodeId::from_slice(&[3, 27, 132, 197, 86, 123, 18, 100, 64, 153, 93, 62, 213, 170, 186, 5, 101, 215, 30, 24, 52, 96, 72, 25, 255, 156, 23, 245, 233, 213, 221, 7, 143]).unwrap();
+		let node = nodes.get(&node_id).unwrap();
+		let announcement_info = node.announcement_info.as_ref().unwrap();
+		let addresses = announcement_info.addresses();
+		assert_eq!(addresses.len(), 5);
+	}
+
 	#[test]
 	#[cfg(feature = "std")]
 	fn incremental_only_update_ignores_missing_channel() {
@@ -645,7 +787,7 @@ mod tests {
 	#[cfg(feature = "std")]
 	pub fn update_fails_with_unknown_version() {
 		let unknown_version_input = vec![
-			76, 68, 75, 2, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
+			76, 68, 75, 3, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
 			79, 147, 30, 131, 101, 225, 90, 8, 156, 104, 214, 25, 0, 0, 0, 0, 0, 97, 227, 98, 218,
 			0, 0, 0, 4, 2, 22, 7, 207, 206, 25, 164, 197, 231, 230, 231, 56, 102, 61, 250, 251,
 			187, 172, 38, 46, 79, 247, 108, 44, 155, 48, 219, 238, 252, 53, 192, 6, 67, 2, 36, 125,