20260903 同步 RRDP RFC9674 同源防护

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<Option<Duration>> = 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<reqwest::blocking::ClientBuilder, String> {
+        Ok(
+            Self::client_builder(config, connect_timeout, timeout, user_agent)?
+                .redirect(reqwest::redirect::Policy::none()),
+        )
+    }
+
     pub fn fetch_bytes(&self, uri: &str) -> Result<Vec<u8>, 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(&current, resource)?;
+        if &current_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: "<non-UTF-8 header>".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<Vec<u8>, 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<u64, RrdpFetchError> {
+        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<Vec<u8>, 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<u64, RrdpFetchError> {
+        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);
+    }
 }

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<usize, RrdpSyncError> {
+    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,
+            &notification_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() {

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<u8>,
+        notification_calls: AtomicUsize,
+        foreign_calls: AtomicUsize,
+    }
+    impl HttpFetcher for RecordingHttp {
+        fn fetch(&self, uri: &str) -> Result<Vec<u8>, 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<Vec<(String, Vec<u8>)>, 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");

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");

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<Self, RrdpError> {
+        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<Self, RrdpError> {
+        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<T> = Result<T, RrdpSyncError>;
 
 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<Vec<u8>, 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<u64, RrdpFetchError> {
+        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<NotificationSnapshot, R
     })
 }
 
+/// Validate every URI embedded in a notification before reading mutable local
+/// RRDP state or fetching a snapshot/delta. RFC 9674 §3.2 requires the
+/// notification, snapshot, and all delta URIs to share one origin.
+pub fn validate_notification_references(
+    notification_uri: &str,
+    notification: &Notification,
+) -> Result<RrdpOrigin, RrdpError> {
+    let notification_origin = RrdpOrigin::parse(notification_uri, RrdpResourceKind::Notification)?;
+    validate_reference_origin(
+        &notification_origin,
+        &notification.snapshot_uri,
+        RrdpResourceKind::Snapshot,
+    )?;
+    for delta in &notification.deltas {
+        validate_reference_origin(&notification_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<DeltaFile, RrdpError> {
     let doc = parse_rrdp_xml(xml)?;
     let root = doc.root_element();

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, &notif) {
+        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,
+                        &notification_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,
         &notif.snapshot_uri,
         &notif.snapshot_hash_sha256,
+        &notification_origin,
     ) {
         Ok(v) => {
             if let Some(t) = timing.as_ref() {

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,
+    Fetcher, RRDP_SNAPSHOT_APPLY_BATCH_SIZE, RRDP_XMLNS, RrdpError, RrdpFetchError, RrdpOrigin,
-    strip_all_ascii_whitespace,
+    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(RrdpFetchError::Fetch(e)) if e.contains(SNAPSHOT_NON_ASCII_ERROR) => {
-        Err(e) => return Err(RrdpSyncError::Fetch(e)),
+            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() {

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, &notif) {
+        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,
         &notif.snapshot_uri,
         &notif.snapshot_hash_sha256,
+        &notification_origin,
     ) {
         Ok(v) => {
             if let Some(t) = timing.as_ref() {

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)));

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<Vec<u8>, 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, &notification, &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, &notification, &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(&notification_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",
+        &notification,
+    )
+    .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");