#![cfg(feature = "rrdp")]
use std::{error, fmt, hash, io, str};
use std::io::Read;
use std::ops::Deref;
use bytes::Bytes;
use log::info;
use ring::digest;
use uuid::Uuid;
use crate::{uri, xml};
use crate::util::base64;
use crate::xml::decode::{Content, Error as XmlError, Reader, Name};
#[cfg(feature = "serde")] use std::str::FromStr;
#[cfg(feature = "serde")] use serde::{
Deserialize, Deserializer, Serialize, Serializer
};
const MAX_FILE_SIZE: u64 = 100_000_000;
const MAX_HEADER_SIZE: u64 = 1_000_000;
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct NotificationFile {
session_id: Uuid,
serial: u64,
snapshot: SnapshotInfo,
deltas: Result<Vec<DeltaInfo>, DeltaListError>,
}
impl NotificationFile {
pub fn new(
session_id: Uuid,
serial: u64,
snapshot: UriAndHash,
deltas: Vec<DeltaInfo>,
) -> Self {
NotificationFile {
session_id,
serial,
snapshot,
deltas: Ok(deltas),
}
}
pub fn session_id(&self) -> Uuid {
self.session_id
}
pub fn serial(&self) -> u64 {
self.serial
}
pub fn snapshot(&self) -> &SnapshotInfo {
&self.snapshot
}
pub fn deltas(&self) -> &[DeltaInfo] {
match self.deltas {
Ok(ref deltas) => deltas.as_slice(),
Err(_) => &[]
}
}
pub fn delta_status(&self) -> Result<(), DeltaListError> {
match self.deltas {
Ok(_) => Ok(()),
Err(err) => Err(err)
}
}
pub fn sort_deltas(&mut self) {
if let Ok(ref mut deltas) = self.deltas {
deltas.sort_by_key(|delta| delta.serial());
}
}
pub fn reverse_sort_deltas(&mut self) {
if let Ok(ref mut deltas) = self.deltas {
deltas.sort_by_key(|b| std::cmp::Reverse(b.serial));
}
}
pub fn sort_and_verify_deltas(&mut self, limit: Option<usize>) -> bool {
if let Ok(ref mut deltas) = self.deltas {
if !deltas.is_empty() {
deltas.sort_by_key(|delta| delta.serial());
if let Some(limit) = limit {
if limit < deltas.len() {
let offset = deltas.len() - limit;
deltas.drain(..offset);
}
}
let mut last_seen = deltas[0].serial();
for delta in &deltas[1..] {
if last_seen + 1 != delta.serial() {
return false;
} else {
last_seen = delta.serial()
}
}
}
}
true
}
pub fn has_matching_origins(&self, base: &uri::Https) -> bool {
if !base.eq_authority(self.snapshot().uri()) {
return false
}
if let Ok(ref deltas) = self.deltas {
if deltas.iter().any(|delta| !base.eq_authority(delta.uri())) {
return false
}
}
true
}
}
impl NotificationFile {
pub fn parse<R: io::BufRead>(
reader: R,
) -> Result<Self, XmlError> {
Self::_parse(reader, None)
}
pub fn parse_limited<R: io::BufRead>(
reader: R,
delta_limit: usize,
) -> Result<Self, XmlError> {
Self::_parse(reader, Some(delta_limit))
}
pub fn _parse<R: io::BufRead>(
reader: R,
delta_limit: Option<usize>,
) -> Result<Self, XmlError> {
let mut reader = Reader::new(reader);
let mut session_id = None;
let mut serial = None;
let mut outer = reader.start_with_limit(
|element| {
if element.name() != NOTIFICATION {
return Err(XmlError::Malformed)
}
element.attributes(|name, value| {
match name {
b"version" => {
if value.ascii_into::<u8>()? != 1 {
return Err(XmlError::Malformed)
}
Ok(())
}
b"session_id" => {
session_id = Some(value.ascii_into()?);
Ok(())
}
b"serial" => {
serial = Some(value.ascii_into()?);
Ok(())
}
_ => Err(XmlError::Malformed)
}
})
},
MAX_HEADER_SIZE
)?;
let mut snapshot = None;
let mut deltas = Ok(vec![]);
while let Some(mut content) = outer.take_opt_element_with_limit(
&mut reader,
|element| {
match element.name() {
SNAPSHOT => {
if snapshot.is_some() {
return Err(XmlError::Malformed)
}
let mut uri = None;
let mut hash = None;
element.attributes(|name, value| match name {
b"uri" => {
uri = Some(value.ascii_into()?);
Ok(())
}
b"hash" => {
hash = Some(value.ascii_into()?);
Ok(())
}
_ => Err(XmlError::Malformed)
})?;
match (uri, hash) {
(Some(uri), Some(hash)) => {
snapshot = Some(UriAndHash::new(uri, hash));
Ok(())
}
_ => Err(XmlError::Malformed)
}
}
DELTA => {
let mut serial = None;
let mut uri = None;
let mut hash = None;
element.attributes(|name, value| match name {
b"serial" => {
serial = Some(value.ascii_into()?);
Ok(())
}
b"uri" => {
uri = Some(value.ascii_into()?);
Ok(())
}
b"hash" => {
hash = Some(value.ascii_into()?);
Ok(())
}
_ => Err(XmlError::Malformed)
})?;
let (serial, uri, hash) = match (serial, uri, hash) {
(Some(serial), Some(uri), Some(hash)) => {
(serial, uri, hash)
}
_ => return Err(XmlError::Malformed)
};
if let Some(limit) = delta_limit {
let len = deltas.as_ref().map(|deltas| {
deltas.len()
}).unwrap_or(0);
if len >= limit {
deltas = Err(DeltaListError::Oversized);
}
}
if let Ok(ref mut deltas) = deltas {
deltas.push(DeltaInfo::new(serial, uri, hash))
}
Ok(())
}
_ => Err(XmlError::Malformed)
}
},
MAX_HEADER_SIZE
)? {
content.take_end(&mut reader)?;
}
outer.take_end(&mut reader)?;
reader.end()?;
match (session_id, serial, snapshot) {
(Some(session_id), Some(serial), Some(snapshot)) => {
Ok(NotificationFile { session_id, serial, snapshot, deltas })
}
_ => Err(XmlError::Malformed)
}
}
pub fn write_xml(
&self, writer: &mut impl io::Write
) -> Result<(), io::Error> {
let mut writer = xml::encode::Writer::new(writer);
writer.element(NOTIFICATION.into_unqualified())?
.attr("xmlns", NS)?
.attr("version", "1")?
.attr("session_id", &self.session_id)?
.attr("serial", &self.serial)?
.content(|content| {
content.element(SNAPSHOT.into_unqualified())?
.attr("uri", self.snapshot.uri())?
.attr("hash", &self.snapshot.hash())?
;
for delta in self.deltas() {
content.element(DELTA.into_unqualified())?
.attr("serial", &delta.serial())?
.attr("uri", delta.uri())?
.attr("hash", &delta.hash())?
;
}
Ok(())
})?;
writer.done()
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PublishElement {
uri: uri::Rsync,
data: Bytes,
}
impl PublishElement {
pub fn new(
uri: uri::Rsync,
data: Bytes,
) -> Self {
PublishElement { uri, data }
}
pub fn uri(&self) -> &uri::Rsync {
&self.uri
}
pub fn data(&self) -> &Bytes {
&self.data
}
pub fn unpack(self) -> (uri::Rsync, Bytes) {
(self.uri, self.data)
}
fn write_xml(
&self,
content: &mut xml::encode::Content<impl io::Write>
) -> Result<(), io::Error> {
content.element(PUBLISH.into_unqualified())?
.attr("uri", &self.uri)?
.content(|content| {
content.base64(self.data.as_ref())
})?;
Ok(())
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct UpdateElement {
uri: uri::Rsync,
hash: Hash,
data: Bytes,
}
impl UpdateElement {
pub fn new(uri: uri::Rsync, hash: Hash, data: Bytes) -> Self {
UpdateElement { uri, hash, data }
}
pub fn uri(&self) -> &uri::Rsync {
&self.uri
}
pub fn hash(&self) -> &Hash {
&self.hash
}
pub fn data(&self) -> &Bytes {
&self.data
}
pub fn unpack(self) -> (uri::Rsync, Hash, Bytes) {
(self.uri, self.hash, self.data)
}
}
impl UpdateElement {
fn write_xml(
&self,
content: &mut xml::encode::Content<impl io::Write>
) -> Result<(), io::Error> {
content.element(PUBLISH.into_unqualified())?
.attr("uri", &self.uri)?
.attr("hash", &self.hash)?
.content(|content| {
content.base64(self.data.as_ref())
})?;
Ok(())
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct WithdrawElement {
uri: uri::Rsync,
hash: Hash,
}
impl WithdrawElement {
pub fn new(uri: uri::Rsync, hash: Hash) -> Self {
WithdrawElement { uri, hash }
}
pub fn uri(&self) -> &uri::Rsync {
&self.uri
}
pub fn hash(&self) -> &Hash {
&self.hash
}
pub fn unpack(self) -> (uri::Rsync, Hash) {
(self.uri, self.hash)
}
}
impl WithdrawElement {
fn write_xml(
&self,
content: &mut xml::encode::Content<impl io::Write>
) -> Result<(), io::Error> {
content.element(WITHDRAW.into_unqualified())?
.attr("uri", &self.uri)?
.attr("hash", &self.hash)?;
Ok(())
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum DeltaElement {
Publish(PublishElement),
Update(UpdateElement),
Withdraw(WithdrawElement)
}
impl DeltaElement {
fn write_xml(
&self,
content: &mut xml::encode::Content<impl io::Write>
) -> Result<(), io::Error> {
match self {
DeltaElement::Publish(p) => p.write_xml(content),
DeltaElement::Update(u) => u.write_xml(content),
DeltaElement::Withdraw(w) => w.write_xml(content)
}
}
}
impl From<PublishElement> for DeltaElement {
fn from(src: PublishElement) -> Self {
DeltaElement::Publish(src)
}
}
impl From<UpdateElement> for DeltaElement {
fn from(src: UpdateElement) -> Self {
DeltaElement::Update(src)
}
}
impl From<WithdrawElement> for DeltaElement {
fn from(src: WithdrawElement) -> Self {
DeltaElement::Withdraw(src)
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Snapshot {
session_id: Uuid,
serial: u64,
elements: Vec<PublishElement>,
}
impl Snapshot {
pub fn new(
session_id: Uuid,
serial: u64,
elements: Vec<PublishElement>,
) -> Self {
Snapshot { session_id, serial, elements }
}
pub fn session_id(&self) -> Uuid {
self.session_id
}
pub fn serial(&self) -> u64 {
self.serial
}
pub fn elements(&self) -> &[PublishElement] {
&self.elements
}
pub fn into_elements(self) -> Vec<PublishElement> {
self.elements
}
}
impl Snapshot {
pub fn parse<R: io::BufRead>(
reader: R
) -> Result<Self, ProcessError> {
let mut builder = SnapshotBuilder {
session_id: None,
serial: None,
elements: vec![]
};
builder.process(reader)?;
builder.try_into()
}
pub fn write_xml(
&self, writer: &mut impl io::Write
) -> Result<(), io::Error> {
let mut writer = xml::encode::Writer::new(writer);
writer.element(SNAPSHOT.into_unqualified())?
.attr("xmlns", NS)?
.attr("version", "1")?
.attr("session_id", &self.session_id)?
.attr("serial", &self.serial)?
.content(|content| {
for el in &self.elements {
el.write_xml(content)?;
}
Ok(())
})?;
writer.done()
}
}
struct SnapshotBuilder {
session_id: Option<Uuid>,
serial: Option<u64>,
elements: Vec<PublishElement>,
}
impl ProcessSnapshot for SnapshotBuilder {
type Err = ProcessError;
fn meta(&mut self, session_id: Uuid, serial: u64) -> Result<(), Self::Err> {
self.session_id = Some(session_id);
self.serial = Some(serial);
Ok(())
}
fn publish(&mut self, uri: uri::Rsync, data: &mut ObjectReader) -> Result<(), Self::Err> {
let mut buf = Vec::new();
data.read_to_end(&mut buf)?;
let data = Bytes::from(buf);
let element = PublishElement { uri, data };
self.elements.push(element);
Ok(())
}
}
impl TryFrom<SnapshotBuilder> for Snapshot {
type Error = ProcessError;
fn try_from(builder: SnapshotBuilder) -> Result<Self, Self::Error> {
let session_id = builder.session_id.ok_or(
ProcessError::Xml(XmlError::Malformed)
)?;
let serial = builder.serial.ok_or(
ProcessError::Xml(XmlError::Malformed)
)?;
Ok(Snapshot { session_id, serial, elements: builder.elements })
}
}
pub trait ProcessSnapshot {
type Err: From<ProcessError>;
fn meta(
&mut self,
session_id: Uuid,
serial: u64,
) -> Result<(), Self::Err>;
fn publish(
&mut self,
uri: uri::Rsync,
data: &mut ObjectReader,
) -> Result<(), Self::Err>;
fn process<R: io::BufRead>(
&mut self,
reader: R
) -> Result<(), Self::Err> {
let mut reader = Reader::new(reader);
let mut session_id = None;
let mut serial = None;
let mut outer = reader.start_with_limit(
|element| {
if element.name() != SNAPSHOT {
info!("Bad outer: not snapshot, but {:?}", element.name());
return Err(XmlError::Malformed)
}
element.attributes(|name, value| match name {
b"version" => {
if value.ascii_into::<u8>()? != 1 {
info!("Bad version");
return Err(XmlError::Malformed)
}
Ok(())
}
b"session_id" => {
session_id = Some(value.ascii_into()?);
Ok(())
}
b"serial" => {
serial = Some(value.ascii_into()?);
Ok(())
}
_ => {
info!("Bad attribute on snapshot.");
Err(XmlError::Malformed)
}
})
},
MAX_HEADER_SIZE,
).map_err(Into::into)?;
match (session_id, serial) {
(Some(session_id), Some(serial)) => {
self.meta(session_id, serial)?;
}
_ => {
info!("Missing session or serial");
return Err(ProcessError::malformed().into())
}
}
loop {
let mut uri = None;
let inner = outer.take_opt_element_with_limit(
&mut reader,
|element| {
if element.name() != PUBLISH {
info!("Bad inner: not publish");
return Err(ProcessError::malformed())
}
element.attributes(|name, value| match name {
b"uri" => {
uri = Some(value.ascii_into()?);
Ok(())
}
_ => {
info!("Bad attribute on publish.");
Err(ProcessError::malformed())
}
})
},
MAX_FILE_SIZE,
)?;
let mut inner = match inner {
Some(inner) => inner,
None => break
};
let uri = match uri {
Some(uri) => uri,
None => return Err(ProcessError::malformed().into())
};
ObjectReader::process(&mut inner, &mut reader, |reader| {
self.publish(uri, reader)
})?;
}
outer.take_end(&mut reader).map_err(Into::into)?;
reader.end().map_err(Into::into)?;
Ok(())
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Delta {
session_id: Uuid,
serial: u64,
elements: Vec<DeltaElement>
}
impl Delta {
pub fn new(
session_id: Uuid,
serial: u64,
elements: Vec<DeltaElement>
) -> Self {
Delta { session_id, serial, elements }
}
pub fn session_id(&self) -> Uuid {
self.session_id
}
pub fn serial(&self) -> u64 {
self.serial
}
pub fn elements(&self) -> &[DeltaElement] {
&self.elements
}
pub fn into_elements(self) -> Vec<DeltaElement> {
self.elements
}
}
impl Delta {
pub fn parse<R: io::BufRead>(
reader: R
) -> Result<Self, ProcessError> {
let mut builder = DeltaBuilder {
session_id: None,
serial: None,
elements: vec![]
};
builder.process(reader)?;
builder.try_into()
}
pub fn write_xml(
&self, writer: &mut impl io::Write
) -> Result<(), io::Error> {
let mut writer = xml::encode::Writer::new(writer);
writer.element(DELTA.into_unqualified())?
.attr("xmlns", NS)?
.attr("version", "1")?
.attr("session_id", &self.session_id)?
.attr("serial", &self.serial)?
.content(|content| {
for el in &self.elements {
el.write_xml(content)?;
}
Ok(())
})?;
writer.done()
}
}
struct DeltaBuilder {
session_id: Option<Uuid>,
serial: Option<u64>,
elements: Vec<DeltaElement>
}
impl ProcessDelta for DeltaBuilder {
type Err = ProcessError;
fn meta(
&mut self,
session_id: Uuid,
serial: u64,
) -> Result<(), Self::Err> {
self.session_id = Some(session_id);
self.serial = Some(serial);
Ok(())
}
fn publish(
&mut self,
uri: uri::Rsync,
hash_opt: Option<Hash>,
data: &mut ObjectReader,
) -> Result<(), Self::Err> {
let mut buf = Vec::new();
data.read_to_end(&mut buf)?;
let data = Bytes::from(buf);
match hash_opt {
Some(hash) => {
let update = UpdateElement { uri, hash, data};
self.elements.push(DeltaElement::Update(update));
},
None => {
let publish = PublishElement { uri, data};
self.elements.push(DeltaElement::Publish(publish));
}
}
Ok(())
}
fn withdraw(
&mut self,
uri: uri::Rsync,
hash: Hash,
) -> Result<(), Self::Err> {
let withdraw = WithdrawElement { uri, hash };
self.elements.push(DeltaElement::Withdraw(withdraw));
Ok(())
}
}
impl TryFrom<DeltaBuilder> for Delta {
type Error = ProcessError;
fn try_from(builder: DeltaBuilder) -> Result<Self, Self::Error> {
let session_id = builder.session_id.ok_or(
ProcessError::Xml(XmlError::Malformed)
)?;
let serial = builder.serial.ok_or(
ProcessError::Xml(XmlError::Malformed)
)?;
Ok(Delta { session_id, serial, elements: builder.elements })
}
}
pub trait ProcessDelta {
type Err: From<ProcessError>;
fn meta(
&mut self,
session_id: Uuid,
serial: u64,
) -> Result<(), Self::Err>;
fn publish(
&mut self,
uri: uri::Rsync,
hash: Option<Hash>,
data: &mut ObjectReader,
) -> Result<(), Self::Err>;
fn withdraw(
&mut self,
uri: uri::Rsync,
hash: Hash,
) -> Result<(), Self::Err>;
fn process<R: io::BufRead>(
&mut self,
reader: R
) -> Result<(), Self::Err> {
let mut reader = Reader::new(reader);
let mut session_id = None;
let mut serial = None;
let mut outer = reader.start_with_limit(
|element| {
if element.name() != DELTA {
return Err(ProcessError::malformed())
}
element.attributes(|name, value| match name {
b"version" => {
if value.ascii_into::<u8>()? != 1 {
return Err(ProcessError::malformed())
}
Ok(())
}
b"session_id" => {
session_id = Some(value.ascii_into()?);
Ok(())
}
b"serial" => {
serial = Some(value.ascii_into()?);
Ok(())
}
_ => Err(ProcessError::malformed())
})
},
MAX_HEADER_SIZE,
)?;
match (session_id, serial) {
(Some(session_id), Some(serial)) => {
self.meta(session_id, serial)?;
}
_ => return Err(ProcessError::malformed().into())
}
loop {
let mut action = None;
let mut uri = None;
let mut hash = None;
let inner = outer.take_opt_element_with_limit(
&mut reader,
|element| {
match element.name() {
PUBLISH => action = Some(Action::Publish),
WITHDRAW => action = Some(Action::Withdraw),
_ => return Err(ProcessError::malformed()),
};
element.attributes(|name, value| match name {
b"uri" => {
uri = Some(value.ascii_into()?);
Ok(())
}
b"hash" => {
hash = Some(value.ascii_into()?);
Ok(())
}
_ => Err(ProcessError::malformed())
})
},
MAX_FILE_SIZE,
)?;
let mut inner = match inner {
Some(inner) => inner,
None => break
};
let uri = match uri {
Some(uri) => uri,
None => return Err(ProcessError::malformed().into())
};
match action.unwrap() { Action::Publish => {
ObjectReader::process(
&mut inner, &mut reader,
|reader| self.publish(uri, hash, reader)
)?;
}
Action::Withdraw => {
let hash = match hash {
Some(hash) => hash,
None => return Err(ProcessError::malformed().into())
};
self.withdraw(uri, hash)?;
inner.take_end(&mut reader).map_err(Into::into)?;
}
}
}
outer.take_end(&mut reader).map_err(Into::into)?;
reader.end().map_err(Into::into)?;
Ok(())
}
}
pub type SnapshotInfo = UriAndHash;
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct DeltaInfo {
serial: u64,
uri_and_hash: UriAndHash
}
impl DeltaInfo {
pub fn new(serial: u64, uri: uri::Https, hash: Hash) -> Self {
DeltaInfo {
serial,
uri_and_hash: UriAndHash::new(uri, hash)
}
}
pub fn serial(&self) -> u64 {
self.serial
}
}
impl Deref for DeltaInfo {
type Target = UriAndHash;
fn deref(&self) -> &Self::Target {
&self.uri_and_hash
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct UriAndHash {
uri: uri::Https,
hash: Hash,
}
impl UriAndHash {
pub fn new(uri: uri::Https, hash: Hash) -> Self {
UriAndHash { uri, hash }
}
pub fn uri(&self) -> &uri::Https {
&self.uri
}
pub fn hash(&self) -> Hash {
self.hash
}
pub fn into_uri(self) -> uri::Https {
self.uri
}
pub fn into_pair(self) -> (uri::Https, Hash) {
(self.uri, self.hash)
}
}
#[derive(Clone, Copy, Eq, hash::Hash, PartialEq)]
#[repr(transparent)] pub struct Hash([u8; 32]);
impl Hash {
pub fn as_slice(&self) -> &[u8] {
self.0.as_ref()
}
pub fn from_data(data: &[u8]) -> Self {
let digest = digest::digest(&digest::SHA256, data);
Self::try_from(digest.as_ref()).unwrap()
}
pub fn matches(&self, data: &[u8]) -> bool {
let data_hash = Self::from_data(data);
*self == data_hash
}
}
impl From<[u8;32]> for Hash {
fn from(value: [u8;32]) -> Hash {
Hash(value)
}
}
impl From<Hash> for [u8; 32] {
fn from(src: Hash) -> Self {
src.0
}
}
impl<'a> TryFrom<&'a [u8]> for Hash {
type Error = std::array::TryFromSliceError;
fn try_from(src: &'a [u8]) -> Result<Self, Self::Error> {
TryFrom::try_from(src).map(Hash)
}
}
impl TryFrom<digest::Digest> for Hash {
type Error = AlgorithmError;
fn try_from(digest: digest::Digest) -> Result<Self, Self::Error> {
TryFrom::try_from(
digest.as_ref()
).map(Hash).map_err(|_| AlgorithmError(()))
}
}
impl str::FromStr for Hash {
type Err = ParseHashError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
if s.len() != 64 {
return Err(ParseHashError::BAD_LENGTH)
}
let mut res = [0u8; 32];
let mut s = s.chars();
for octet in &mut res {
let first = s.next().ok_or(
ParseHashError::BAD_LENGTH
)?.to_digit(16).ok_or(
ParseHashError::BAD_CHARS
)?;
let second = s.next().ok_or(
ParseHashError::BAD_LENGTH
)?.to_digit(16).ok_or(
ParseHashError::BAD_CHARS
)?;
*octet = ((first << 4) | second) as u8;
}
Ok(Hash(res))
}
}
impl AsRef<[u8]> for Hash {
fn as_ref(&self) -> &[u8] {
self.0.as_ref()
}
}
impl PartialEq<digest::Digest> for Hash {
fn eq(&self, other: &digest::Digest) -> bool {
self.0.as_ref() == other.as_ref()
}
}
impl fmt::Display for Hash {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
for &ch in self.as_slice() {
write!(f, "{ch:02x}")?;
}
Ok(())
}
}
impl fmt::Debug for Hash {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "Hash({self})")
}
}
#[cfg(feature = "serde")]
impl Serialize for Hash {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where S: Serializer {
self.to_string().serialize(serializer)
}
}
#[cfg(feature = "serde")]
impl<'de> Deserialize<'de> for Hash {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where D: Deserializer<'de> {
let hex_str = String::deserialize(deserializer)?;
Hash::from_str(&hex_str).map_err(serde::de::Error::custom)
}
}
enum Action {
Publish,
Withdraw,
}
pub struct ObjectReader<'a>(
base64::XmlDecoderReader<'a>
);
impl ObjectReader<'_> {
fn process<R, T, E, F> (
content: &mut Content,
reader: &mut Reader<R>,
op: F
) -> Result<T, E>
where
R: io::BufRead,
E: From<ProcessError>,
F: FnOnce(&mut ObjectReader) -> Result<T, E>
{
enum Error<E> {
Xml(XmlError),
User(E),
}
impl<E> From<XmlError> for Error<E> {
fn from(err: XmlError) -> Self {
Error::Xml(err)
}
}
content.take_opt_final_text(reader, |text| {
let b64 = match text.as_ref() {
Some(text) => text.to_ascii()?,
None => Default::default(),
};
op(
&mut ObjectReader(base64::Xml.decode_reader(b64.as_ref()))
).map_err(Error::User)
}).map_err(|err| match err {
Error::Xml(err) => ProcessError::Xml(err).into(),
Error::User(err) => err
})
}
}
impl io::Read for ObjectReader<'_> {
fn read(&mut self, buf: &mut [u8]) -> Result<usize, io::Error> {
self.0.read(buf)
}
}
const NS: &[u8] = b"http://www.ripe.net/rpki/rrdp";
const NOTIFICATION: Name = Name::qualified(NS, b"notification");
const SNAPSHOT: Name = Name::qualified(NS, b"snapshot");
const DELTA: Name = Name::qualified(NS, b"delta");
const PUBLISH: Name = Name::qualified(NS, b"publish");
const WITHDRAW: Name = Name::qualified(NS, b"withdraw");
#[derive(Clone, Copy, Debug)]
pub struct AlgorithmError(());
impl fmt::Display for AlgorithmError {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.write_str("algorithm mismatch")
}
}
impl error::Error for AlgorithmError { }
#[derive(Clone, Copy, Debug)]
pub struct ParseHashError(&'static str);
impl ParseHashError {
const BAD_LENGTH: Self = ParseHashError("invalid length");
const BAD_CHARS: Self = ParseHashError("invalid characters");
}
impl fmt::Display for ParseHashError {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.write_str(self.0)
}
}
impl error::Error for ParseHashError { }
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DeltaListError {
Oversized,
}
impl fmt::Display for DeltaListError {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
Self::Oversized => write!(f, "excessively large delta list")
}
}
}
impl error::Error for DeltaListError { }
#[derive(Debug)]
pub enum ProcessError {
Io(io::Error),
Xml(XmlError),
}
impl ProcessError {
fn malformed() -> Self {
ProcessError::Xml(XmlError::Malformed)
}
}
impl From<io::Error> for ProcessError {
fn from(err: io::Error) -> Self {
ProcessError::Io(err)
}
}
impl From<XmlError> for ProcessError {
fn from(err: XmlError) -> Self {
ProcessError::Xml(err)
}
}
impl fmt::Display for ProcessError {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
ProcessError::Io(ref inner) => inner.fmt(f),
ProcessError::Xml(ref inner) => inner.fmt(f),
}
}
}
impl error::Error for ProcessError {
fn source(&self) -> Option<&(dyn error::Error + 'static)> {
match self {
ProcessError::Io(error) => Some(error),
ProcessError::Xml(error) => Some(error),
}
}
}
#[cfg(test)]
mod test {
use std::str::from_utf8_unchecked;
use std::str::FromStr;
use super::*;
pub struct Test;
impl ProcessSnapshot for Test {
type Err = ProcessError;
fn meta(
&mut self,
_session_id: Uuid,
_serial: u64,
) -> Result<(), Self::Err> {
Ok(())
}
fn publish(
&mut self,
_uri: uri::Rsync,
_data: &mut ObjectReader,
) -> Result<(), Self::Err> {
Ok(())
}
}
impl ProcessDelta for Test {
type Err = ProcessError;
fn meta(
&mut self,
_session_id: Uuid,
_serial: u64,
) -> Result<(), Self::Err> {
Ok(())
}
fn publish(
&mut self,
_uri: uri::Rsync,
_hash: Option<Hash>,
_data: &mut ObjectReader,
) -> Result<(), Self::Err> {
Ok(())
}
fn withdraw(
&mut self,
_uri: uri::Rsync,
_hash: Hash,
) -> Result<(), Self::Err> {
Ok(())
}
}
#[test]
fn ripe_notification() {
NotificationFile::parse(
include_bytes!(
"../test-data/rrdp/ripe-notification.xml"
).as_ref()
).unwrap();
}
#[test]
fn lolz_notification() {
assert!(
NotificationFile::parse(
include_bytes!(
"../test-data/rrdp/lolz-notification.xml"
).as_ref()
).is_err()
);
}
#[test]
fn gaps_notification() {
let mut notification_without_gaps = NotificationFile::parse(
include_bytes!("../test-data/rrdp/ripe-notification.xml").as_ref()
).unwrap();
assert!(notification_without_gaps.sort_and_verify_deltas(None));
let mut notification_with_gaps = NotificationFile::parse(
include_bytes!(
"../test-data/rrdp/ripe-notification-with-gaps.xml"
).as_ref()
).unwrap();
assert!(!notification_with_gaps.sort_and_verify_deltas(None));
}
#[test]
fn limit_notification_deltas() {
let mut notification_without_gaps = NotificationFile::parse(
include_bytes!("../test-data/rrdp/ripe-notification.xml").as_ref()
).unwrap();
assert!(notification_without_gaps.sort_and_verify_deltas(Some(2)));
assert_eq!(2, notification_without_gaps.deltas().len());
assert_eq!(
notification_without_gaps.deltas().first().unwrap().serial(),
notification_without_gaps.serial() - 1
);
assert_eq!(
notification_without_gaps.deltas().last().unwrap().serial(),
notification_without_gaps.serial()
);
}
#[test]
fn unsorted_notification() {
let mut from_sorted = NotificationFile::parse(
include_bytes!("../test-data/rrdp/ripe-notification.xml").as_ref()
).unwrap();
let mut from_unsorted = NotificationFile::parse(
include_bytes!(
"../test-data/rrdp/ripe-notification-unsorted.xml"
).as_ref()
).unwrap();
assert_ne!(from_sorted, from_unsorted);
from_unsorted.reverse_sort_deltas();
assert_eq!(from_sorted, from_unsorted);
from_unsorted.sort_deltas();
assert_ne!(from_sorted, from_unsorted);
from_sorted.sort_deltas();
assert_eq!(from_sorted, from_unsorted);
}
#[test]
fn notification_parse_limited() {
let bytes = include_bytes!(
"../test-data/rrdp/ripe-notification.xml"
).as_ref();
let full = NotificationFile::parse(bytes).unwrap();
assert!(!full.deltas().is_empty());
let limited = NotificationFile::parse_limited(
bytes, full.deltas().len()
).unwrap();
assert_eq!(limited.deltas().len(), full.deltas().len());
assert!(limited.delta_status().is_ok());
let limited = NotificationFile::parse_limited(
bytes, full.deltas().len() - 1
).unwrap();
assert_eq!(limited.deltas().len(), 0);
assert!(limited.delta_status().is_err());
}
#[test]
fn ripe_snapshot() {
<Test as ProcessSnapshot>::process(
&mut Test,
include_bytes!("../test-data/rrdp/ripe-snapshot.xml").as_ref()
).unwrap();
}
#[test]
fn ripe_delta() {
<Test as ProcessDelta>::process(
&mut Test,
include_bytes!("../test-data/rrdp/ripe-delta.xml").as_ref()
).unwrap();
}
#[test]
fn hash_to_hash() {
use std::str::FromStr;
let string = "this is a test";
let sha256 =
"2e99758548972a8e8822ad47fa1017ff72f06f3ff6a016851f45c398732bc50c";
let hash = Hash::from_str(sha256).unwrap();
let hash_from_data = Hash::from_data(string.as_bytes());
assert_eq!(hash, hash_from_data);
assert!(hash.matches(string.as_bytes()));
}
#[test]
fn notification_from_to_xml() {
let notification = NotificationFile::parse(
include_bytes!("../test-data/rrdp/ripe-notification.xml").as_ref()
).unwrap();
let mut vec = vec![];
notification.write_xml(&mut vec).unwrap();
let xml = unsafe {
from_utf8_unchecked(vec.as_ref())
};
let notification_parsed = NotificationFile::parse(xml.as_bytes()).unwrap();
assert_eq!(notification, notification_parsed);
}
#[test]
fn snapshot_from_to_xml() {
let data = include_bytes!("../test-data/rrdp/ripe-snapshot.xml");
let snapshot = Snapshot::parse(data.as_ref()).unwrap();
let mut vec = vec![];
snapshot.write_xml(&mut vec).unwrap();
let xml = unsafe {
from_utf8_unchecked(vec.as_ref())
};
let snapshot_parsed = Snapshot::parse(xml.as_bytes()).unwrap();
assert_eq!(snapshot, snapshot_parsed);
}
#[test]
fn delta_from_to_xml() {
let data = include_bytes!("../test-data/rrdp/ripe-delta.xml");
let delta = Delta::parse(data.as_ref()).unwrap();
let mut vec = vec![];
delta.write_xml(&mut vec).unwrap();
let xml = unsafe {
from_utf8_unchecked(vec.as_ref())
};
let delta_parsed = Delta::parse(xml.as_bytes()).unwrap();
assert_eq!(delta, delta_parsed);
}
#[test]
fn snapshot_content() {
const CONTENT: &[u8] = b"foo bar\n";
let snapshot = br#"
<snapshot version="1"
session_id="a2d845c4-5b91-4015-a2b7-988c03ce232a"
serial="1742"
xmlns="http://www.ripe.net/rpki/rrdp"
>
<publish
uri="rsync://example.com/some/path"
>
Zm9vIGJhcgo=
</publish>
<publish
uri="rsync://example.com/some/other"
/>
<publish
uri="rsync://example.com/some/third"
></publish>
</snapshot>
"#;
let snapshot = Snapshot::parse(&mut snapshot.as_ref()).unwrap();
assert_eq!(snapshot.elements.len(), 3);
assert_eq!(
snapshot.elements[0],
PublishElement::new(
uri::Rsync::from_str("rsync://example.com/some/path").unwrap(),
Bytes::copy_from_slice(CONTENT)
)
);
assert_eq!(
snapshot.elements[1],
PublishElement::new(
uri::Rsync::from_str("rsync://example.com/some/other").unwrap(),
Bytes::new()
)
);
assert_eq!(
snapshot.elements[2],
PublishElement::new(
uri::Rsync::from_str("rsync://example.com/some/third").unwrap(),
Bytes::new()
)
);
}
#[test]
fn has_matching_origins() {
fn check<const N: usize>(
snapshot: &str, deltas: [&str; N]
) -> bool {
let hash = Hash::from_data(b"12");
NotificationFile::new(
Uuid::nil(), 0,
SnapshotInfo::new(
uri::Https::from_str(snapshot).unwrap(), hash
),
deltas.iter().map(|uri| {
DeltaInfo::new(
0, uri::Https::from_str(uri).unwrap(), hash
)
}).collect()
).has_matching_origins(
&uri::Https::from_str("https://foo.bar/n/o").unwrap()
)
}
assert!(check(
"https://foo.bar/1/2/3",
["https://foo.bar/", "https://foo.bar/4", "https://foo.bar/7/8"],
));
assert!(!check(
"https://foo.bar/1/2/3",
["https://foo.bar/", "https://foo.local/4"],
));
}
#[test]
fn doctype_panic_regression() {
let reader = std::io::BufReader::with_capacity(
8, "<!d[<>23456789".as_bytes()
);
let _ = NotificationFile::parse(reader);
}
}