use core::cmp;
use core::fmt;
use derive_builder::Builder;
use serde::de::{SeqAccess, Visitor};
use serde::ser::{SerializeSeq, Serializer};
use serde::{de, Deserialize, Deserializer, Serialize};
use std::convert::TryFrom;
use std::time::{SystemTime, UNIX_EPOCH};
use super::canonical::*;
use super::crc::*;
use super::dtntime::*;
use super::eid::*;
use super::primary::*;
pub const DTN_VERSION: u32 = 7;
pub type ByteBuffer = Vec<u8>;
pub type DtnVersionType = u32;
pub type CanonicalBlockNumberType = u64;
pub type FragOffsetType = u64;
pub type LifetimeType = u64;
pub type TotalDataLengthType = u64;
#[derive(Debug, Clone)]
pub enum Bp7Error {
CanonicalBlockError(String),
PrimaryBlockError(String),
EIDError(String),
DtnTimeError(String),
CrcError(String),
BundleError(String),
StcpError(String),
BundleControlFlagError(String),
BlockControlFlagError(String),
}
impl fmt::Display for Bp7Error {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "{:?}", self)
}
}
pub type Bp7ErrorList = Vec<Bp7Error>;
pub trait Block {
fn to_cbor(&self) -> ByteBuffer;
}
pub type BlockControlFlags = u8;
pub const BLOCK_DELETE_BUNDLE: BlockControlFlags = 0x08;
pub const BLOCK_STATUS_REPORT: BlockControlFlags = 0x04;
pub const BLOCK_REMOVE: BlockControlFlags = 0x02;
pub const BLOCK_REPLICATE: BlockControlFlags = 0x01;
pub const BLOCK_CFRESERVED_FIELDS: BlockControlFlags = 0xF0;
pub trait BlockValidation {
fn has(self, flag: BlockControlFlags) -> bool;
fn validation_error(self) -> Option<Bp7Error>;
}
impl BlockValidation for BlockControlFlags {
fn has(self, flag: BlockControlFlags) -> bool {
(self & flag) != 0
}
fn validation_error(self) -> Option<Bp7Error> {
if self.has(BLOCK_CFRESERVED_FIELDS) {
return Some(Bp7Error::BlockControlFlagError(
"Given flag contains reserved bits".to_string(),
));
}
None
}
}
pub type BundleControlFlags = u16;
pub const BUNDLE_STATUS_REQUEST_DELETION: BundleControlFlags = 0x1000;
pub const BUNDLE_STATUS_REQUEST_DELIVERY: BundleControlFlags = 0x0800;
pub const BUNDLE_STATUS_REQUEST_FORWARD: BundleControlFlags = 0x0400;
pub const BUNDLE_STATUS_REQUEST_RECEPTION: BundleControlFlags = 0x0100;
pub const BUNDLE_CONTAINS_MANIFEST: BundleControlFlags = 0x0080;
pub const BUNDLE_REQUEST_STATUS_TIME: BundleControlFlags = 0x0040;
pub const BUNDLE_REQUEST_USER_APPLICATION_ACK: BundleControlFlags = 0x0020;
pub const BUNDLE_MUST_NOT_FRAGMENTED: BundleControlFlags = 0x0004;
pub const BUNDLE_ADMINISTRATIVE_RECORD_PAYLOAD: BundleControlFlags = 0x0002;
pub const BUNDLE_IS_FRAGMENT: BundleControlFlags = 0x0001;
pub const BUNDLE_CFRESERVED_FIELDS: BundleControlFlags = 0xE218;
pub trait BundleValidation {
fn has(self, flag: BundleControlFlags) -> bool;
fn validation_errors(self) -> Option<Bp7ErrorList>;
}
impl BundleValidation for BundleControlFlags {
fn has(self, flag: BundleControlFlags) -> bool {
(self & flag) != 0
}
fn validation_errors(self) -> Option<Bp7ErrorList> {
let mut errors: Bp7ErrorList = Vec::new();
if self.has(BUNDLE_CFRESERVED_FIELDS) {
errors.push(Bp7Error::BundleControlFlagError(
"Given flag contains reserved bits".to_string(),
));
}
if self.has(BUNDLE_IS_FRAGMENT) && self.has(BUNDLE_MUST_NOT_FRAGMENTED) {
errors.push(Bp7Error::BundleControlFlagError(
"Both 'bundle is a fragment' and 'bundle must not be fragmented' flags are set"
.to_string(),
));
}
let admin_rec_check = !self.has(BUNDLE_ADMINISTRATIVE_RECORD_PAYLOAD)
|| (!self.has(BUNDLE_STATUS_REQUEST_RECEPTION)
&& !self.has(BUNDLE_STATUS_REQUEST_FORWARD)
&& !self.has(BUNDLE_STATUS_REQUEST_DELIVERY)
&& !self.has(BUNDLE_STATUS_REQUEST_DELETION));
if !admin_rec_check {
errors.push(Bp7Error::BundleControlFlagError(
"\"payload is administrative record => no status report request flags\" failed"
.to_string(),
))
}
if !errors.is_empty() {
return Some(errors);
}
None
}
}
#[derive(Debug, Clone, PartialEq, Builder)]
#[builder(default)]
pub struct Bundle {
pub primary: PrimaryBlock,
pub canonicals: Vec<CanonicalBlock>,
}
impl Serialize for Bundle {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
let mut seq = serializer.serialize_seq(Some(self.canonicals.len() + 1))?;
seq.serialize_element(&self.primary)?;
for e in &self.canonicals {
seq.serialize_element(&e)?;
}
seq.end()
}
}
impl<'de> Deserialize<'de> for Bundle {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
struct BundleVisitor;
impl<'de> Visitor<'de> for BundleVisitor {
type Value = Bundle;
fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
formatter.write_str("packet")
}
fn visit_seq<V>(self, mut seq: V) -> Result<Self::Value, V::Error>
where
V: SeqAccess<'de>,
{
let primary: PrimaryBlock = seq
.next_element()?
.ok_or_else(|| de::Error::invalid_length(0, &self))?;
let mut canonicals: Vec<CanonicalBlock> = Vec::new();
while let Some(next) = seq.next_element::<CanonicalBlock>()? {
canonicals.push(next);
}
Ok(Bundle {
primary,
canonicals,
})
}
}
deserializer.deserialize_any(BundleVisitor)
}
}
impl Default for Bundle {
fn default() -> Self {
Bundle {
primary: PrimaryBlock::new(),
canonicals: Vec::new(),
}
}
}
impl Bundle {
pub fn new(primary: PrimaryBlock, canonicals: Vec<CanonicalBlock>) -> Bundle {
Bundle {
primary,
canonicals,
}
}
pub fn validation_errors(&self) -> Option<Bp7ErrorList> {
let mut errors: Bp7ErrorList = Vec::new();
let mut b_num = std::collections::HashSet::new();
let mut b_types = std::collections::HashSet::new();
if let Some(mut err) = self.primary.validation_errors() {
errors.append(&mut err);
}
for blck in &self.canonicals {
if let Some(mut err) = blck.validation_errors() {
errors.append(&mut err);
}
if (self
.primary
.bundle_control_flags
.has(BUNDLE_ADMINISTRATIVE_RECORD_PAYLOAD)
|| self.primary.source == DTN_NONE)
&& blck.block_control_flags.has(BLOCK_STATUS_REPORT)
{
errors.push(Bp7Error::BundleError(
"Bundle Processing Control Flags indicate that this bundle's payload is an administrative record or the source node is omitted, but the \"Transmit status report if block cannot be processed\" Block Processing Control Flag was set in a Canonical Block".to_string()
));
}
if !b_num.insert(blck.block_number) {
errors.push(Bp7Error::BundleError(
"Block numbers occurred multiple times".to_string(),
));
}
if !b_types.insert(blck.block_type)
&& (blck.block_type == BUNDLE_AGE_BLOCK
|| blck.block_type == HOP_COUNT_BLOCK
|| blck.block_type == PREVIOUS_NODE_BLOCK)
{
errors.push(Bp7Error::BundleError(
"PreviousNode, BundleAge and HopCound blocks must not occure multiple times"
.to_string(),
));
}
}
if self.primary.creation_timestamp.dtntime() == 0 && b_types.contains(&BUNDLE_AGE_BLOCK) {
errors.push(Bp7Error::BundleError(
"Creation Timestamp is zero, but no Bundle Age block is present".to_string(),
));
}
if !errors.is_empty() {
return Some(errors);
}
None
}
pub fn sort_canonicals(&mut self) {
self.canonicals
.sort_by(|a, b| a.block_number.cmp(&b.block_number));
}
fn next_canonical_block_number(&self) -> u64 {
let mut highest_block_number = 0;
for c in self.canonicals.iter() {
highest_block_number = cmp::max(highest_block_number, c.block_number);
}
highest_block_number + 1
}
pub fn add_canonical_block(&mut self, mut cblock: CanonicalBlock) {
if (cblock.block_type == PAYLOAD_BLOCK
|| cblock.block_type == HOP_COUNT_BLOCK
|| cblock.block_type == BUNDLE_AGE_BLOCK
|| cblock.block_type == PREVIOUS_NODE_BLOCK)
&& self.extension_block(cblock.block_type).is_some()
{
return;
}
let mut block_num = self.next_canonical_block_number();
if cblock.block_type == PAYLOAD_BLOCK {
block_num = 0;
} else if block_num == 0 && cblock.block_type != PAYLOAD_BLOCK {
block_num = 1;
}
cblock.block_number = block_num;
self.canonicals.push(cblock);
self.sort_canonicals();
}
pub fn is_administrative_record(&self) -> bool {
self.primary
.bundle_control_flags
.has(BUNDLE_ADMINISTRATIVE_RECORD_PAYLOAD)
}
pub fn payload(&self) -> Option<&ByteBuffer> {
let pb = self.extension_block(crate::canonical::PAYLOAD_BLOCK);
if pb.is_some() {
pb.unwrap().payload_data()
} else {
None
}
}
pub fn set_crc(&mut self, crc_type: CRCType) {
self.primary.set_crc_type(crc_type);
for b in &mut self.canonicals {
b.set_crc_type(crc_type);
}
}
pub fn crc_valid(&mut self) -> bool {
if !self.primary.check_crc() {
return false;
}
for b in &mut self.canonicals {
if !b.check_crc() {
return false;
}
}
true
}
pub fn calculate_crc(&mut self) {
self.primary.update_crc();
for b in &mut self.canonicals {
b.update_crc();
}
}
pub fn extension_block(&self, block_type: CanonicalBlockType) -> Option<(&CanonicalBlock)> {
for b in &self.canonicals {
if b.block_type == block_type && b.extension_validation_error().is_none() {
return Some(b);
}
}
None
}
pub fn extension_block_mut(
&mut self,
block_type: CanonicalBlockType,
) -> Option<(&mut CanonicalBlock)> {
for b in &mut self.canonicals {
if b.block_type == block_type && b.extension_validation_error().is_none() {
return Some(b);
}
}
None
}
pub fn to_cbor(&mut self) -> ByteBuffer {
self.calculate_crc();
let mut bytebuf = serde_cbor::to_vec(&self).expect("Error serializing bundle as cbor.");
bytebuf[0] = 0x9f;
bytebuf.push(0xff);
bytebuf
}
pub fn to_json(&mut self) -> String {
self.calculate_crc();
serde_json::to_string(&self).unwrap()
}
pub fn id(&self) -> String {
let mut id = format!(
"{}-{}-{}-{}",
self.primary.source,
self.primary.creation_timestamp.dtntime(),
self.primary.creation_timestamp.seqno(),
self.primary.destination
);
if self.primary.has_fragmentation() {
id = format!("{}-{}", id, self.primary.fragmentation_offset);
}
id
}
pub fn update_extensions(&mut self, local_node: EndpointID, residence_time: u64) -> bool {
if let Some(hcblock) = self.extension_block_mut(HOP_COUNT_BLOCK) {
hcblock.hop_count_increase();
if hcblock.hop_count_exceeded() {
return false;
}
}
if let Some(pnblock) = self.extension_block_mut(PREVIOUS_NODE_BLOCK) {
pnblock.previous_node_update(local_node);
}
if let Some(bablock) = self.extension_block_mut(BUNDLE_AGE_BLOCK) {
if let Some(ba_orig) = bablock.bundle_age_get() {
bablock.bundle_age_update(ba_orig + residence_time);
if ba_orig + residence_time > self.primary.lifetime * 1000 {
return false;
}
}
}
!self.primary.is_lifetime_exceeded()
}
}
impl TryFrom<ByteBuffer> for Bundle {
type Error = String;
fn try_from(item: ByteBuffer) -> Result<Self, Self::Error> {
match serde_cbor::from_slice(&item) {
Ok(bndl) => Ok(bndl),
Err(err) => Err(format!("Decoding bundle failed: {:?}", err)),
}
}
}
impl TryFrom<String> for Bundle {
type Error = String;
fn try_from(item: String) -> Result<Self, Self::Error> {
match serde_json::from_str(&item) {
Ok(bndl) => Ok(bndl),
Err(err) => Err(format!("Decoding bundle failed: {:?}", err)),
}
}
}
pub fn new_standard_bundle(src: EndpointID, dst: EndpointID, data: ByteBuffer) -> Bundle {
let pblock = crate::primary::PrimaryBlockBuilder::default()
.destination(dst)
.source(src.clone())
.report_to(src)
.creation_timestamp(CreationTimestamp::now())
.lifetime(60 * 60 * 1_000_000)
.build()
.unwrap();
let mut b = crate::bundle::BundleBuilder::default()
.primary(pblock)
.canonicals(vec![
crate::canonical::new_payload_block(0, data),
crate::canonical::new_bundle_age_block(
1,
0,
SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("Time went backwards")
.as_millis() as u64,
),
])
.build()
.unwrap();
b.set_crc(crate::crc::CRC_32);
b
}
pub fn new_std_payload_bundle(src: EndpointID, dst: EndpointID, data: ByteBuffer) -> Bundle {
let now = CreationTimestamp::now();
let pblock = PrimaryBlockBuilder::default()
.destination(dst)
.source(src.clone())
.report_to(src)
.creation_timestamp(now)
.lifetime(60 * 60 * 1_000_000)
.build()
.unwrap();
let mut b = BundleBuilder::default()
.primary(pblock)
.canonicals(vec![
new_payload_block(0, data),
new_bundle_age_block(1, 0, 0),
new_hop_count_block(2, 0, 32),
])
.build()
.unwrap();
b.set_crc(CRC_NO);
b.sort_canonicals();
b
}