use ack_manager::Ack;
#[cfg(not(feature = "use-mock-crust"))]
use crust::PeerId;
use data::{AppendWrapper, Data, DataIdentifier};
use error::RoutingError;
use event::Event;
use id::{FullId, PublicId};
use itertools::Itertools;
use lru_time_cache::LruCache;
use maidsafe_utilities;
use maidsafe_utilities::serialisation::{deserialise, serialise};
#[cfg(feature = "use-mock-crust")]
use mock_crust::crust::PeerId;
use peer_manager::SectionMap;
use routing_table::{Prefix, Xorable};
use routing_table::Authority;
use rust_sodium::crypto::{box_, sign};
use rust_sodium::crypto::hash::sha256;
use std::collections::{BTreeMap, BTreeSet, HashSet};
use std::fmt::{self, Debug, Formatter};
use std::iter;
use std::time::Duration;
use super::QUORUM;
use types::MessageId;
use utils;
use xor_name::XorName;
pub const MAX_PART_LEN: usize = 20 * 1024;
pub const RELOCATE_PRIORITY: u8 = 1;
pub const DEFAULT_PRIORITY: u8 = 2;
pub const CLIENT_GET_PRIORITY: u8 = 3;
#[derive(Debug, RustcEncodable, RustcDecodable)]
pub enum Message {
Direct(DirectMessage),
Hop(HopMessage),
TunnelDirect {
content: DirectMessage,
src: PeerId,
dst: PeerId,
},
TunnelHop {
content: HopMessage,
src: PeerId,
dst: PeerId,
},
}
impl Message {
pub fn priority(&self) -> u8 {
match *self {
Message::Direct(ref content) |
Message::TunnelDirect { ref content, .. } => content.priority(),
Message::Hop(ref content) |
Message::TunnelHop { ref content, .. } => content.content.content.priority(),
}
}
}
#[derive(RustcEncodable, RustcDecodable)]
pub enum DirectMessage {
MessageSignature(sha256::Digest, sign::Signature),
SectionListSignature(Prefix<XorName>, SectionList, sign::Signature),
BootstrapIdentify {
public_id: PublicId,
},
BootstrapDeny,
ClientIdentify {
serialised_public_id: Vec<u8>,
signature: sign::Signature,
client_restriction: bool,
},
NodeIdentify {
serialised_public_id: Vec<u8>,
signature: sign::Signature,
},
CandidateIdentify {
serialised_public_id: Vec<u8>,
signature: sign::Signature,
},
TunnelRequest(PeerId),
TunnelSuccess(PeerId),
TunnelClosed(PeerId),
TunnelDisconnect(PeerId),
ResourceProof {
seed: Vec<u8>,
target_size: usize,
difficulty: u8,
},
ResourceProofResponse {
part_index: usize,
part_count: usize,
proof: Vec<u8>,
leading_zero_bytes: u64,
},
ResourceProofResponseReceipt,
}
impl DirectMessage {
pub fn priority(&self) -> u8 {
match *self {
DirectMessage::ResourceProofResponse { .. } => 9,
_ => 0,
}
}
}
#[derive(RustcEncodable, RustcDecodable)]
pub struct HopMessage {
pub content: SignedMessage,
pub route: u8,
pub sent_to: BTreeSet<XorName>,
signature: sign::Signature,
}
impl HopMessage {
pub fn new(content: SignedMessage,
route: u8,
sent_to: BTreeSet<XorName>,
signing_key: &sign::SecretKey)
-> Result<HopMessage, RoutingError> {
let bytes_to_sign = serialise(&content)?;
Ok(HopMessage {
content: content,
route: route,
sent_to: sent_to,
signature: sign::sign_detached(&bytes_to_sign, signing_key),
})
}
pub fn verify(&self, verification_key: &sign::PublicKey) -> Result<(), RoutingError> {
let signed_bytes = serialise(&self.content)?;
if sign::verify_detached(&self.signature, &signed_bytes, verification_key) {
Ok(())
} else {
Err(RoutingError::FailedSignature)
}
}
}
#[derive(Ord, PartialOrd, Eq, PartialEq, Clone, Hash, RustcEncodable, RustcDecodable, Debug)]
pub struct SectionList {
prefix: Prefix<XorName>,
pub_ids: BTreeSet<PublicId>,
}
impl SectionList {
pub fn new(prefix: Prefix<XorName>, pub_ids: BTreeSet<PublicId>) -> Self {
SectionList {
prefix: prefix,
pub_ids: pub_ids,
}
}
pub fn from<I: IntoIterator<Item = PublicId>>(prefix: Prefix<XorName>, pub_ids: I) -> Self {
Self::new(prefix, pub_ids.into_iter().collect())
}
}
#[derive(Ord, PartialOrd, Eq, PartialEq, Clone, Hash, RustcEncodable, RustcDecodable)]
pub struct SignedMessage {
content: RoutingMessage,
src_sections: Vec<SectionList>,
signatures: BTreeMap<PublicId, sign::Signature>,
}
impl SignedMessage {
pub fn new(content: RoutingMessage,
full_id: &FullId,
mut src_sections: Vec<SectionList>)
-> Result<SignedMessage, RoutingError> {
src_sections.sort_by_key(|list| list.prefix);
let sig = sign::sign_detached(&serialise(&content)?, full_id.signing_private_key());
Ok(SignedMessage {
content: content,
src_sections: src_sections,
signatures: iter::once((*full_id.public_id(), sig)).collect(),
})
}
pub fn check_integrity(&self, min_section_size: usize) -> Result<(), RoutingError> {
let signed_bytes = serialise(&self.content)?;
if !self.find_invalid_sigs(signed_bytes).is_empty() {
return Err(RoutingError::FailedSignature);
}
if !self.has_enough_sigs(min_section_size) {
return Err(RoutingError::NotEnoughSignatures);
}
Ok(())
}
pub fn signed_by(&self, pub_id: &PublicId) -> bool {
self.signatures.contains_key(pub_id)
}
pub fn add_signature(&mut self, pub_id: PublicId, sig: sign::Signature) {
if self.content.src.is_multiple() && self.is_sender(&pub_id) {
let _ = self.signatures.insert(pub_id, sig);
}
}
pub fn add_signatures(&mut self, msg: SignedMessage) {
if self.content.src.is_multiple() {
self.signatures.extend(msg.signatures);
}
}
pub fn into_routing_message(self) -> RoutingMessage {
self.content
}
pub fn routing_message(&self) -> &RoutingMessage {
&self.content
}
pub fn priority(&self) -> u8 {
self.content.priority()
}
pub fn check_fully_signed(&mut self, min_section_size: usize) -> bool {
if !self.has_enough_sigs(min_section_size) {
return false;
}
let signed_bytes = match serialise(&self.content) {
Ok(serialised) => serialised,
Err(error) => {
warn!("Failed to serialise {:?}: {:?}", self, error);
return false;
}
};
for invalid_signature in &self.find_invalid_sigs(signed_bytes) {
let _ = self.signatures.remove(invalid_signature);
}
self.has_enough_sigs(min_section_size)
}
fn is_sender(&self, pub_id: &PublicId) -> bool {
self.src_sections.iter().any(|list| list.pub_ids.contains(pub_id))
}
fn find_invalid_sigs(&self, signed_bytes: Vec<u8>) -> Vec<PublicId> {
let invalid = self.signatures
.iter()
.filter_map(|(pub_id, sig)| {
let is_valid = if let Authority::Client { ref client_key, .. } = self.content.src {
client_key == pub_id.signing_public_key() &&
sign::verify_detached(sig, &signed_bytes, client_key)
} else {
self.is_sender(pub_id) &&
sign::verify_detached(sig, &signed_bytes, pub_id.signing_public_key())
};
if is_valid { None } else { Some(*pub_id) }
})
.collect_vec();
if !invalid.is_empty() {
debug!("{:?}: invalid signatures: {:?}", self, invalid);
}
invalid
}
fn has_enough_sigs(&self, min_section_size: usize) -> bool {
use Authority::*;
match self.content.src {
ClientManager(_) | NaeManager(_) | NodeManager(_) => {
let valid_names: HashSet<_> = self.src_sections
.iter()
.flat_map(|list| list.pub_ids.iter().map(PublicId::name))
.sorted_by(|lhs, rhs| self.content.src.name().cmp_distance(lhs, rhs))
.into_iter()
.take(min_section_size)
.collect();
let valid_sigs = self.signatures
.keys()
.filter(|pub_id| valid_names.contains(pub_id.name()))
.count();
QUORUM * valid_names.len() <= 100 * valid_sigs
}
Section(_) => {
let num_sending =
self.src_sections.iter().fold(0, |count, list| count + list.pub_ids.len());
let valid_sigs = self.signatures.len();
QUORUM * num_sending <= 100 * valid_sigs
}
PrefixSection(_) => {
self.src_sections.iter().all(|list| {
let valid_sigs = self.signatures
.keys()
.filter(|pub_id| list.pub_ids.contains(pub_id))
.count();
QUORUM * list.pub_ids.len() <= 100 * valid_sigs
})
}
ManagedNode(_) | Client { .. } => self.signatures.len() == 1,
}
}
}
#[derive(Ord, PartialOrd, Eq, PartialEq, Clone, Hash, Debug, RustcEncodable, RustcDecodable)]
pub struct RoutingMessage {
pub src: Authority<XorName>,
pub dst: Authority<XorName>,
pub content: MessageContent,
}
impl RoutingMessage {
pub fn ack_from(msg: &RoutingMessage, src: Authority<XorName>) -> Result<Self, RoutingError> {
Ok(RoutingMessage {
src: src,
dst: msg.src,
content: MessageContent::Ack(Ack::compute(msg)?, msg.priority()),
})
}
pub fn priority(&self) -> u8 {
self.content.priority()
}
pub fn to_signature(&self,
signing_key: &sign::SecretKey)
-> Result<DirectMessage, RoutingError> {
let serialised_msg = serialise(self)?;
let hash = sha256::hash(&serialised_msg);
let sig = sign::sign_detached(&serialised_msg, signing_key);
Ok(DirectMessage::MessageSignature(hash, sig))
}
}
#[derive(Ord, PartialOrd, Eq, PartialEq, Clone, Hash, RustcEncodable, RustcDecodable)]
pub enum MessageContent {
GetNodeName {
current_id: PublicId,
message_id: MessageId,
},
ExpectCandidate {
expect_id: PublicId,
client_auth: Authority<XorName>,
message_id: MessageId,
},
ConnectionInfoRequest {
encrypted_conn_info: Vec<u8>,
nonce: [u8; box_::NONCEBYTES],
pub_id: PublicId,
msg_id: MessageId,
},
ConnectionInfoResponse {
encrypted_conn_info: Vec<u8>,
nonce: [u8; box_::NONCEBYTES],
pub_id: PublicId,
msg_id: MessageId,
},
GetNodeNameResponse {
relocated_id: PublicId,
section: BTreeSet<PublicId>,
message_id: MessageId,
},
SectionUpdate {
prefix: Prefix<XorName>,
members: BTreeSet<PublicId>,
},
RoutingTableRequest(MessageId, sha256::Digest),
RoutingTableResponse {
prefix: Prefix<XorName>,
members: BTreeSet<PublicId>,
message_id: MessageId,
},
SectionSplit(Prefix<XorName>, XorName),
OwnSectionMerge {
sender_prefix: Prefix<XorName>,
merge_prefix: Prefix<XorName>,
sections: SectionMap,
},
OtherSectionMerge {
prefix: Prefix<XorName>,
section: BTreeSet<PublicId>,
},
Ack(Ack, u8),
UserMessagePart {
hash: u64,
part_count: u32,
part_index: u32,
priority: u8,
cacheable: bool,
payload: Vec<u8>,
},
AcceptAsCandidate {
expect_id: PublicId,
client_auth: Authority<XorName>,
message_id: MessageId,
},
CandidateApproval {
candidate_id: PublicId,
client_auth: Authority<XorName>,
sections: SectionMap,
},
NodeApproval {
sections: SectionMap,
},
}
impl MessageContent {
pub fn priority(&self) -> u8 {
match *self {
MessageContent::Ack(_, priority) |
MessageContent::UserMessagePart { priority, .. } => priority,
_ => 0,
}
}
}
impl Debug for DirectMessage {
fn fmt(&self, formatter: &mut Formatter) -> fmt::Result {
use self::DirectMessage::*;
match *self {
MessageSignature(ref digest, _) => {
write!(formatter,
"MessageSignature ({}, ..)",
utils::format_binary_array(&digest.0))
}
SectionListSignature(ref prefix, _, _) => {
write!(formatter, "SectionListSignature({:?}, ..)", prefix)
}
BootstrapIdentify { ref public_id } => {
write!(formatter, "BootstrapIdentify {{ {:?} }}", public_id)
}
BootstrapDeny => write!(formatter, "BootstrapDeny"),
ClientIdentify { client_restriction: true, .. } => {
write!(formatter, "ClientIdentify (client only)")
}
ClientIdentify { client_restriction: false, .. } => {
write!(formatter, "ClientIdentify (joining node)")
}
NodeIdentify { .. } => write!(formatter, "NodeIdentify {{ .. }}"),
CandidateIdentify { .. } => write!(formatter, "CandidateIdentify {{ .. }}"),
TunnelRequest(peer_id) => write!(formatter, "TunnelRequest({:?})", peer_id),
TunnelSuccess(peer_id) => write!(formatter, "TunnelSuccess({:?})", peer_id),
TunnelClosed(peer_id) => write!(formatter, "TunnelClosed({:?})", peer_id),
TunnelDisconnect(peer_id) => write!(formatter, "TunnelDisconnect({:?})", peer_id),
ResourceProof { ref seed, ref target_size, ref difficulty } => {
write!(formatter,
"ResourceProof {{ seed: {:?}, target_size: {:?}, difficulty: {:?} }}",
seed,
target_size,
difficulty)
}
ResourceProofResponse { part_index, part_count, ref proof, leading_zero_bytes } => {
write!(formatter,
"ResourceProofResponse {{ part {}/{}, proof_len: {:?}, leading_zero_bytes: \
{:?} }}",
part_index + 1,
part_count,
proof.len(),
leading_zero_bytes)
}
ResourceProofResponseReceipt => write!(formatter, "ResourceProofResponseReceipt"),
}
}
}
impl Debug for HopMessage {
fn fmt(&self, formatter: &mut Formatter) -> fmt::Result {
write!(formatter,
"HopMessage {{ content: {:?}, route: {}, sent_to: .., signature: .. }}",
self.content,
self.route)
}
}
impl Debug for SignedMessage {
fn fmt(&self, formatter: &mut Formatter) -> fmt::Result {
write!(formatter,
"SignedMessage {{ content: {:?}, sending nodes: {:?}, signatures: {:?} }}",
self.content,
self.src_sections,
self.signatures.keys().collect_vec())
}
}
impl Debug for MessageContent {
fn fmt(&self, formatter: &mut Formatter) -> fmt::Result {
use self::MessageContent::*;
match *self {
GetNodeName { ref current_id, ref message_id } => {
write!(formatter,
"GetNodeName {{ {:?}, {:?} }}",
current_id,
message_id)
}
ExpectCandidate { ref expect_id, ref client_auth, ref message_id } => {
write!(formatter,
"ExpectCandidate {{ {:?}, {:?}, {:?} }}",
expect_id,
client_auth,
message_id)
}
ConnectionInfoRequest { ref pub_id, ref msg_id, .. } => {
write!(formatter,
"ConnectionInfoRequest {{ {:?}, {:?}, .. }}",
pub_id,
msg_id)
}
ConnectionInfoResponse { ref pub_id, ref msg_id, .. } => {
write!(formatter,
"ConnectionInfoResponse {{ {:?}, {:?}, .. }}",
pub_id,
msg_id)
}
GetNodeNameResponse { ref relocated_id, ref section, ref message_id } => {
write!(formatter,
"GetNodeNameResponse {{ {:?}, {:?}, {:?} }}",
relocated_id,
section,
message_id)
}
SectionUpdate { ref prefix, ref members } => {
write!(formatter, "SectionUpdate {{ {:?}, {:?} }}", prefix, members)
}
RoutingTableRequest(ref msg_id, ref digest) => {
write!(formatter,
"RoutingTableRequest({:?}, {})",
msg_id,
utils::format_binary_array(&digest.0))
}
RoutingTableResponse { ref prefix, ref members, ref message_id } => {
write!(formatter,
"RoutingTableResponse {{ {:?}, {:?}, {:?} }}",
prefix,
members,
message_id)
}
SectionSplit(ref prefix, ref joining_node) => {
write!(formatter, "SectionSplit({:?}, {:?})", prefix, joining_node)
}
OwnSectionMerge { ref sender_prefix, ref merge_prefix, ref sections } => {
write!(formatter,
"OwnSectionMerge {{ {:?}, {:?}, {:?} }}",
sender_prefix,
merge_prefix,
sections)
}
OtherSectionMerge { ref prefix, ref section } => {
write!(formatter,
"OtherSectionMerge {{ {:?}, {:?} }}",
prefix,
section)
}
Ack(ack, priority) => write!(formatter, "Ack({}, {})", ack, priority),
UserMessagePart { hash, part_count, part_index, priority, cacheable, .. } => {
write!(formatter,
"UserMessagePart {{ {}/{}, priority: {}, cacheable: {}, {:x} }}",
part_index + 1,
part_count,
priority,
cacheable,
hash)
}
AcceptAsCandidate { ref expect_id, ref client_auth, ref message_id } => {
write!(formatter,
"AcceptAsCandidate {{ {:?}, {:?}, {:?} }}",
expect_id,
client_auth,
message_id)
}
CandidateApproval { ref candidate_id, ref client_auth, ref sections } => {
write!(formatter,
"CandidateApproval {{ candidate_id: {:?}, client_auth: {:?}, sections: \
{:?} }}",
candidate_id,
client_auth,
sections)
}
NodeApproval { ref sections } => write!(formatter, "NodeApproval {{ {:?} }}", sections),
}
}
}
#[derive(Ord, PartialOrd, Eq, PartialEq, Clone, Debug, Hash, RustcEncodable, RustcDecodable)]
pub enum UserMessage {
Request(Request),
Response(Response),
}
impl UserMessage {
pub fn to_parts(&self, priority: u8) -> Result<Vec<MessageContent>, RoutingError> {
let hash = maidsafe_utilities::big_endian_sip_hash(self);
let payload = serialise(self)?;
let len = payload.len();
let part_count = (len + MAX_PART_LEN - 1) / MAX_PART_LEN;
Ok((0..part_count)
.map(|i| {
MessageContent::UserMessagePart {
hash: hash,
part_count: part_count as u32,
part_index: i as u32,
cacheable: self.is_cacheable(),
payload: payload[(i * len / part_count)..((i + 1) * len / part_count)].to_vec(),
priority: priority,
}
})
.collect())
}
pub fn from_parts<'a, I: Iterator<Item = &'a Vec<u8>>>(hash: u64,
parts: I)
-> Result<UserMessage, RoutingError> {
let mut payload = Vec::new();
for part in parts {
payload.extend_from_slice(part);
}
let user_msg = deserialise(&payload[..])?;
if hash != maidsafe_utilities::big_endian_sip_hash(&user_msg) {
Err(RoutingError::HashMismatch)
} else {
Ok(user_msg)
}
}
pub fn into_event(self, src: Authority<XorName>, dst: Authority<XorName>) -> Event {
match self {
UserMessage::Request(request) => {
Event::Request {
request: request,
src: src,
dst: dst,
}
}
UserMessage::Response(response) => {
Event::Response {
response: response,
src: src,
dst: dst,
}
}
}
}
fn is_cacheable(&self) -> bool {
match *self {
UserMessage::Request(ref request) => request.is_cacheable(),
UserMessage::Response(ref response) => response.is_cacheable(),
}
}
}
#[derive(Ord, PartialOrd, Eq, PartialEq, Clone, Hash, RustcEncodable, RustcDecodable)]
pub enum Request {
Refresh(Vec<u8>, MessageId),
Get(DataIdentifier, MessageId),
Put(Data, MessageId),
Post(Data, MessageId),
Delete(Data, MessageId),
Append(AppendWrapper, MessageId),
GetAccountInfo(MessageId),
}
#[derive(Ord, PartialOrd, Eq, PartialEq, Clone, Hash, RustcEncodable, RustcDecodable)]
pub enum Response {
GetSuccess(Data, MessageId),
PutSuccess(DataIdentifier, MessageId),
PostSuccess(DataIdentifier, MessageId),
DeleteSuccess(DataIdentifier, MessageId),
AppendSuccess(DataIdentifier, MessageId),
GetAccountInfoSuccess {
id: MessageId,
data_stored: u64,
space_available: u64,
},
GetFailure {
id: MessageId,
data_id: DataIdentifier,
external_error_indicator: Vec<u8>,
},
PutFailure {
id: MessageId,
data_id: DataIdentifier,
external_error_indicator: Vec<u8>,
},
PostFailure {
id: MessageId,
data_id: DataIdentifier,
external_error_indicator: Vec<u8>,
},
DeleteFailure {
id: MessageId,
data_id: DataIdentifier,
external_error_indicator: Vec<u8>,
},
AppendFailure {
id: MessageId,
data_id: DataIdentifier,
external_error_indicator: Vec<u8>,
},
GetAccountInfoFailure {
id: MessageId,
external_error_indicator: Vec<u8>,
},
}
impl Request {
pub fn priority(&self) -> u8 {
match *self {
Request::Refresh(..) => 2,
Request::Get(..) |
Request::GetAccountInfo(..) => 3,
Request::Append(..) => 4,
Request::Put(ref data, _) |
Request::Post(ref data, _) |
Request::Delete(ref data, _) => {
match *data {
Data::Structured(..) => 4,
_ => 5,
}
}
}
}
pub fn is_cacheable(&self) -> bool {
if let Request::Get(DataIdentifier::Immutable(..), _) = *self {
true
} else {
false
}
}
}
impl Response {
pub fn priority(&self) -> u8 {
match *self {
Response::GetSuccess(ref data, _) => {
match *data {
Data::Structured(..) => 4,
_ => 5,
}
}
Response::PutSuccess(..) |
Response::PostSuccess(..) |
Response::DeleteSuccess(..) |
Response::AppendSuccess(..) |
Response::GetAccountInfoSuccess { .. } |
Response::GetFailure { .. } |
Response::PutFailure { .. } |
Response::PostFailure { .. } |
Response::DeleteFailure { .. } |
Response::AppendFailure { .. } |
Response::GetAccountInfoFailure { .. } => 3,
}
}
pub fn is_cacheable(&self) -> bool {
if let Response::GetSuccess(Data::Immutable(..), _) = *self {
true
} else {
false
}
}
}
impl Debug for Request {
fn fmt(&self, formatter: &mut Formatter) -> fmt::Result {
match *self {
Request::Refresh(ref data, ref message_id) => {
write!(formatter,
"Refresh({}, {:?})",
utils::format_binary_array(data),
message_id)
}
Request::Get(ref data_request, ref message_id) => {
write!(formatter, "Get({:?}, {:?})", data_request, message_id)
}
Request::Put(ref data, ref message_id) => {
write!(formatter, "Put({:?}, {:?})", data, message_id)
}
Request::Post(ref data, ref message_id) => {
write!(formatter, "Post({:?}, {:?})", data, message_id)
}
Request::Delete(ref data, ref message_id) => {
write!(formatter, "Delete({:?}, {:?})", data, message_id)
}
Request::Append(ref wrapper, ref message_id) => {
write!(formatter, "Append({:?}, {:?})", wrapper, message_id)
}
Request::GetAccountInfo(ref message_id) => {
write!(formatter, "GetAccountInfo({:?})", message_id)
}
}
}
}
impl Debug for Response {
fn fmt(&self, formatter: &mut Formatter) -> fmt::Result {
match *self {
Response::GetSuccess(ref data, ref message_id) => {
write!(formatter, "GetSuccess({:?}, {:?})", data, message_id)
}
Response::PutSuccess(ref name, ref message_id) => {
write!(formatter, "PutSuccess({:?}, {:?})", name, message_id)
}
Response::PostSuccess(ref name, ref message_id) => {
write!(formatter, "PostSuccess({:?}, {:?})", name, message_id)
}
Response::DeleteSuccess(ref name, ref message_id) => {
write!(formatter, "DeleteSuccess({:?}, {:?})", name, message_id)
}
Response::AppendSuccess(ref name, ref message_id) => {
write!(formatter, "AppendSuccess({:?}, {:?})", name, message_id)
}
Response::GetAccountInfoSuccess { ref id, .. } => {
write!(formatter, "GetAccountInfoSuccess {{ {:?}, .. }}", id)
}
Response::GetFailure { ref id, ref data_id, .. } => {
write!(formatter, "GetFailure {{ {:?}, {:?}, .. }}", id, data_id)
}
Response::PutFailure { ref id, ref data_id, .. } => {
write!(formatter, "PutFailure {{ {:?}, {:?}, .. }}", id, data_id)
}
Response::PostFailure { ref id, ref data_id, .. } => {
write!(formatter, "PostFailure {{ {:?}, {:?}, .. }}", id, data_id)
}
Response::DeleteFailure { ref id, ref data_id, .. } => {
write!(formatter, "DeleteFailure {{ {:?}, {:?}, .. }}", id, data_id)
}
Response::AppendFailure { ref id, ref data_id, .. } => {
write!(formatter, "AppendFailure {{ {:?}, {:?}, .. }}", id, data_id)
}
Response::GetAccountInfoFailure { ref id, .. } => {
write!(formatter, "GetAccountInfoFailure {{ {:?}, .. }}", id)
}
}
}
}
pub struct UserMessageCache(LruCache<(u64, u32), BTreeMap<u32, Vec<u8>>>);
impl UserMessageCache {
pub fn with_expiry_duration(duration: Duration) -> Self {
UserMessageCache(LruCache::with_expiry_duration(duration))
}
pub fn add(&mut self,
hash: u64,
part_count: u32,
part_index: u32,
payload: Vec<u8>)
-> Option<UserMessage> {
{
let entry = self.0.entry((hash, part_count)).or_insert_with(BTreeMap::new);
if let Some(value) = entry.insert(part_index, payload) {
debug!("Duplicate message with value {:?}", value);
}
if entry.len() != part_count as usize {
return None;
}
}
self.0
.remove(&(hash, part_count))
.and_then(|part_map| UserMessage::from_parts(hash, part_map.values()).ok())
}
}
#[cfg(test)]
mod tests {
#[cfg(not(feature = "use-mock-crust"))]
use crust::PeerId;
use data::{Data, ImmutableData};
use id::FullId;
use maidsafe_utilities;
use maidsafe_utilities::serialisation::serialise;
#[cfg(feature = "use-mock-crust")]
use mock_crust::crust::PeerId;
use rand;
use routing_table::{Authority, Prefix};
use rust_sodium::crypto::hash::sha256;
use rust_sodium::crypto::sign;
use std::collections::BTreeSet;
use std::iter;
use super::*;
use types::MessageId;
use xor_name::XorName;
#[cfg(not(feature = "use-mock-crust"))]
fn make_peer_id() -> PeerId {
PeerId(*FullId::new().public_id().encrypting_public_key())
}
#[cfg(feature = "use-mock-crust")]
fn make_peer_id() -> PeerId {
PeerId(0)
}
#[test]
fn signed_message_check_integrity() {
let min_section_size = 1000;
let name: XorName = rand::random();
let full_id = FullId::new();
let routing_message = RoutingMessage {
src: Authority::Client {
client_key: *full_id.public_id().signing_public_key(),
peer_id: make_peer_id(),
proxy_node_name: name,
},
dst: Authority::ClientManager(name),
content: MessageContent::SectionSplit(Prefix::new(0, name), name),
};
let senders = iter::empty().collect();
let signed_message_result = SignedMessage::new(routing_message.clone(), &full_id, senders);
let mut signed_message = unwrap!(signed_message_result);
assert_eq!(routing_message, *signed_message.routing_message());
assert_eq!(1, signed_message.signatures.len());
assert_eq!(Some(full_id.public_id()),
signed_message.signatures.keys().next());
unwrap!(signed_message.check_integrity(min_section_size));
let full_id = FullId::new();
let bytes_to_sign = unwrap!(serialise(&(&routing_message, full_id.public_id())));
let signature = sign::sign_detached(&bytes_to_sign, full_id.signing_private_key());
signed_message.signatures = iter::once((*full_id.public_id(), signature)).collect();
assert!(signed_message.check_integrity(min_section_size).is_err());
assert!(signed_message.has_enough_sigs(min_section_size));
}
#[test]
fn msg_signatures() {
let min_section_size = 8;
let full_id_0 = FullId::new();
let prefix = Prefix::new(0, *full_id_0.public_id().name());
let full_id_1 = FullId::new();
let full_id_2 = FullId::new();
let irrelevant_full_id = FullId::new();
let data_bytes: Vec<u8> = (0..10).map(|i| i as u8).collect();
let data = Data::Immutable(ImmutableData::new(data_bytes));
let user_msg = UserMessage::Request(Request::Put(data, MessageId::new()));
let parts = unwrap!(user_msg.to_parts(1));
assert_eq!(1, parts.len());
let part = parts[0].clone();
let name: XorName = rand::random();
let routing_message = RoutingMessage {
src: Authority::ClientManager(name),
dst: Authority::ClientManager(name),
content: part,
};
let src_sections = vec![SectionList::from(prefix,
vec![*full_id_0.public_id(),
*full_id_1.public_id(),
*full_id_2.public_id()])];
let mut signed_msg = unwrap!(SignedMessage::new(routing_message, &full_id_0, src_sections));
assert_eq!(signed_msg.signatures.len(), 1);
let irrelevant_sig = match unwrap!(signed_msg.routing_message()
.to_signature(irrelevant_full_id.signing_private_key())) {
DirectMessage::MessageSignature(_, sig) => {
signed_msg.add_signature(*irrelevant_full_id.public_id(), sig);
sig
}
msg => panic!("Unexpected message: {:?}", msg),
};
assert_eq!(signed_msg.signatures.len(), 1);
assert!(!signed_msg.signatures.contains_key(irrelevant_full_id.public_id()));
assert!(!signed_msg.check_fully_signed(min_section_size));
match unwrap!(signed_msg.routing_message().to_signature(full_id_1.signing_private_key())) {
DirectMessage::MessageSignature(hash, sig) => {
let serialised_msg = unwrap!(serialise(signed_msg.routing_message()));
assert_eq!(hash, sha256::hash(&serialised_msg));
signed_msg.add_signature(*full_id_1.public_id(), sig);
}
msg => panic!("Unexpected message: {:?}", msg),
}
let bad_sig = sign::Signature([0; sign::SIGNATUREBYTES]);
signed_msg.add_signature(*full_id_2.public_id(), bad_sig);
assert_eq!(signed_msg.signatures.len(), 3);
assert!(signed_msg.check_fully_signed(min_section_size));
assert_eq!(signed_msg.signatures.len(), 2);
assert!(!signed_msg.signatures.contains_key(full_id_2.public_id()));
signed_msg.add_signature(*irrelevant_full_id.public_id(), irrelevant_sig);
assert_eq!(signed_msg.signatures.len(), 2);
assert!(!signed_msg.signatures.contains_key(irrelevant_full_id.public_id()));
}
#[test]
fn hop_message_verify() {
let name: XorName = rand::random();
let routing_message = RoutingMessage {
src: Authority::ClientManager(name),
dst: Authority::ClientManager(name),
content: MessageContent::SectionSplit(Prefix::new(0, name), name),
};
let full_id = FullId::new();
let senders = iter::empty().collect();
let signed_message_result = SignedMessage::new(routing_message.clone(), &full_id, senders);
let signed_message = unwrap!(signed_message_result);
let (public_signing_key, secret_signing_key) = sign::gen_keypair();
let hop_message_result = HopMessage::new(signed_message.clone(),
0,
BTreeSet::new(),
&secret_signing_key);
let hop_message = unwrap!(hop_message_result);
assert_eq!(signed_message, hop_message.content);
assert!(hop_message.verify(&public_signing_key).is_ok());
let (public_signing_key, _) = sign::gen_keypair();
assert!(hop_message.verify(&public_signing_key).is_err());
}
#[test]
fn user_message_parts() {
let data_bytes: Vec<u8> = (0..(MAX_PART_LEN * 2)).map(|i| i as u8).collect();
let data = Data::Immutable(ImmutableData::new(data_bytes));
let user_msg = UserMessage::Request(Request::Put(data, MessageId::new()));
let msg_hash = maidsafe_utilities::big_endian_sip_hash(&user_msg);
let parts = unwrap!(user_msg.to_parts(42));
assert_eq!(parts.len(), 3);
let payloads: Vec<Vec<u8>> = parts.into_iter()
.enumerate()
.map(|(i, msg)| match msg {
MessageContent::UserMessagePart { hash,
part_count,
part_index,
payload,
priority,
cacheable } => {
assert_eq!(msg_hash, hash);
assert_eq!(3, part_count);
assert_eq!(i, part_index as usize);
assert_eq!(42, priority);
assert!(!cacheable);
payload
}
msg => panic!("Unexpected message {:?}", msg),
})
.collect();
let deserialised_user_msg = unwrap!(UserMessage::from_parts(msg_hash, payloads.iter()));
assert_eq!(user_msg, deserialised_user_msg);
}
}