urma_runtime/p2p/
wire.rs

1use crate::config::{
2    P2P_MAX_ADDRESSES, P2P_MAX_HEADERS_PER_MESSAGE, P2P_MAX_INVENTORY, P2P_MAX_PAYLOAD_BYTES,
3};
4use crate::error::{Error, ensure};
5use bitcoin::block::Header;
6use bitcoin::consensus::encode::{Decodable, VarInt, serialize};
7use bitcoin::hashes::{Hash, sha256d};
8use bitcoin::p2p::Magic;
9use bitcoin::p2p::address::{AddrV2Message, Address};
10use bitcoin::p2p::message::{NetworkMessage, RawNetworkMessage};
11use bitcoin::p2p::message_blockdata::Inventory;
12use bitcoin::p2p::message_network::VersionMessage;
13use std::io::{Read, Write};
14
15pub enum Message {
16    Version(VersionMessage),
17    Verack,
18    Ping(u64),
19    Pong(u64),
20    Addr(Vec<(u32, Address)>),
21    AddrV2(Vec<AddrV2Message>),
22    Headers(Vec<Header>),
23    Inv(Vec<Inventory>),
24    NotFound(Vec<Inventory>),
25    Block(Vec<u8>),
26    Other(String),
27}
28
29impl Message {
30    pub fn command(&self) -> String {
31        match self {
32            Self::Version(_) => "version".into(),
33            Self::Verack => "verack".into(),
34            Self::Ping(_) => "ping".into(),
35            Self::Pong(_) => "pong".into(),
36            Self::Addr(rows) => format!("addr({})", rows.len()),
37            Self::AddrV2(rows) => format!("addrv2({})", rows.len()),
38            Self::Headers(rows) => format!("headers({})", rows.len()),
39            Self::Inv(rows) => format!("inv({})", rows.len()),
40            Self::NotFound(rows) => format!("notfound({})", rows.len()),
41            Self::Block(raw) => format!("block({})", raw.len()),
42            Self::Other(name) => name.clone(),
43        }
44    }
45}
46
47pub(super) fn send(
48    stream: &mut impl Write,
49    magic: [u8; 4],
50    payload: NetworkMessage,
51) -> Result<(), Error> {
52    let frame = RawNetworkMessage::new(Magic::from_bytes(magic), payload);
53    stream.write_all(&serialize(&frame))?;
54    stream.flush()?;
55    Ok(())
56}
57
58pub(super) fn receive(stream: &mut impl Read, magic: [u8; 4]) -> Result<Message, Error> {
59    let mut head = [0u8; 24];
60    stream.read_exact(&mut head)?;
61    ensure!(
62        head[..4] == magic,
63        "peer message carries a foreign network magic"
64    );
65    let command = command_name(&head[4..16])?;
66    let length = usize::try_from(u32::from_le_bytes([head[16], head[17], head[18], head[19]]))?;
67    ensure!(
68        length <= P2P_MAX_PAYLOAD_BYTES,
69        "peer message of {length} bytes exceeds the client payload limit"
70    );
71    let mut payload = Vec::new();
72    payload.try_reserve_exact(length)?;
73    payload.resize(length, 0);
74    stream.read_exact(&mut payload)?;
75    let digest = sha256d::Hash::hash(&payload);
76    ensure!(
77        digest.as_byte_array()[..4] == head[20..24],
78        "peer message checksum mismatch"
79    );
80    decode(&command, payload)
81}
82
83fn command_name(raw: &[u8]) -> Result<String, Error> {
84    let end = raw.iter().take_while(|byte| **byte != 0).count();
85    ensure!(
86        raw[end..].iter().all(|byte| *byte == 0),
87        "peer command has bytes after its terminator"
88    );
89    let name = std::str::from_utf8(&raw[..end])?;
90    ensure!(
91        name.bytes()
92            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit()),
93        "peer command is not lowercase ascii"
94    );
95    Ok(name.to_owned())
96}
97
98fn decode(command: &str, payload: Vec<u8>) -> Result<Message, Error> {
99    let mut cursor = payload.as_slice();
100    let message = match command {
101        "version" => Message::Version(exact(&mut cursor)?),
102        "verack" => Message::Verack,
103        "ping" => Message::Ping(exact(&mut cursor)?),
104        "pong" => Message::Pong(exact(&mut cursor)?),
105        "addr" => Message::Addr(bounded_list(&mut cursor, P2P_MAX_ADDRESSES, "addr")?),
106        "addrv2" => Message::AddrV2(bounded_list(&mut cursor, P2P_MAX_ADDRESSES, "addrv2")?),
107        "inv" => Message::Inv(bounded_list(&mut cursor, P2P_MAX_INVENTORY, "inv")?),
108        "notfound" => Message::NotFound(bounded_list(&mut cursor, P2P_MAX_INVENTORY, "notfound")?),
109        "headers" => Message::Headers(headers(&mut cursor)?),
110        "block" => return Ok(Message::Block(payload)),
111        other => return Ok(Message::Other(other.to_owned())),
112    };
113    ensure!(
114        cursor.is_empty(),
115        "peer {command} message has trailing bytes"
116    );
117    Ok(message)
118}
119
120fn exact<T: Decodable>(cursor: &mut &[u8]) -> Result<T, Error> {
121    Ok(T::consensus_decode_from_finite_reader(cursor)?)
122}
123
124fn bounded_list<T: Decodable>(
125    cursor: &mut &[u8],
126    limit: usize,
127    command: &str,
128) -> Result<Vec<T>, Error> {
129    let count = usize::try_from(VarInt::consensus_decode_from_finite_reader(cursor)?.0)?;
130    ensure!(
131        count <= limit,
132        "peer {command} message lists {count} items, above the client limit of {limit}"
133    );
134    let mut rows = Vec::new();
135    rows.try_reserve_exact(count)?;
136    for _index in 0..count {
137        rows.push(exact(cursor)?);
138    }
139    Ok(rows)
140}
141
142fn headers(cursor: &mut &[u8]) -> Result<Vec<Header>, Error> {
143    let count = usize::try_from(VarInt::consensus_decode_from_finite_reader(cursor)?.0)?;
144    ensure!(
145        count <= P2P_MAX_HEADERS_PER_MESSAGE,
146        "peer headers message lists {count} headers, above the protocol limit"
147    );
148    let mut rows = Vec::new();
149    rows.try_reserve_exact(count)?;
150    for _index in 0..count {
151        rows.push(exact::<Header>(cursor)?);
152        let transactions: u8 = exact(cursor)?;
153        ensure!(
154            transactions == 0,
155            "peer headers message embeds transactions"
156        );
157    }
158    Ok(rows)
159}
urma_runtime/p2p/wire.rs

urma_runtime/p2p/
wire.rs
