diff --git a/crates/panda-rpki-validator/src/fetch/http.rs b/crates/panda-rpki-validator/src/fetch/http.rs index 28aecd6..5dedba0 100644 --- a/crates/panda-rpki-validator/src/fetch/http.rs +++ b/crates/panda-rpki-validator/src/fetch/http.rs @@ -3,9 +3,12 @@ use std::io::Write; use std::time::Duration; use reqwest::blocking::Client; -use reqwest::header::HeaderMap; +use reqwest::header::{HeaderMap, LOCATION}; +use url::Url; -use crate::sync::rrdp::Fetcher; +use crate::sync::rrdp::{Fetcher, RrdpError, RrdpFetchError, RrdpOrigin, RrdpResourceKind}; + +const MAX_RRDP_REDIRECTS: usize = 10; thread_local! { static HTTP_TIMEOUT_OVERRIDE: RefCell> = const { RefCell::new(None) }; @@ -77,6 +80,9 @@ pub struct BlockingHttpFetcher { short_client: Client, large_body_client: Client, retry_short_client: Client, + rrdp_short_client: Client, + rrdp_large_body_client: Client, + rrdp_retry_short_client: Client, short_timeout: Duration, large_body_timeout: Duration, } @@ -110,10 +116,37 @@ impl BlockingHttpFetcher { )? .build() .map_err(|e| e.to_string())?; + let rrdp_short_client = Self::rrdp_client_builder( + &config, + connect_timeout, + config.timeout, + config.user_agent.clone(), + )? + .build() + .map_err(|e| e.to_string())?; + let rrdp_large_body_client = Self::rrdp_client_builder( + &config, + connect_timeout, + large_body_timeout, + config.user_agent.clone(), + )? + .build() + .map_err(|e| e.to_string())?; + let rrdp_retry_short_client = Self::rrdp_client_builder( + &config, + Duration::from_secs(1), + Duration::from_secs(1), + config.user_agent.clone(), + )? + .build() + .map_err(|e| e.to_string())?; Ok(Self { short_client, large_body_client, retry_short_client, + rrdp_short_client, + rrdp_large_body_client, + rrdp_retry_short_client, short_timeout, large_body_timeout, }) @@ -137,6 +170,18 @@ impl BlockingHttpFetcher { Ok(builder) } + fn rrdp_client_builder( + config: &HttpFetcherConfig, + connect_timeout: Duration, + timeout: Duration, + user_agent: String, + ) -> Result { + Ok( + Self::client_builder(config, connect_timeout, timeout, user_agent)? + .redirect(reqwest::redirect::Policy::none()), + ) + } + pub fn fetch_bytes(&self, uri: &str) -> Result, String> { let started = std::time::Instant::now(); let (client, timeout_profile, timeout_value) = self.client_for_uri(uri); @@ -260,6 +305,143 @@ impl BlockingHttpFetcher { (&self.short_client, "short", self.short_timeout) } } + + fn rrdp_client_for_uri(&self, uri: &str) -> (&Client, &'static str, Duration) { + if let Some(timeout) = HTTP_TIMEOUT_OVERRIDE.with(|cell| *cell.borrow()) { + return (&self.rrdp_retry_short_client, "retry_short", timeout); + } + if uses_large_body_timeout(uri) { + ( + &self.rrdp_large_body_client, + "large_body", + self.large_body_timeout, + ) + } else { + (&self.rrdp_short_client, "short", self.short_timeout) + } + } + + fn rrdp_response( + &self, + resource: RrdpResourceKind, + uri: &str, + notification_origin: &RrdpOrigin, + ) -> Result<(reqwest::blocking::Response, &'static str, Duration), RrdpFetchError> { + let mut current = Url::parse(uri).map_err(|e| RrdpError::InvalidRrdpUri { + resource, + detail: e.to_string(), + })?; + let current_origin = RrdpOrigin::from_url(¤t, resource)?; + if ¤t_origin != notification_origin { + return Err(RrdpError::CrossOriginReference { + resource, + notification_origin: notification_origin.to_string(), + resource_origin: current_origin.to_string(), + } + .into()); + } + + for redirects in 0..=MAX_RRDP_REDIRECTS { + let (client, profile, timeout) = self.rrdp_client_for_uri(current.as_str()); + let response = client + .get(current.clone()) + .send() + .map_err(|e| RrdpFetchError::Fetch(format!("http request failed: {e:?}")))?; + if !response.status().is_redirection() { + return Ok((response, profile, timeout)); + } + if redirects == MAX_RRDP_REDIRECTS { + return Err(RrdpError::RedirectLimitExceeded { + resource, + max: MAX_RRDP_REDIRECTS, + } + .into()); + } + let location = response + .headers() + .get(LOCATION) + .ok_or(RrdpError::RedirectLocationMissing { resource })? + .to_str() + .map_err(|_| RrdpError::InvalidRedirectLocation { + resource, + location: "".to_string(), + })?; + let next = current + .join(location) + .map_err(|_| RrdpError::InvalidRedirectLocation { + resource, + location: location.to_string(), + })?; + let next_origin = RrdpOrigin::from_url(&next, resource)?; + crate::progress_log::emit( + "rrdp_redirect_checked", + serde_json::json!({ + "resource": resource.to_string(), + "from": current.as_str(), + "to": next.as_str(), + "notification_origin": notification_origin.to_string(), + "same_origin": &next_origin == notification_origin, + "rfc": "RFC 9674 §3.2", + }), + ); + if &next_origin != notification_origin { + crate::progress_log::emit( + "rrdp_cross_origin_rejected", + serde_json::json!({ + "resource": resource.to_string(), + "notification_origin": notification_origin.to_string(), + "resource_origin": next_origin.to_string(), + "reason": "redirect", + "rfc": "RFC 9674 §3.2", + }), + ); + return Err(RrdpError::CrossOriginRedirect { + resource, + notification_origin: notification_origin.to_string(), + redirect_origin: next_origin.to_string(), + } + .into()); + } + current = next; + } + unreachable!("redirect loop either returns or reaches its limit") + } + + fn rrdp_fetch_bytes( + &self, + resource: RrdpResourceKind, + uri: &str, + notification_origin: &RrdpOrigin, + ) -> Result, RrdpFetchError> { + let (response, _profile, _timeout) = + self.rrdp_response(resource, uri, notification_origin)?; + let status = response.status(); + if !status.is_success() { + return Err(RrdpFetchError::Fetch(format!("http status {status}"))); + } + response + .bytes() + .map(|b| b.to_vec()) + .map_err(|e| RrdpFetchError::Fetch(format!("http read body failed: {e}"))) + } + + fn rrdp_fetch_to_writer( + &self, + resource: RrdpResourceKind, + uri: &str, + notification_origin: &RrdpOrigin, + out: &mut dyn Write, + ) -> Result { + let (mut response, _profile, _timeout) = + self.rrdp_response(resource, uri, notification_origin)?; + let status = response.status(); + if !status.is_success() { + return Err(RrdpFetchError::Fetch(format!("http status {status}"))); + } + response + .copy_to(out) + .map_err(|e| RrdpFetchError::Fetch(format!("http stream body failed: {e}"))) + } } impl Fetcher for BlockingHttpFetcher { @@ -376,6 +558,25 @@ impl Fetcher for BlockingHttpFetcher { } } } + + fn fetch_rrdp( + &self, + resource: RrdpResourceKind, + uri: &str, + notification_origin: &RrdpOrigin, + ) -> Result, RrdpFetchError> { + self.rrdp_fetch_bytes(resource, uri, notification_origin) + } + + fn fetch_rrdp_to_writer( + &self, + resource: RrdpResourceKind, + uri: &str, + notification_origin: &RrdpOrigin, + out: &mut dyn Write, + ) -> Result { + self.rrdp_fetch_to_writer(resource, uri, notification_origin, out) + } } fn header_value(headers: &HeaderMap, name: &str) -> String { @@ -398,6 +599,8 @@ mod tests { use super::*; use std::io::{Read, Write}; use std::net::TcpListener; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::thread; use std::time::Duration as StdDuration; @@ -649,4 +852,85 @@ mod tests { "{request}" ); } + + #[test] + fn rrdp_same_origin_redirect_is_followed_manually() { + let listener = TcpListener::bind(("127.0.0.1", 0)).expect("bind"); + let addr = listener.local_addr().expect("addr"); + thread::spawn(move || { + for response in [ + b"HTTP/1.1 302 Found\r\nLocation: /final\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + .as_slice(), + b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok".as_slice(), + ] { + let (mut stream, _) = listener.accept().expect("accept"); + let mut buf = [0u8; 1024]; + let _ = stream.read(&mut buf); + stream.write_all(response).expect("response"); + } + }); + let uri = format!("http://{addr}/start"); + let origin = RrdpOrigin::parse(&uri, RrdpResourceKind::Notification).expect("origin"); + let http = BlockingHttpFetcher::new(HttpFetcherConfig { + timeout: Duration::from_secs(2), + ..HttpFetcherConfig::default() + }) + .expect("http"); + let body = http + .fetch_rrdp(RrdpResourceKind::Notification, &uri, &origin) + .expect("same-origin redirect accepted"); + assert_eq!(body, b"ok"); + } + + #[test] + fn rrdp_cross_origin_redirect_is_rejected_before_target_request() { + let target = TcpListener::bind(("127.0.0.1", 0)).expect("bind target"); + target.set_nonblocking(true).expect("nonblocking target"); + let target_addr = target.local_addr().expect("target addr"); + let target_hits = Arc::new(AtomicUsize::new(0)); + let target_hits_worker = Arc::clone(&target_hits); + let target_thread = thread::spawn(move || { + let deadline = std::time::Instant::now() + StdDuration::from_millis(400); + while std::time::Instant::now() < deadline { + match target.accept() { + Ok((mut stream, _)) => { + target_hits_worker.fetch_add(1, Ordering::SeqCst); + let _ = stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n"); + } + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { + std::thread::sleep(StdDuration::from_millis(5)); + } + Err(e) => panic!("target accept: {e}"), + } + } + }); + + let source = TcpListener::bind(("127.0.0.1", 0)).expect("bind source"); + let source_addr = source.local_addr().expect("source addr"); + thread::spawn(move || { + let (mut stream, _) = source.accept().expect("accept source"); + let mut buf = [0u8; 1024]; + let _ = stream.read(&mut buf); + let response = format!( + "HTTP/1.1 302 Found\r\nLocation: http://{target_addr}/foreign\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + ); + stream.write_all(response.as_bytes()).expect("redirect"); + }); + let uri = format!("http://{source_addr}/start"); + let origin = RrdpOrigin::parse(&uri, RrdpResourceKind::Notification).expect("origin"); + let http = BlockingHttpFetcher::new(HttpFetcherConfig { + timeout: Duration::from_secs(2), + ..HttpFetcherConfig::default() + }) + .expect("http"); + let err = http + .fetch_rrdp(RrdpResourceKind::Notification, &uri, &origin) + .unwrap_err(); + assert!(matches!( + err, + RrdpFetchError::Rrdp(RrdpError::CrossOriginRedirect { .. }) + )); + target_thread.join().expect("target thread"); + assert_eq!(target_hits.load(Ordering::SeqCst), 0); + } } diff --git a/crates/panda-rpki-validator/src/sync/repo.rs b/crates/panda-rpki-validator/src/sync/repo.rs index 827c03b..eff84a1 100644 --- a/crates/panda-rpki-validator/src/sync/repo.rs +++ b/crates/panda-rpki-validator/src/sync/repo.rs @@ -9,7 +9,10 @@ use crate::replay::delta_archive::{ReplayDeltaArchiveIndex, ReplayDeltaRrdpKind} use crate::report::{RfcRef, Warning}; use crate::storage::RocksStore; use crate::sync::rrdp::sync_from_notification_with_timing_and_download_log; -use crate::sync::rrdp::{Fetcher as HttpFetcher, RrdpSyncError, load_rrdp_local_state}; +use crate::sync::rrdp::{ + Fetcher as HttpFetcher, RrdpFetchError, RrdpOrigin, RrdpResourceKind, RrdpSyncError, + load_rrdp_local_state, +}; use crate::sync::store_projection::{ build_repository_view_present_entry, build_repository_view_withdrawn_entry, prepare_repo_bytes_batch_owned, @@ -127,9 +130,13 @@ pub fn sync_publication_point( "rrdp_error": err.to_string(), }), ); + let mut rfc_refs = vec![RfcRef("RFC 8182 §3.4.5")]; + if matches!(&err, RrdpSyncError::Rrdp(e) if e.is_cross_origin_violation()) { + rfc_refs.push(RfcRef("RFC 9674 §3.2")); + } let warnings = vec![ Warning::new(format!("RRDP failed; falling back to rsync: {err}")) - .with_rfc_refs(&[RfcRef("RFC 8182 §3.4.5")]) + .with_rfc_refs(&rfc_refs) .with_context(notification_uri), ]; let written = rsync_sync_into_current_store( @@ -419,6 +426,7 @@ fn try_rrdp_sync( timing: Option<&TimingHandle>, download_log: Option<&DownloadLogHandle>, ) -> Result { + let notification_origin = RrdpOrigin::parse(notification_uri, RrdpResourceKind::Notification)?; let notification_xml = { let _step = timing .as_ref() @@ -428,7 +436,11 @@ fn try_rrdp_sync( .map(|t| t.span_phase("rrdp_fetch_notification_total")); let mut dl_span = download_log .map(|dl| dl.span_download(AuditDownloadKind::RrdpNotification, notification_uri)); - match http_fetcher.fetch(notification_uri) { + match http_fetcher.fetch_rrdp( + RrdpResourceKind::Notification, + notification_uri, + ¬ification_origin, + ) { Ok(v) => { if let Some(t) = timing.as_ref() { t.record_count("rrdp_notification_fetch_ok_total", 1); @@ -439,7 +451,7 @@ fn try_rrdp_sync( } v } - Err(e) => { + Err(RrdpFetchError::Fetch(e)) => { if let Some(t) = timing.as_ref() { t.record_count("rrdp_notification_fetch_fail_total", 1); } @@ -448,6 +460,18 @@ fn try_rrdp_sync( } return Err(RrdpSyncError::Fetch(e)); } + Err(RrdpFetchError::Rrdp(e)) => { + if let Some(t) = timing.as_ref() { + t.record_count("rrdp_notification_fetch_fail_total", 1); + if e.is_cross_origin_violation() { + t.record_count("rrdp_rejected_cross_origin_total", 1); + } + } + if let Some(s) = dl_span.as_mut() { + s.set_err(e.to_string()); + } + return Err(e.into()); + } } }; if let Some(t) = timing.as_ref() { diff --git a/crates/panda-rpki-validator/src/sync/repo/tests_parts/fallback_and_replay.rs b/crates/panda-rpki-validator/src/sync/repo/tests_parts/fallback_and_replay.rs index 097a546..8c0697d 100644 --- a/crates/panda-rpki-validator/src/sync/repo/tests_parts/fallback_and_replay.rs +++ b/crates/panda-rpki-validator/src/sync/repo/tests_parts/fallback_and_replay.rs @@ -114,6 +114,86 @@ fn rrdp_protocol_error_does_not_retry_and_falls_back_to_rsync() { ); } +#[test] +fn rrdp_cross_origin_reference_falls_back_to_rsync_without_fetching_foreign_uri() { + let temp = tempfile::tempdir().expect("tempdir"); + let store = RocksStore::open(&temp.path().join("db")).expect("open rocksdb"); + let notification_uri = "https://origin.example/notification.xml"; + let foreign_snapshot_uri = "https://foreign.example/snapshot.xml"; + let notification = notification_xml( + "9df4b597-af9e-4dca-bdda-719cce2c4e28", + 1, + foreign_snapshot_uri, + &"00".repeat(32), + ); + struct RecordingHttp { + notification: Vec, + notification_calls: AtomicUsize, + foreign_calls: AtomicUsize, + } + impl HttpFetcher for RecordingHttp { + fn fetch(&self, uri: &str) -> Result, String> { + if uri == "https://origin.example/notification.xml" { + self.notification_calls.fetch_add(1, Ordering::SeqCst); + return Ok(self.notification.clone()); + } + if uri == "https://foreign.example/snapshot.xml" { + self.foreign_calls.fetch_add(1, Ordering::SeqCst); + } + Err(format!("unexpected HTTP fetch: {uri}")) + } + } + struct OneObjectRsync; + impl RsyncFetcher for OneObjectRsync { + fn fetch_objects( + &self, + _rsync_base_uri: &str, + ) -> Result)>, RsyncFetchError> { + Ok(vec![( + "rsync://origin.example/repo/a.mft".to_string(), + b"mft".to_vec(), + )]) + } + } + + let http = RecordingHttp { + notification, + notification_calls: AtomicUsize::new(0), + foreign_calls: AtomicUsize::new(0), + }; + let policy = Policy { + sync_preference: SyncPreference::RrdpThenRsync, + ..Policy::default() + }; + let out = sync_publication_point( + &store, + &policy, + Some(notification_uri), + "rsync://origin.example/repo/", + &http, + &OneObjectRsync, + None, + None, + ) + .expect("rsync fallback succeeds"); + + assert_eq!(out.phase, RepoSyncPhase::RrdpFailedRsyncOk); + assert_eq!(http.notification_calls.load(Ordering::SeqCst), 1); + assert_eq!(http.foreign_calls.load(Ordering::SeqCst), 0); + assert!( + out.warnings[0] + .message + .contains("crosses notification origin") + ); + assert!( + out.warnings[0] + .rfc_refs + .iter() + .any(|reference| reference.0 == "RFC 9674 §3.2") + ); + assert_current_object(&store, "rsync://origin.example/repo/a.mft", b"mft"); +} + #[test] fn rrdp_delta_fetches_are_logged_even_if_snapshot_fallback_is_used() { let temp = tempfile::tempdir().expect("tempdir"); diff --git a/crates/panda-rpki-validator/src/sync/rrdp.rs b/crates/panda-rpki-validator/src/sync/rrdp.rs index 469f61e..5c7ecd9 100644 --- a/crates/panda-rpki-validator/src/sync/rrdp.rs +++ b/crates/panda-rpki-validator/src/sync/rrdp.rs @@ -17,12 +17,34 @@ use crate::sync::store_projection::{ use base64::Engine; use serde::{Deserialize, Serialize}; use sha2::Digest; +use std::fmt; use std::io::Write; +use url::Url; use uuid::Uuid; const RRDP_XMLNS: &str = "http://www.ripe.net/rpki/rrdp"; const RRDP_SNAPSHOT_APPLY_BATCH_SIZE: usize = 1024; +/// The RRDP object currently being fetched. Keeping this in protocol errors +/// makes an origin rejection actionable without treating it as a transient +/// transport failure. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum RrdpResourceKind { + Notification, + Snapshot, + Delta, +} + +impl fmt::Display for RrdpResourceKind { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(match self { + Self::Notification => "notification", + Self::Snapshot => "snapshot", + Self::Delta => "delta", + }) + } +} + include!("rrdp/models_and_parsing.rs"); include!("rrdp/snapshot_sync.rs"); include!("rrdp/notification_sync.rs"); diff --git a/crates/panda-rpki-validator/src/sync/rrdp/models_and_parsing.rs b/crates/panda-rpki-validator/src/sync/rrdp/models_and_parsing.rs index b52129a..def5f35 100644 --- a/crates/panda-rpki-validator/src/sync/rrdp/models_and_parsing.rs +++ b/crates/panda-rpki-validator/src/sync/rrdp/models_and_parsing.rs @@ -154,6 +154,110 @@ pub enum RrdpError { #[error("delta file contains unexpected element <{0}> (RFC 8182 §3.5.3.3)")] DeltaUnexpectedElement(String), + + #[error("invalid RRDP {resource} URI: {detail} (RFC 9674 §3.2)")] + InvalidRrdpUri { + resource: RrdpResourceKind, + detail: String, + }, + + #[error( + "RRDP {resource} URI crosses notification origin: notification={notification_origin}, resource={resource_origin} (RFC 9674 §3.2)" + )] + CrossOriginReference { + resource: RrdpResourceKind, + notification_origin: String, + resource_origin: String, + }, + + #[error( + "RRDP {resource} redirect crosses notification origin: notification={notification_origin}, redirect={redirect_origin} (RFC 9674 §3.2)" + )] + CrossOriginRedirect { + resource: RrdpResourceKind, + notification_origin: String, + redirect_origin: String, + }, + + #[error("RRDP {resource} redirect has no Location header (RFC 9674 §3.2)")] + RedirectLocationMissing { resource: RrdpResourceKind }, + + #[error("RRDP {resource} redirect Location is invalid: {location} (RFC 9674 §3.2)")] + InvalidRedirectLocation { + resource: RrdpResourceKind, + location: String, + }, + + #[error("RRDP {resource} redirect limit exceeded ({max}) (RFC 9674 §3.2)")] + RedirectLimitExceeded { + resource: RrdpResourceKind, + max: usize, + }, +} + +impl RrdpError { + pub fn is_cross_origin_violation(&self) -> bool { + matches!( + self, + Self::CrossOriginReference { .. } | Self::CrossOriginRedirect { .. } + ) + } +} + +/// RFC 9674 §3.2 compares scheme, host name, and port. `Url` normalizes an +/// omitted well-known port, so `https://example/` and `https://example:443/` +/// intentionally have the same effective origin. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RrdpOrigin { + scheme: String, + host: String, + port: u16, +} + +impl fmt::Display for RrdpOrigin { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}://{}:{}", self.scheme, self.host, self.port) + } +} + +impl RrdpOrigin { + pub fn parse(uri: &str, resource: RrdpResourceKind) -> Result { + let url = Url::parse(uri).map_err(|e| RrdpError::InvalidRrdpUri { + resource, + detail: e.to_string(), + })?; + Self::from_url(&url, resource) + } + + pub fn from_url(url: &Url, resource: RrdpResourceKind) -> Result { + if !matches!(url.scheme(), "http" | "https") { + return Err(RrdpError::InvalidRrdpUri { + resource, + detail: format!("unsupported scheme {}", url.scheme()), + }); + } + if !url.username().is_empty() || url.password().is_some() { + return Err(RrdpError::InvalidRrdpUri { + resource, + detail: "userinfo is not permitted".to_string(), + }); + } + let host = url.host_str().ok_or_else(|| RrdpError::InvalidRrdpUri { + resource, + detail: "host is missing".to_string(), + })?; + let port = url + .port_or_known_default() + .ok_or_else(|| RrdpError::InvalidRrdpUri { + resource, + detail: "port is missing or unknown".to_string(), + })?; + Ok(Self { + scheme: url.scheme().to_ascii_lowercase(), + host: host.to_ascii_lowercase(), + port, + }) + } } #[derive(Debug, thiserror::Error)] @@ -168,6 +272,15 @@ pub enum RrdpSyncError { Storage(String), } +#[derive(Debug, thiserror::Error)] +pub enum RrdpFetchError { + #[error("{0}")] + Rrdp(#[from] RrdpError), + + #[error("{0}")] + Fetch(String), +} + pub type RrdpSyncResult = Result; pub trait Fetcher: Send + Sync { @@ -179,6 +292,29 @@ pub trait Fetcher: Send + Sync { .map_err(|e| format!("write sink failed: {e}"))?; Ok(bytes.len() as u64) } + + /// RRDP-only fetch path. Implementations that do not perform HTTP may use + /// the compatibility default; HTTP implementations must enforce redirect + /// origin checks before issuing the next request. + fn fetch_rrdp( + &self, + _resource: RrdpResourceKind, + uri: &str, + _notification_origin: &RrdpOrigin, + ) -> Result, RrdpFetchError> { + self.fetch(uri).map_err(RrdpFetchError::Fetch) + } + + fn fetch_rrdp_to_writer( + &self, + _resource: RrdpResourceKind, + uri: &str, + _notification_origin: &RrdpOrigin, + out: &mut dyn Write, + ) -> Result { + self.fetch_to_writer(uri, out) + .map_err(RrdpFetchError::Fetch) + } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -385,6 +521,51 @@ pub fn parse_notification_snapshot(xml: &[u8]) -> Result Result { + let notification_origin = RrdpOrigin::parse(notification_uri, RrdpResourceKind::Notification)?; + validate_reference_origin( + ¬ification_origin, + ¬ification.snapshot_uri, + RrdpResourceKind::Snapshot, + )?; + for delta in ¬ification.deltas { + validate_reference_origin(¬ification_origin, &delta.uri, RrdpResourceKind::Delta)?; + } + Ok(notification_origin) +} + +fn validate_reference_origin( + notification_origin: &RrdpOrigin, + uri: &str, + resource: RrdpResourceKind, +) -> Result<(), RrdpError> { + let resource_origin = RrdpOrigin::parse(uri, resource)?; + if &resource_origin != notification_origin { + crate::progress_log::emit( + "rrdp_cross_origin_rejected", + serde_json::json!({ + "resource": resource.to_string(), + "notification_origin": notification_origin.to_string(), + "resource_origin": resource_origin.to_string(), + "reason": "direct_reference", + "rfc": "RFC 9674 §3.2", + }), + ); + return Err(RrdpError::CrossOriginReference { + resource, + notification_origin: notification_origin.to_string(), + resource_origin: resource_origin.to_string(), + }); + } + Ok(()) +} + pub fn parse_delta_file(xml: &[u8]) -> Result { let doc = parse_rrdp_xml(xml)?; let root = doc.root_element(); diff --git a/crates/panda-rpki-validator/src/sync/rrdp/notification_sync.rs b/crates/panda-rpki-validator/src/sync/rrdp/notification_sync.rs index af6b343..743a0d3 100644 --- a/crates/panda-rpki-validator/src/sync/rrdp/notification_sync.rs +++ b/crates/panda-rpki-validator/src/sync/rrdp/notification_sync.rs @@ -71,6 +71,18 @@ fn sync_from_notification_inner( .as_ref() .map(|t| t.span_phase("rrdp_parse_notification_total")); let notif = parse_notification(notification_xml)?; + let notification_origin = match validate_notification_references(notification_uri, ¬if) { + Ok(origin) => origin, + Err(e) => { + if let Some(t) = timing.as_ref() { + t.record_count( + "rrdp_rejected_cross_origin_total", + e.is_cross_origin_violation() as u64, + ); + } + return Err(e.into()); + } + }; drop(_parse_step); drop(_parse_total); if let Some(t) = timing.as_ref() { @@ -161,7 +173,11 @@ fn sync_from_notification_inner( let mut dl_span = download_log .map(|dl| dl.span_download(AuditDownloadKind::RrdpDelta, &dref.uri)); - match fetcher.fetch(&dref.uri) { + match fetcher.fetch_rrdp( + RrdpResourceKind::Delta, + &dref.uri, + ¬ification_origin, + ) { Ok(bytes) => { if let Some(t) = timing.as_ref() { t.record_count("rrdp_delta_fetch_ok_total", 1); @@ -173,7 +189,7 @@ fn sync_from_notification_inner( } fetched.push((serial, dref.hash_sha256, bytes)) } - Err(e) => { + Err(RrdpFetchError::Fetch(e)) => { if let Some(t) = timing.as_ref() { t.record_count("rrdp_delta_fetch_fail_total", 1); } @@ -183,6 +199,7 @@ fn sync_from_notification_inner( fetch_ok = false; break; } + Err(RrdpFetchError::Rrdp(e)) => return Err(e.into()), } } drop(_fetch_d_step); @@ -281,6 +298,7 @@ fn sync_from_notification_inner( fetcher, ¬if.snapshot_uri, ¬if.snapshot_hash_sha256, + ¬ification_origin, ) { Ok(v) => { if let Some(t) = timing.as_ref() { diff --git a/crates/panda-rpki-validator/src/sync/rrdp/snapshot_apply.rs b/crates/panda-rpki-validator/src/sync/rrdp/snapshot_apply.rs index 9360b89..613cd57 100644 --- a/crates/panda-rpki-validator/src/sync/rrdp/snapshot_apply.rs +++ b/crates/panda-rpki-validator/src/sync/rrdp/snapshot_apply.rs @@ -15,8 +15,8 @@ use crate::sync::store_projection::{ }; use super::{ - Fetcher, RRDP_SNAPSHOT_APPLY_BATCH_SIZE, RRDP_XMLNS, RrdpError, RrdpSyncError, parse_u64_str, - strip_all_ascii_whitespace, + Fetcher, RRDP_SNAPSHOT_APPLY_BATCH_SIZE, RRDP_XMLNS, RrdpError, RrdpFetchError, RrdpOrigin, + RrdpResourceKind, RrdpSyncError, parse_u64_str, strip_all_ascii_whitespace, }; #[cfg(test)] @@ -422,14 +422,23 @@ pub(super) fn fetch_snapshot_into_tempfile( fetcher: &dyn Fetcher, snapshot_uri: &str, expected_hash_sha256: &[u8; 32], + notification_origin: &RrdpOrigin, ) -> Result<(tempfile::NamedTempFile, u64), RrdpSyncError> { let mut tmp = tempfile::NamedTempFile::new() .map_err(|e| RrdpSyncError::Fetch(format!("tempfile create failed: {e}")))?; let mut spool = SnapshotSpoolWriter::new(tmp.as_file_mut()); - let bytes_written = match fetcher.fetch_to_writer(snapshot_uri, &mut spool) { + let bytes_written = match fetcher.fetch_rrdp_to_writer( + RrdpResourceKind::Snapshot, + snapshot_uri, + notification_origin, + &mut spool, + ) { Ok(bytes) => bytes, - Err(e) if e.contains(SNAPSHOT_NON_ASCII_ERROR) => return Err(RrdpError::NotAscii.into()), - Err(e) => return Err(RrdpSyncError::Fetch(e)), + Err(RrdpFetchError::Fetch(e)) if e.contains(SNAPSHOT_NON_ASCII_ERROR) => { + return Err(RrdpError::NotAscii.into()); + } + Err(RrdpFetchError::Fetch(e)) => return Err(RrdpSyncError::Fetch(e)), + Err(RrdpFetchError::Rrdp(e)) => return Err(e.into()), }; let computed = spool.finalize_hash(); if computed.as_slice() != expected_hash_sha256.as_slice() { diff --git a/crates/panda-rpki-validator/src/sync/rrdp/snapshot_sync.rs b/crates/panda-rpki-validator/src/sync/rrdp/snapshot_sync.rs index 3038b78..2c86de7 100644 --- a/crates/panda-rpki-validator/src/sync/rrdp/snapshot_sync.rs +++ b/crates/panda-rpki-validator/src/sync/rrdp/snapshot_sync.rs @@ -65,7 +65,19 @@ fn sync_from_notification_snapshot_inner( let _parse_total = timing .as_ref() .map(|t| t.span_phase("rrdp_parse_notification_total")); - let notif = parse_notification_snapshot(notification_xml)?; + let notif = parse_notification(notification_xml)?; + let notification_origin = match validate_notification_references(notification_uri, ¬if) { + Ok(origin) => origin, + Err(e) => { + if let Some(t) = timing.as_ref() { + t.record_count( + "rrdp_rejected_cross_origin_total", + e.is_cross_origin_violation() as u64, + ); + } + return Err(e.into()); + } + }; drop(_parse_step); drop(_parse_total); @@ -81,6 +93,7 @@ fn sync_from_notification_snapshot_inner( fetcher, ¬if.snapshot_uri, ¬if.snapshot_hash_sha256, + ¬ification_origin, ) { Ok(v) => { if let Some(t) = timing.as_ref() { diff --git a/crates/panda-rpki-validator/src/sync/rrdp/tests_parts/parsing.rs b/crates/panda-rpki-validator/src/sync/rrdp/tests_parts/parsing.rs index c932a87..52035d2 100644 --- a/crates/panda-rpki-validator/src/sync/rrdp/tests_parts/parsing.rs +++ b/crates/panda-rpki-validator/src/sync/rrdp/tests_parts/parsing.rs @@ -131,9 +131,10 @@ fn fetch_snapshot_into_tempfile_streams_and_validates_hash() { let fetcher = WriterOnlyFetcher { map: HashMap::from([(snapshot_uri.to_string(), snapshot.clone())]), }; + let origin = RrdpOrigin::parse(snapshot_uri, RrdpResourceKind::Notification).expect("origin"); let (mut file, bytes_written) = - fetch_snapshot_into_tempfile(&fetcher, snapshot_uri, &expected_hash) + fetch_snapshot_into_tempfile(&fetcher, snapshot_uri, &expected_hash, &origin) .expect("fetch snapshot into tempfile"); assert_eq!(bytes_written, snapshot.len() as u64); let mut got = Vec::new(); @@ -144,7 +145,8 @@ fn fetch_snapshot_into_tempfile_streams_and_validates_hash() { let mut wrong_hash = expected_hash; wrong_hash[0] ^= 0xff; - let err = fetch_snapshot_into_tempfile(&fetcher, snapshot_uri, &wrong_hash).unwrap_err(); + let err = + fetch_snapshot_into_tempfile(&fetcher, snapshot_uri, &wrong_hash, &origin).unwrap_err(); assert!(matches!( err, RrdpSyncError::Rrdp(RrdpError::SnapshotHashMismatch) @@ -157,6 +159,11 @@ fn fetch_snapshot_into_tempfile_maps_stream_non_ascii_error() { &NonAsciiWriterFetcher, "https://example.test/snapshot.xml", &[0u8; 32], + &RrdpOrigin::parse( + "https://example.test/notification.xml", + RrdpResourceKind::Notification, + ) + .expect("origin"), ) .unwrap_err(); assert!(matches!(err, RrdpSyncError::Rrdp(RrdpError::NotAscii))); diff --git a/crates/panda-rpki-validator/src/sync/rrdp/tests_parts/sync.rs b/crates/panda-rpki-validator/src/sync/rrdp/tests_parts/sync.rs index f04f6f4..dcf0321 100644 --- a/crates/panda-rpki-validator/src/sync/rrdp/tests_parts/sync.rs +++ b/crates/panda-rpki-validator/src/sync/rrdp/tests_parts/sync.rs @@ -35,6 +35,95 @@ fn sync_from_notification_snapshot_rejects_cross_source_owner_conflict() { assert!(err.to_string().contains("owner conflict"), "{err}"); } +struct PanicOnRrdpFetch; + +impl Fetcher for PanicOnRrdpFetch { + fn fetch(&self, uri: &str) -> Result, String> { + panic!("cross-origin notification must be rejected before fetch: {uri}"); + } +} + +#[test] +fn cross_origin_snapshot_reference_is_rejected_before_fetch_or_state_write() { + let temp = tempfile::tempdir().expect("tempdir"); + let store = RocksStore::open(temp.path()).expect("open rocksdb"); + let notification_uri = "https://origin.example.test/notification.xml"; + let notification = notification_xml( + "550e8400-e29b-41d4-a716-446655440000", + 1, + "https://foreign.example.test/snapshot.xml", + &"00".repeat(32), + ); + + let err = + sync_from_notification_snapshot(&store, notification_uri, ¬ification, &PanicOnRrdpFetch) + .expect_err("cross-origin snapshot must be rejected"); + assert!(matches!( + err, + RrdpSyncError::Rrdp(RrdpError::CrossOriginReference { + resource: RrdpResourceKind::Snapshot, + .. + }) + )); + assert!( + load_rrdp_local_state(&store, notification_uri) + .expect("read state") + .is_none(), + "rejection must precede RRDP state writes" + ); +} + +#[test] +fn cross_origin_delta_reference_is_rejected_even_when_delta_would_not_be_used() { + let temp = tempfile::tempdir().expect("tempdir"); + let store = RocksStore::open(temp.path()).expect("open rocksdb"); + let notification_uri = "https://origin.example.test/notification.xml"; + let notification = notification_xml_with_deltas( + "550e8400-e29b-41d4-a716-446655440000", + 1, + "https://origin.example.test/snapshot.xml", + &"00".repeat(32), + &[( + "delta-1", + 1, + "https://foreign.example.test/delta-1.xml", + &"11".repeat(32), + )], + ); + + let err = sync_from_notification(&store, notification_uri, ¬ification, &PanicOnRrdpFetch) + .expect_err("cross-origin delta must be rejected before state/fetch"); + assert!(matches!( + err, + RrdpSyncError::Rrdp(RrdpError::CrossOriginReference { + resource: RrdpResourceKind::Delta, + .. + }) + )); + assert!( + load_rrdp_local_state(&store, notification_uri) + .expect("read state") + .is_none(), + "all delta references are checked before state reads or writes" + ); +} + +#[test] +fn rrdp_origin_treats_https_default_port_as_equivalent() { + let notification = parse_notification(¬ification_xml( + "550e8400-e29b-41d4-a716-446655440000", + 1, + "https://origin.example.test:443/snapshot.xml", + &"00".repeat(32), + )) + .expect("notification"); + validate_notification_references( + "https://origin.example.test/notification.xml", + ¬ification, + ) + .expect("explicit HTTPS default port must remain same-origin"); +} + #[test] fn sync_from_notification_snapshot_applies_snapshot_and_stores_state() { let tmp = tempfile::tempdir().expect("tempdir");