use num::FromPrimitive;
use std::fmt;
use std::io::{Read, Write};
use std::time;
use crate::core::core::hash::Hash;
use crate::core::core::BlockHeader;
use crate::core::pow::Difficulty;
use crate::core::ser::{self, FixedLength, Readable, Reader, StreamingReader, Writeable, Writer};
use crate::core::{consensus, global};
use crate::types::{
Capabilities, Error, PeerAddr, ReasonForBan, MAX_BLOCK_HEADERS, MAX_LOCATORS, MAX_PEER_ADDRS,
};
use crate::util::read_write::read_exact;
const PROTOCOL_VERSION: u32 = 1;
pub const USER_AGENT: &'static str = concat!("MW/Grin ", env!("CARGO_PKG_VERSION"));
const OTHER_MAGIC: [u8; 2] = [73, 43];
const FLOONET_MAGIC: [u8; 2] = [83, 59];
const MAINNET_MAGIC: [u8; 2] = [97, 61];
enum_from_primitive! {
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum Type {
Error = 0,
Hand = 1,
Shake = 2,
Ping = 3,
Pong = 4,
GetPeerAddrs = 5,
PeerAddrs = 6,
GetHeaders = 7,
Header = 8,
Headers = 9,
GetBlock = 10,
Block = 11,
GetCompactBlock = 12,
CompactBlock = 13,
StemTransaction = 14,
Transaction = 15,
TxHashSetRequest = 16,
TxHashSetArchive = 17,
BanReason = 18,
GetTransaction = 19,
TransactionKernel = 20,
KernelDataRequest = 21,
KernelDataResponse = 22,
}
}
fn max_block_size() -> u64 {
(global::max_block_weight() / consensus::BLOCK_OUTPUT_WEIGHT * 708) as u64
}
fn default_max_msg_size() -> u64 {
max_block_size()
}
fn max_msg_size(msg_type: Type) -> u64 {
match msg_type {
Type::Error => 0,
Type::Hand => 128,
Type::Shake => 88,
Type::Ping => 16,
Type::Pong => 16,
Type::GetPeerAddrs => 4,
Type::PeerAddrs => 4 + (1 + 16 + 2) * MAX_PEER_ADDRS as u64,
Type::GetHeaders => 1 + 32 * MAX_LOCATORS as u64,
Type::Header => 365,
Type::Headers => 2 + 365 * MAX_BLOCK_HEADERS as u64,
Type::GetBlock => 32,
Type::Block => max_block_size(),
Type::GetCompactBlock => 32,
Type::CompactBlock => max_block_size() / 10,
Type::StemTransaction => max_block_size(),
Type::Transaction => max_block_size(),
Type::TxHashSetRequest => 40,
Type::TxHashSetArchive => 64,
Type::BanReason => 64,
Type::GetTransaction => 32,
Type::TransactionKernel => 32,
Type::KernelDataRequest => 0,
Type::KernelDataResponse => 8,
}
}
fn magic() -> [u8; 2] {
match *global::CHAIN_TYPE.read() {
global::ChainTypes::Floonet => FLOONET_MAGIC,
global::ChainTypes::Mainnet => MAINNET_MAGIC,
_ => OTHER_MAGIC,
}
}
pub fn read_header(
stream: &mut dyn Read,
msg_type: Option<Type>,
) -> Result<MsgHeaderWrapper, Error> {
let mut head = vec![0u8; MsgHeader::LEN];
if Some(Type::Hand) == msg_type {
read_exact(stream, &mut head, time::Duration::from_millis(10), true)?;
} else {
read_exact(stream, &mut head, time::Duration::from_secs(10), false)?;
}
let header = ser::deserialize::<MsgHeaderWrapper>(&mut &head[..])?;
Ok(header)
}
pub fn read_item<T: Readable>(stream: &mut dyn Read) -> Result<(T, u64), Error> {
let timeout = time::Duration::from_secs(20);
let mut reader = StreamingReader::new(stream, timeout);
let res = T::read(&mut reader)?;
Ok((res, reader.total_bytes_read()))
}
pub fn read_body<T: Readable>(h: &MsgHeader, stream: &mut dyn Read) -> Result<T, Error> {
let mut body = vec![0u8; h.msg_len as usize];
read_exact(stream, &mut body, time::Duration::from_secs(20), true)?;
ser::deserialize(&mut &body[..]).map_err(From::from)
}
pub fn read_discard(msg_len: u64, stream: &mut dyn Read) -> Result<(), Error> {
let mut buffer = vec![0u8; msg_len as usize];
read_exact(stream, &mut buffer, time::Duration::from_secs(20), true)?;
Ok(())
}
pub fn read_message<T: Readable>(stream: &mut dyn Read, msg_type: Type) -> Result<T, Error> {
match read_header(stream, Some(msg_type))? {
MsgHeaderWrapper::Known(header) => {
if header.msg_type == msg_type {
read_body(&header, stream)
} else {
Err(Error::BadMessage)
}
}
MsgHeaderWrapper::Unknown(msg_len) => {
read_discard(msg_len, stream)?;
Err(Error::BadMessage)
}
}
}
pub fn write_to_buf<T: Writeable>(msg: T, msg_type: Type) -> Result<Vec<u8>, Error> {
let mut body_buf = vec![];
ser::serialize(&mut body_buf, &msg)?;
let mut msg_buf = vec![];
let blen = body_buf.len() as u64;
ser::serialize(&mut msg_buf, &MsgHeader::new(msg_type, blen))?;
msg_buf.append(&mut body_buf);
Ok(msg_buf)
}
pub fn write_message<T: Writeable>(
stream: &mut dyn Write,
msg: T,
msg_type: Type,
) -> Result<(), Error> {
let buf = write_to_buf(msg, msg_type)?;
stream.write_all(&buf[..])?;
Ok(())
}
#[derive(Clone)]
pub enum MsgHeaderWrapper {
Known(MsgHeader),
Unknown(u64),
}
#[derive(Clone)]
pub struct MsgHeader {
magic: [u8; 2],
pub msg_type: Type,
pub msg_len: u64,
}
impl MsgHeader {
pub fn new(msg_type: Type, len: u64) -> MsgHeader {
MsgHeader {
magic: magic(),
msg_type: msg_type,
msg_len: len,
}
}
}
impl FixedLength for MsgHeader {
const LEN: usize = 2 + 1 + 8;
}
impl Writeable for MsgHeader {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
ser_multiwrite!(
writer,
[write_u8, self.magic[0]],
[write_u8, self.magic[1]],
[write_u8, self.msg_type as u8],
[write_u64, self.msg_len]
);
Ok(())
}
}
impl Readable for MsgHeaderWrapper {
fn read(reader: &mut dyn Reader) -> Result<MsgHeaderWrapper, ser::Error> {
let m = magic();
reader.expect_u8(m[0])?;
reader.expect_u8(m[1])?;
let (t, msg_len) = ser_multiread!(reader, read_u8, read_u64);
match Type::from_u8(t) {
Some(msg_type) => {
let max_len = max_msg_size(msg_type) * 4;
if msg_len > max_len {
error!(
"Too large read {:?}, max_len: {}, msg_len: {}.",
msg_type, max_len, msg_len
);
return Err(ser::Error::TooLargeReadErr);
}
Ok(MsgHeaderWrapper::Known(MsgHeader {
magic: m,
msg_type,
msg_len,
}))
}
None => {
let max_len = default_max_msg_size() * 4;
if msg_len > max_len {
error!(
"Too large read (unknown msg type) {:?}, max_len: {}, msg_len: {}.",
t, max_len, msg_len
);
return Err(ser::Error::TooLargeReadErr);
}
Ok(MsgHeaderWrapper::Unknown(msg_len))
}
}
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialOrd, PartialEq, Serialize)]
pub struct ProtocolVersion(pub u32);
impl Default for ProtocolVersion {
fn default() -> ProtocolVersion {
ProtocolVersion(PROTOCOL_VERSION)
}
}
impl From<ProtocolVersion> for u32 {
fn from(v: ProtocolVersion) -> u32 {
v.0
}
}
impl fmt::Display for ProtocolVersion {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "{}", self.0)
}
}
impl Writeable for ProtocolVersion {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
writer.write_u32(self.0)
}
}
impl Readable for ProtocolVersion {
fn read(reader: &mut dyn Reader) -> Result<ProtocolVersion, ser::Error> {
let version = reader.read_u32()?;
Ok(ProtocolVersion(version))
}
}
pub struct Hand {
pub version: ProtocolVersion,
pub capabilities: Capabilities,
pub nonce: u64,
pub genesis: Hash,
pub total_difficulty: Difficulty,
pub sender_addr: PeerAddr,
pub receiver_addr: PeerAddr,
pub user_agent: String,
}
impl Writeable for Hand {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
self.version.write(writer)?;
ser_multiwrite!(
writer,
[write_u32, self.capabilities.bits()],
[write_u64, self.nonce]
);
self.total_difficulty.write(writer)?;
self.sender_addr.write(writer)?;
self.receiver_addr.write(writer)?;
writer.write_bytes(&self.user_agent)?;
self.genesis.write(writer)?;
Ok(())
}
}
impl Readable for Hand {
fn read(reader: &mut dyn Reader) -> Result<Hand, ser::Error> {
let version = ProtocolVersion::read(reader)?;
let (capab, nonce) = ser_multiread!(reader, read_u32, read_u64);
let capabilities = Capabilities::from_bits_truncate(capab);
let total_difficulty = Difficulty::read(reader)?;
let sender_addr = PeerAddr::read(reader)?;
let receiver_addr = PeerAddr::read(reader)?;
let ua = reader.read_bytes_len_prefix()?;
let user_agent = String::from_utf8(ua).map_err(|_| ser::Error::CorruptedData)?;
let genesis = Hash::read(reader)?;
Ok(Hand {
version,
capabilities,
nonce,
genesis,
total_difficulty,
sender_addr,
receiver_addr,
user_agent,
})
}
}
pub struct Shake {
pub version: ProtocolVersion,
pub capabilities: Capabilities,
pub genesis: Hash,
pub total_difficulty: Difficulty,
pub user_agent: String,
}
impl Writeable for Shake {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
self.version.write(writer)?;
writer.write_u32(self.capabilities.bits())?;
self.total_difficulty.write(writer)?;
writer.write_bytes(&self.user_agent)?;
self.genesis.write(writer)?;
Ok(())
}
}
impl Readable for Shake {
fn read(reader: &mut dyn Reader) -> Result<Shake, ser::Error> {
let version = ProtocolVersion::read(reader)?;
let capab = reader.read_u32()?;
let capabilities = Capabilities::from_bits_truncate(capab);
let total_difficulty = Difficulty::read(reader)?;
let ua = reader.read_bytes_len_prefix()?;
let user_agent = String::from_utf8(ua).map_err(|_| ser::Error::CorruptedData)?;
let genesis = Hash::read(reader)?;
Ok(Shake {
version,
capabilities,
genesis,
total_difficulty,
user_agent,
})
}
}
pub struct GetPeerAddrs {
pub capabilities: Capabilities,
}
impl Writeable for GetPeerAddrs {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
writer.write_u32(self.capabilities.bits())
}
}
impl Readable for GetPeerAddrs {
fn read(reader: &mut dyn Reader) -> Result<GetPeerAddrs, ser::Error> {
let capab = reader.read_u32()?;
let capabilities = Capabilities::from_bits_truncate(capab);
Ok(GetPeerAddrs { capabilities })
}
}
#[derive(Debug)]
pub struct PeerAddrs {
pub peers: Vec<PeerAddr>,
}
impl Writeable for PeerAddrs {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
writer.write_u32(self.peers.len() as u32)?;
for p in &self.peers {
p.write(writer)?;
}
Ok(())
}
}
impl Readable for PeerAddrs {
fn read(reader: &mut dyn Reader) -> Result<PeerAddrs, ser::Error> {
let peer_count = reader.read_u32()?;
if peer_count > MAX_PEER_ADDRS {
return Err(ser::Error::TooLargeReadErr);
} else if peer_count == 0 {
return Ok(PeerAddrs { peers: vec![] });
}
let mut peers = Vec::with_capacity(peer_count as usize);
for _ in 0..peer_count {
peers.push(PeerAddr::read(reader)?);
}
Ok(PeerAddrs { peers: peers })
}
}
pub struct PeerError {
pub code: u32,
pub message: String,
}
impl Writeable for PeerError {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
ser_multiwrite!(writer, [write_u32, self.code], [write_bytes, &self.message]);
Ok(())
}
}
impl Readable for PeerError {
fn read(reader: &mut dyn Reader) -> Result<PeerError, ser::Error> {
let code = reader.read_u32()?;
let msg = reader.read_bytes_len_prefix()?;
let message = String::from_utf8(msg).map_err(|_| ser::Error::CorruptedData)?;
Ok(PeerError {
code: code,
message: message,
})
}
}
#[derive(Debug)]
pub struct Locator {
pub hashes: Vec<Hash>,
}
impl Writeable for Locator {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
writer.write_u8(self.hashes.len() as u8)?;
for h in &self.hashes {
h.write(writer)?
}
Ok(())
}
}
impl Readable for Locator {
fn read(reader: &mut dyn Reader) -> Result<Locator, ser::Error> {
let len = reader.read_u8()?;
if len > (MAX_LOCATORS as u8) {
return Err(ser::Error::TooLargeReadErr);
}
let mut hashes = Vec::with_capacity(len as usize);
for _ in 0..len {
hashes.push(Hash::read(reader)?);
}
Ok(Locator { hashes: hashes })
}
}
pub struct Headers {
pub headers: Vec<BlockHeader>,
}
impl Writeable for Headers {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
writer.write_u16(self.headers.len() as u16)?;
for h in &self.headers {
h.write(writer)?
}
Ok(())
}
}
pub struct Ping {
pub total_difficulty: Difficulty,
pub height: u64,
}
impl Writeable for Ping {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
self.total_difficulty.write(writer)?;
self.height.write(writer)?;
Ok(())
}
}
impl Readable for Ping {
fn read(reader: &mut dyn Reader) -> Result<Ping, ser::Error> {
let total_difficulty = Difficulty::read(reader)?;
let height = reader.read_u64()?;
Ok(Ping {
total_difficulty,
height,
})
}
}
pub struct Pong {
pub total_difficulty: Difficulty,
pub height: u64,
}
impl Writeable for Pong {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
self.total_difficulty.write(writer)?;
self.height.write(writer)?;
Ok(())
}
}
impl Readable for Pong {
fn read(reader: &mut dyn Reader) -> Result<Pong, ser::Error> {
let total_difficulty = Difficulty::read(reader)?;
let height = reader.read_u64()?;
Ok(Pong {
total_difficulty,
height,
})
}
}
#[derive(Debug)]
pub struct BanReason {
pub ban_reason: ReasonForBan,
}
impl Writeable for BanReason {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
let ban_reason_i32 = self.ban_reason as i32;
ban_reason_i32.write(writer)?;
Ok(())
}
}
impl Readable for BanReason {
fn read(reader: &mut dyn Reader) -> Result<BanReason, ser::Error> {
let ban_reason_i32 = match reader.read_i32() {
Ok(h) => h,
Err(_) => 0,
};
let ban_reason = ReasonForBan::from_i32(ban_reason_i32).ok_or(ser::Error::CorruptedData)?;
Ok(BanReason { ban_reason })
}
}
pub struct TxHashSetRequest {
pub hash: Hash,
pub height: u64,
}
impl Writeable for TxHashSetRequest {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
self.hash.write(writer)?;
writer.write_u64(self.height)?;
Ok(())
}
}
impl Readable for TxHashSetRequest {
fn read(reader: &mut dyn Reader) -> Result<TxHashSetRequest, ser::Error> {
Ok(TxHashSetRequest {
hash: Hash::read(reader)?,
height: reader.read_u64()?,
})
}
}
pub struct TxHashSetArchive {
pub hash: Hash,
pub height: u64,
pub bytes: u64,
}
impl Writeable for TxHashSetArchive {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
self.hash.write(writer)?;
ser_multiwrite!(writer, [write_u64, self.height], [write_u64, self.bytes]);
Ok(())
}
}
impl Readable for TxHashSetArchive {
fn read(reader: &mut dyn Reader) -> Result<TxHashSetArchive, ser::Error> {
let hash = Hash::read(reader)?;
let (height, bytes) = ser_multiread!(reader, read_u64, read_u64);
Ok(TxHashSetArchive {
hash,
height,
bytes,
})
}
}
pub struct KernelDataRequest {}
impl Writeable for KernelDataRequest {
fn write<W: Writer>(&self, _writer: &mut W) -> Result<(), ser::Error> {
Ok(())
}
}
pub struct KernelDataResponse {
pub bytes: u64,
}
impl Writeable for KernelDataResponse {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
writer.write_u64(self.bytes)?;
Ok(())
}
}
impl Readable for KernelDataResponse {
fn read(reader: &mut dyn Reader) -> Result<KernelDataResponse, ser::Error> {
let bytes = reader.read_u64()?;
Ok(KernelDataResponse { bytes })
}
}