use std::fmt;
use chrono::Duration;
use tokio::sync::RwLock;
use rpki::{repository::resources::ResourceSet, repository::x509::Time};
use crate::{
commons::{
api::{AsNumber, ConfiguredRoa, RoaPayload},
bgp::{
make_roa_tree, make_validated_announcement_tree, Announcement, AnnouncementValidity, Announcements,
BgpAnalysisEntry, BgpAnalysisReport, BgpAnalysisState, BgpAnalysisSuggestion, IpRange, RisDumpError,
RisDumpLoader, ValidatedAnnouncement,
},
},
constants::{test_announcements_enabled, BGP_RIS_REFRESH_MINUTES},
};
pub struct BgpAnalyser {
dump_loader: Option<RisDumpLoader>,
seen: RwLock<Announcements>,
}
impl BgpAnalyser {
pub fn new(ris_enabled: bool, ris_v4_uri: &str, ris_v6_uri: &str) -> Self {
if test_announcements_enabled() {
Self::with_test_announcements()
} else {
let dump_loader = if ris_enabled {
Some(RisDumpLoader::new(ris_v4_uri, ris_v6_uri))
} else {
None
};
BgpAnalyser {
dump_loader,
seen: RwLock::new(Announcements::default()),
}
}
}
pub async fn update(&self) -> Result<bool, BgpAnalyserError> {
if let Some(loader) = &self.dump_loader {
let mut seen = self.seen.write().await;
if let Some(last_time) = seen.last_checked() {
if (last_time + Duration::minutes(BGP_RIS_REFRESH_MINUTES)) > Time::now() {
trace!("Will not check BGP Ris Dumps until the refresh interval has passed");
return Ok(false); }
}
let announcements = loader.download_updates().await?;
if seen.equivalent(&announcements) {
debug!("BGP Ris Dumps unchanged");
seen.update_checked();
Ok(false)
} else {
info!("Updated announcements ({}) based on BGP Ris Dumps", announcements.len());
seen.update(announcements);
Ok(true)
}
} else {
Ok(false)
}
}
pub async fn analyse(
&self,
roas: &[ConfiguredRoa],
resources_held: &ResourceSet,
limited_scope: Option<ResourceSet>,
) -> BgpAnalysisReport {
let seen = self.seen.read().await;
let mut entries = vec![];
let roas: Vec<ConfiguredRoa> = match &limited_scope {
None => roas.to_vec(),
Some(limit) => roas
.iter()
.filter(|roa| limit.contains_roa_address(&roa.as_roa_ip_address()))
.cloned()
.collect(),
};
let (roas_held, roas_not_held): (Vec<ConfiguredRoa>, _) = roas
.into_iter()
.partition(|roa| resources_held.contains_roa_address(&roa.as_roa_ip_address()));
for not_held in roas_not_held {
entries.push(BgpAnalysisEntry::roa_not_held(not_held));
}
if seen.last_checked().is_none() {
for roa in roas_held {
entries.push(BgpAnalysisEntry::roa_no_announcement_info(roa));
}
} else {
let scope = match &limited_scope {
Some(limit) => limit,
None => resources_held,
};
let (v4_scope, v6_scope) = IpRange::for_resource_set(scope);
let mut scoped_announcements = vec![];
for block in v4_scope.into_iter() {
scoped_announcements.append(&mut seen.contained_by(block));
}
for block in v6_scope.into_iter() {
scoped_announcements.append(&mut seen.contained_by(block));
}
let roa_payloads: Vec<_> = roas_held.iter().map(|configured| configured.payload()).collect();
let roa_tree = make_roa_tree(&roa_payloads);
let validated: Vec<ValidatedAnnouncement> = scoped_announcements
.into_iter()
.map(|a| a.validate(&roa_tree))
.collect();
let validated_tree = make_validated_announcement_tree(validated.as_slice());
for roa in roas_held {
let covered = validated_tree.matching_or_more_specific(&roa.prefix());
let other_roas_covering_this_prefix: Vec<_> = roa_tree
.matching_or_less_specific(&roa.prefix())
.into_iter()
.filter(|other| roa.payload() != **other)
.cloned()
.collect();
let other_roas_including_this_definition: Vec<_> = other_roas_covering_this_prefix
.iter()
.filter(|other| {
other.asn() == roa.asn()
&& other.prefix().addr_len() <= roa.prefix().addr_len()
&& other.effective_max_length() >= roa.effective_max_length()
})
.cloned()
.collect();
let authorizes: Vec<Announcement> = covered
.iter()
.filter(|va| {
va.validity() == AnnouncementValidity::Valid
&& va.announcement().prefix().addr_len() <= roa.effective_max_length()
&& va.announcement().asn() == &roa.asn()
})
.map(|va| va.announcement())
.collect();
let disallows: Vec<Announcement> = covered
.iter()
.filter(|va| {
let validity = va.validity();
validity == AnnouncementValidity::InvalidLength || validity == AnnouncementValidity::InvalidAsn
})
.map(|va| va.announcement())
.collect();
let authorizes_excess = {
let max_length = roa.effective_max_length();
let nr_of_specific_ann = authorizes
.iter()
.filter(|ann| ann.prefix().addr_len() == max_length)
.count() as u128;
nr_of_specific_ann > 0 && nr_of_specific_ann < roa.nr_of_specific_prefixes()
};
if roa.asn() == AsNumber::zero() {
if other_roas_covering_this_prefix.is_empty() {
let announcements = covered.iter().map(|va| va.announcement()).collect();
entries.push(BgpAnalysisEntry::roa_as0(roa, announcements));
} else {
entries.push(BgpAnalysisEntry::roa_as0_redundant(
roa,
other_roas_covering_this_prefix,
));
}
} else if !other_roas_including_this_definition.is_empty() {
entries.push(BgpAnalysisEntry::roa_redundant(
roa,
authorizes,
disallows,
other_roas_including_this_definition,
))
} else if authorizes.is_empty() && disallows.is_empty() {
entries.push(BgpAnalysisEntry::roa_unseen(roa))
} else if authorizes_excess {
entries.push(BgpAnalysisEntry::roa_too_permissive(roa, authorizes, disallows))
} else if authorizes.is_empty() {
entries.push(BgpAnalysisEntry::roa_disallowing(roa, disallows))
} else {
entries.push(BgpAnalysisEntry::roa_seen(roa, authorizes, disallows))
}
}
for v in validated.into_iter() {
let (announcement, validity, allowed_by, invalidating_roas) = v.unpack();
match validity {
AnnouncementValidity::Valid => {
entries.push(BgpAnalysisEntry::announcement_valid(
announcement,
allowed_by.unwrap(), ))
}
AnnouncementValidity::Disallowed => {
entries.push(BgpAnalysisEntry::announcement_disallowed(
announcement,
invalidating_roas,
));
}
AnnouncementValidity::InvalidLength => {
entries.push(BgpAnalysisEntry::announcement_invalid_length(
announcement,
invalidating_roas,
));
}
AnnouncementValidity::InvalidAsn => {
entries.push(BgpAnalysisEntry::announcement_invalid_asn(
announcement,
invalidating_roas,
));
}
AnnouncementValidity::NotFound => {
entries.push(BgpAnalysisEntry::announcement_not_found(announcement));
}
}
}
}
BgpAnalysisReport::new(entries)
}
pub async fn suggest(
&self,
roas: &[ConfiguredRoa],
resources_held: &ResourceSet,
limited_scope: Option<ResourceSet>,
) -> BgpAnalysisSuggestion {
let mut suggestion = BgpAnalysisSuggestion::default();
let entries = self.analyse(roas, resources_held, limited_scope).await.into_entries();
for entry in &entries {
match entry.state() {
BgpAnalysisState::RoaUnseen => suggestion.add_stale(entry.configured_roa()),
BgpAnalysisState::RoaTooPermissive => {
let replace_with = entry
.authorizes()
.iter()
.filter(|ann| {
!entries
.iter()
.any(|other| other != entry && other.authorizes().contains(*ann))
})
.map(|auth| RoaPayload::from(*auth))
.collect();
suggestion.add_too_permissive(entry.configured_roa(), replace_with);
}
BgpAnalysisState::RoaSeen | BgpAnalysisState::RoaAs0 => suggestion.add_keep(entry.configured_roa()),
BgpAnalysisState::RoaDisallowing => suggestion.add_disallowing(entry.configured_roa()),
BgpAnalysisState::RoaRedundant => suggestion.add_redundant(entry.configured_roa()),
BgpAnalysisState::RoaNotHeld => suggestion.add_not_held(entry.configured_roa()),
BgpAnalysisState::RoaAs0Redundant => suggestion.add_as0_redundant(entry.configured_roa()),
BgpAnalysisState::AnnouncementValid => {}
BgpAnalysisState::AnnouncementNotFound => suggestion.add_not_found(entry.announcement()),
BgpAnalysisState::AnnouncementInvalidAsn => suggestion.add_invalid_asn(entry.announcement()),
BgpAnalysisState::AnnouncementInvalidLength => suggestion.add_invalid_length(entry.announcement()),
BgpAnalysisState::AnnouncementDisallowed => suggestion.add_keep_disallowing(entry.announcement()),
BgpAnalysisState::RoaNoAnnouncementInfo => suggestion.add_keep(entry.configured_roa()),
}
}
suggestion
}
fn test_announcements() -> Vec<Announcement> {
use crate::test::announcement;
vec![
announcement("10.0.0.0/22 => 64496"),
announcement("10.0.2.0/23 => 64496"),
announcement("10.0.0.0/24 => 64496"),
announcement("10.0.0.0/22 => 64497"),
announcement("10.0.0.0/21 => 64497"),
announcement("192.168.0.0/24 => 64497"),
announcement("192.168.0.0/24 => 64496"),
announcement("192.168.1.0/24 => 64497"),
announcement("2001:DB8::/32 => 64498"),
]
}
fn with_test_announcements() -> Self {
let mut announcements = Announcements::default();
announcements.update(Self::test_announcements());
BgpAnalyser {
dump_loader: None,
seen: RwLock::new(announcements),
}
}
}
#[derive(Debug)]
pub enum BgpAnalyserError {
RisDump(RisDumpError),
}
impl fmt::Display for BgpAnalyserError {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
BgpAnalyserError::RisDump(e) => write!(f, "BGP RIS update error: {}", e),
}
}
}
impl From<RisDumpError> for BgpAnalyserError {
fn from(e: RisDumpError) -> Self {
BgpAnalyserError::RisDump(e)
}
}
#[cfg(test)]
mod tests {
use crate::commons::api::RoaConfigurationUpdates;
use crate::commons::bgp::BgpAnalysisState;
use crate::test::*;
use super::*;
#[tokio::test]
#[ignore]
async fn download_ris_dumps() {
let bgp_ris_dump_v4_uri = "http://www.ris.ripe.net/dumps/riswhoisdump.IPv4.gz";
let bgp_ris_dump_v6_uri = "http://www.ris.ripe.net/dumps/riswhoisdump.IPv6.gz";
let analyser = BgpAnalyser::new(true, bgp_ris_dump_v4_uri, bgp_ris_dump_v6_uri);
assert!(analyser.seen.read().await.is_empty());
assert!(analyser.seen.read().await.last_checked().is_none());
analyser.update().await.unwrap();
assert!(!analyser.seen.read().await.is_empty());
assert!(analyser.seen.read().await.last_checked().is_some());
}
#[tokio::test]
async fn analyse_bgp() {
let roa_too_permissive = configured_roa("10.0.0.0/22-23 => 64496");
let roa_as0 = configured_roa("10.0.4.0/24 => 0");
let roa_unseen_completely = configured_roa("10.0.3.0/24 => 64497");
let roa_not_held = configured_roa("10.1.0.0/24 => 64497");
let roa_authorizing_single = configured_roa("192.168.1.0/24 => 64497");
let roa_unseen_redundant = configured_roa("192.168.1.0/24 => 64498");
let roa_as0_redundant = configured_roa("192.168.1.0/24 => 0");
let resources_held = ResourceSet::from_strs("", "10.0.0.0/16, 192.168.0.0/16", "").unwrap();
let limit = None;
let analyser = BgpAnalyser::with_test_announcements();
let report = analyser
.analyse(
&[
roa_too_permissive,
roa_as0,
roa_unseen_completely,
roa_not_held,
roa_authorizing_single,
roa_unseen_redundant,
roa_as0_redundant,
],
&resources_held,
limit,
)
.await;
let expected: BgpAnalysisReport =
serde_json::from_str(include_str!("../../../test-resources/bgp/expected_full_report.json")).unwrap();
assert_eq!(report, expected);
}
#[tokio::test]
async fn analyse_bgp_disallowed_announcements() {
let roa = configured_roa("10.0.0.0/22 => 0");
let roas = &[roa];
let analyser = BgpAnalyser::with_test_announcements();
let resources_held = ResourceSet::from_strs("", "10.0.0.0/8, 192.168.0.0/16", "").unwrap();
let report = analyser.analyse(roas, &resources_held, None).await;
assert!(!report.contains_invalids());
let mut disallowed = report.matching_announcements(BgpAnalysisState::AnnouncementDisallowed);
disallowed.sort();
let disallowed_1 = announcement("10.0.0.0/22 => 64496");
let disallowed_2 = announcement("10.0.0.0/22 => 64497");
let disallowed_3 = announcement("10.0.0.0/24 => 64496");
let disallowed_4 = announcement("10.0.2.0/23 => 64496");
let mut expected = vec![disallowed_1, disallowed_2, disallowed_3, disallowed_4];
expected.sort();
assert_eq!(disallowed, expected);
let suggestion = analyser.suggest(roas, &resources_held, None).await;
let updates = RoaConfigurationUpdates::from(suggestion);
let added = updates.added();
for announcement in disallowed {
assert!(!added.iter().any(|added_roa| {
let added_payload = added_roa.payload();
let announcement_payload = RoaPayload::from(announcement);
added_payload.includes(&announcement_payload)
}));
}
}
#[tokio::test]
async fn analyse_bgp_no_announcements() {
let roa1 = configured_roa("10.0.0.0/23-24 => 64496");
let roa2 = configured_roa("10.0.3.0/24 => 64497");
let roa3 = configured_roa("10.0.4.0/24 => 0");
let roas = vec![roa1, roa2, roa3];
let resources_held = ResourceSet::from_strs("", "10.0.0.0/16", "").unwrap();
let analyser = BgpAnalyser::new(false, "", "");
let table = analyser.analyse(&roas, &resources_held, None).await;
let table_entries = table.entries();
assert_eq!(3, table_entries.len());
let roas_no_info: Vec<ConfiguredRoa> = table_entries
.iter()
.filter(|e| e.state() == BgpAnalysisState::RoaNoAnnouncementInfo)
.map(|e| e.configured_roa().clone())
.collect();
assert_eq!(roas_no_info, roas);
}
#[tokio::test]
async fn make_bgp_analysis_suggestion() {
let roa_too_permissive = configured_roa("10.0.0.0/22-23 => 64496");
let roa_redundant = configured_roa("10.0.0.0/23 => 64496");
let roa_as0 = configured_roa("10.0.4.0/24 => 0");
let roa_unseen_completely = configured_roa("10.0.3.0/24 => 64497");
let roa_authorizing_single = configured_roa("192.168.1.0/24 => 64497");
let roa_unseen_redundant = configured_roa("192.168.1.0/24 => 64498");
let roa_as0_redundant = configured_roa("192.168.1.0/24 => 0");
let roas = &[
roa_too_permissive,
roa_redundant,
roa_as0,
roa_unseen_completely,
roa_authorizing_single,
roa_unseen_redundant,
roa_as0_redundant,
];
let analyser = BgpAnalyser::with_test_announcements();
let resources_held = ResourceSet::from_strs("", "10.0.0.0/8, 192.168.0.0/16", "").unwrap();
let limit = Some(ResourceSet::from_strs("", "10.0.0.0/22", "").unwrap());
let suggestion_resource_subset = analyser.suggest(roas, &resources_held, limit).await;
let expected: BgpAnalysisSuggestion = serde_json::from_str(include_str!(
"../../../test-resources/bgp/expected_suggestion_some_roas.json"
))
.unwrap();
assert_eq!(suggestion_resource_subset, expected);
let suggestion_all_roas_in_scope = analyser.suggest(roas, &resources_held, None).await;
let expected: BgpAnalysisSuggestion = serde_json::from_str(include_str!(
"../../../test-resources/bgp/expected_suggestion_all_roas.json"
))
.unwrap();
assert_eq!(suggestion_all_roas_in_scope, expected);
}
}