use std::sync::Arc;
use async_trait::async_trait;
use dig_dht::ProviderRecord;
use dig_nat::{AvailabilityItem, AvailabilityResponse, RangeRequest};
use crate::error::DownloadError;
use crate::source::{FetchedRange, RangeTransport};
pub const MAX_HOP_PATH: usize = 8;
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum StreamRelayRefusal {
#[error("onion mode is disabled on this node")]
Disabled,
#[error("this node relays asks but not streams")]
AsksOnly,
#[error("the stream's hop budget could not be read")]
UnreadableHopBudget,
#[error("the stream's hop budget is exhausted")]
HopBudgetSpent,
#[error("the stream declared no readable length")]
UnreadableLength,
#[error("the stream is larger than this node will relay ({declared} > {ceiling} bytes)")]
StreamTooLarge {
declared: u64,
ceiling: u64,
},
#[error("the relay byte allowance is spent ({declared} needed, {available} left)")]
RelayByteBudgetSpent {
declared: u64,
available: u64,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StreamRelayDecision {
Carry {
hops_remaining: u8,
byte_ceiling: u64,
},
Refuse(StreamRelayRefusal),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct InboundStream {
pub hops_remaining: Option<u8>,
pub declared_len: Option<u64>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct StreamRelayConfig {
pub enabled: bool,
pub relays_asks_only: bool,
pub max_bytes_per_stream: u64,
}
pub const DEFAULT_MAX_BYTES_PER_STREAM: u64 = 16 * 1024 * 1024;
pub const DEFAULT_RELAY_BYTES_PER_WINDOW: u64 = 256 * 1024 * 1024;
impl Default for StreamRelayConfig {
fn default() -> Self {
StreamRelayConfig {
enabled: false,
relays_asks_only: true,
max_bytes_per_stream: DEFAULT_MAX_BYTES_PER_STREAM,
}
}
}
#[must_use]
pub fn decide_relay_stream(
config: &StreamRelayConfig,
inbound: &InboundStream,
relay_bytes_available: u64,
) -> StreamRelayDecision {
if !config.enabled {
return StreamRelayDecision::Refuse(StreamRelayRefusal::Disabled);
}
if config.relays_asks_only {
return StreamRelayDecision::Refuse(StreamRelayRefusal::AsksOnly);
}
let Some(hops_remaining) = inbound.hops_remaining else {
return StreamRelayDecision::Refuse(StreamRelayRefusal::UnreadableHopBudget);
};
if hops_remaining == 0 {
return StreamRelayDecision::Refuse(StreamRelayRefusal::HopBudgetSpent);
}
let Some(declared) = inbound.declared_len else {
return StreamRelayDecision::Refuse(StreamRelayRefusal::UnreadableLength);
};
if declared > config.max_bytes_per_stream {
return StreamRelayDecision::Refuse(StreamRelayRefusal::StreamTooLarge {
declared,
ceiling: config.max_bytes_per_stream,
});
}
if declared > relay_bytes_available {
return StreamRelayDecision::Refuse(StreamRelayRefusal::RelayByteBudgetSpent {
declared,
available: relay_bytes_available,
});
}
StreamRelayDecision::Carry {
hops_remaining: hops_remaining - 1,
byte_ceiling: declared,
}
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum HopPathError {
#[error("an onion hop path must contain at least one hop")]
Empty,
#[error("hop {0} appears more than once in the path")]
DuplicateHop(String),
#[error("an onion hop path may not exceed {MAX_HOP_PATH} hops (got {0})")]
TooLong(usize),
}
impl From<HopPathError> for DownloadError {
fn from(e: HopPathError) -> Self {
DownloadError::state(e)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HopPath {
hops: Vec<String>,
}
impl HopPath {
pub fn try_new(hops: Vec<String>) -> Result<Self, HopPathError> {
if hops.is_empty() {
return Err(HopPathError::Empty);
}
if hops.len() > MAX_HOP_PATH {
return Err(HopPathError::TooLong(hops.len()));
}
for (index, hop) in hops.iter().enumerate() {
if hops[..index].contains(hop) {
return Err(HopPathError::DuplicateHop(hop.clone()));
}
}
Ok(HopPath { hops })
}
#[must_use]
pub fn hops(&self) -> &[String] {
&self.hops
}
#[must_use]
pub fn len(&self) -> usize {
self.hops.len()
}
#[must_use]
pub fn is_empty(&self) -> bool {
false
}
}
#[async_trait]
pub trait OnionChannel: Send + Sync {
async fn ask_availability_through(
&self,
path: &HopPath,
provider: &ProviderRecord,
items: Vec<AvailabilityItem>,
) -> Result<AvailabilityResponse, DownloadError>;
async fn fetch_range_through(
&self,
path: &HopPath,
provider: &ProviderRecord,
req: &RangeRequest,
) -> Result<FetchedRange, DownloadError>;
}
pub struct OnionRangeTransport {
channel: Arc<dyn OnionChannel>,
path: HopPath,
config: StreamRelayConfig,
}
impl OnionRangeTransport {
#[must_use]
pub fn new(channel: Arc<dyn OnionChannel>, path: HopPath, config: StreamRelayConfig) -> Self {
OnionRangeTransport {
channel,
path,
config,
}
}
#[must_use]
pub fn path(&self) -> &HopPath {
&self.path
}
fn require_enabled(&self) -> Result<(), DownloadError> {
if self.config.enabled {
Ok(())
} else {
Err(DownloadError::state(StreamRelayRefusal::Disabled))
}
}
}
#[async_trait]
impl RangeTransport for OnionRangeTransport {
async fn query_availability(
&self,
provider: &ProviderRecord,
items: Vec<AvailabilityItem>,
) -> Result<AvailabilityResponse, DownloadError> {
self.require_enabled()?;
self.channel
.ask_availability_through(&self.path, provider, items)
.await
}
async fn fetch_range(
&self,
provider: &ProviderRecord,
req: &RangeRequest,
) -> Result<FetchedRange, DownloadError> {
self.require_enabled()?;
if req.length > self.config.max_bytes_per_stream {
return Err(DownloadError::state(StreamRelayRefusal::StreamTooLarge {
declared: req.length,
ceiling: self.config.max_bytes_per_stream,
}));
}
self.channel
.fetch_range_through(&self.path, provider, req)
.await
}
}
#[cfg(test)]
mod tests {
use super::*;
fn relaying() -> StreamRelayConfig {
StreamRelayConfig {
enabled: true,
relays_asks_only: false,
..Default::default()
}
}
fn inbound(hops: u8, len: u64) -> InboundStream {
InboundStream {
hops_remaining: Some(hops),
declared_len: Some(len),
}
}
#[test]
fn the_config_surface_declares_no_window_bound_it_cannot_enforce() {
let StreamRelayConfig {
enabled,
relays_asks_only,
max_bytes_per_stream,
} = StreamRelayConfig::default();
assert!(!enabled);
assert!(relays_asks_only);
assert_eq!(max_bytes_per_stream, DEFAULT_MAX_BYTES_PER_STREAM);
let mut caller_owned_window = DEFAULT_RELAY_BYTES_PER_WINDOW;
caller_owned_window -= 1024;
assert_eq!(
decide_relay_stream(&relaying(), &inbound(2, 1024), caller_owned_window),
StreamRelayDecision::Carry {
hops_remaining: 1,
byte_ceiling: 1024
},
"the bound that applies is the one the caller passed, never a configured field"
);
}
#[test]
fn a_node_carries_nothing_by_default() {
let config = StreamRelayConfig::default();
assert!(
!config.enabled,
"onion mode is off until an operator says so"
);
assert!(
config.relays_asks_only,
"and even switched on, streams are refused until an operator opts in"
);
assert_eq!(
decide_relay_stream(&config, &inbound(2, 1024), u64::MAX),
StreamRelayDecision::Refuse(StreamRelayRefusal::Disabled)
);
}
#[test]
fn a_hop_may_relay_asks_while_refusing_streams() {
let asks_only = StreamRelayConfig {
enabled: true,
relays_asks_only: true,
..Default::default()
};
assert_eq!(
decide_relay_stream(&asks_only, &inbound(2, 1024), u64::MAX),
StreamRelayDecision::Refuse(StreamRelayRefusal::AsksOnly),
"refusing to carry bytes is its own answer, distinguishable from being switched off"
);
assert_eq!(
decide_relay_stream(&relaying(), &inbound(2, 1024), u64::MAX),
StreamRelayDecision::Carry {
hops_remaining: 1,
byte_ceiling: 1024
}
);
}
#[test]
fn an_unreadable_declared_length_is_refused_not_carried_optimistically() {
let unreadable = InboundStream {
hops_remaining: Some(2),
declared_len: None,
};
assert_eq!(
decide_relay_stream(&relaying(), &unreadable, u64::MAX),
StreamRelayDecision::Refuse(StreamRelayRefusal::UnreadableLength)
);
}
#[test]
fn an_unreadable_hop_budget_is_refused() {
let unreadable = InboundStream {
hops_remaining: None,
declared_len: Some(1024),
};
assert_eq!(
decide_relay_stream(&relaying(), &unreadable, u64::MAX),
StreamRelayDecision::Refuse(StreamRelayRefusal::UnreadableHopBudget)
);
}
#[test]
fn an_exhausted_hop_budget_ends_the_path_here() {
assert_eq!(
decide_relay_stream(&relaying(), &inbound(0, 1024), u64::MAX),
StreamRelayDecision::Refuse(StreamRelayRefusal::HopBudgetSpent)
);
}
#[test]
fn the_per_stream_ceiling_is_pinned_from_both_sides() {
let config = StreamRelayConfig {
max_bytes_per_stream: 1_000,
..relaying()
};
assert_eq!(
decide_relay_stream(&config, &inbound(2, 1_000), u64::MAX),
StreamRelayDecision::Carry {
hops_remaining: 1,
byte_ceiling: 1_000
},
"at the ceiling exactly, the transfer is admitted"
);
assert_eq!(
decide_relay_stream(&config, &inbound(2, 1_001), u64::MAX),
StreamRelayDecision::Refuse(StreamRelayRefusal::StreamTooLarge {
declared: 1_001,
ceiling: 1_000
}),
"one byte over it, refused whole rather than truncated"
);
}
#[test]
fn the_window_allowance_is_pinned_from_both_sides() {
let config = StreamRelayConfig {
max_bytes_per_stream: 10_000,
..relaying()
};
assert_eq!(
decide_relay_stream(&config, &inbound(2, 500), 500),
StreamRelayDecision::Carry {
hops_remaining: 1,
byte_ceiling: 500
},
"a transfer that exactly exhausts the remaining allowance still fits in it"
);
assert_eq!(
decide_relay_stream(&config, &inbound(2, 501), 500),
StreamRelayDecision::Refuse(StreamRelayRefusal::RelayByteBudgetSpent {
declared: 501,
available: 500
}),
"one byte past it is refused — and named as an allowance, not as a size limit"
);
}
#[test]
fn a_refusal_is_never_an_absence_of_content() {
let reasons = [
decide_relay_stream(&StreamRelayConfig::default(), &inbound(2, 1), 0),
decide_relay_stream(&relaying(), &inbound(0, 1), 0),
decide_relay_stream(
&relaying(),
&InboundStream {
hops_remaining: Some(2),
declared_len: None,
},
0,
),
];
for reason in reasons {
let StreamRelayDecision::Refuse(refusal) = reason else {
panic!("expected a refusal, got {reason:?}");
};
assert!(
!refusal.to_string().is_empty(),
"a refusal states why this node would not carry the transfer"
);
}
}
#[test]
fn an_empty_hop_path_is_refused_because_the_privacy_loss_would_be_silent() {
assert_eq!(HopPath::try_new(Vec::new()), Err(HopPathError::Empty));
}
#[test]
fn a_peer_may_not_occupy_two_positions_on_one_path() {
let repeated = HopPath::try_new(vec!["a".into(), "b".into(), "a".into()]);
assert_eq!(repeated, Err(HopPathError::DuplicateHop("a".into())));
let distinct = HopPath::try_new(vec!["a".into(), "b".into(), "c".into()])
.expect("three distinct hops are a valid path");
assert_eq!(distinct.len(), 3);
assert_eq!(distinct.hops(), ["a", "b", "c"]);
}
#[test]
fn the_hop_path_length_bound_is_pinned_from_both_sides() {
let at_bound: Vec<String> = (0..MAX_HOP_PATH).map(|i| i.to_string()).collect();
assert!(HopPath::try_new(at_bound).is_ok(), "MAX_HOP_PATH hops fit");
let over: Vec<String> = (0..=MAX_HOP_PATH).map(|i| i.to_string()).collect();
assert_eq!(
HopPath::try_new(over),
Err(HopPathError::TooLong(MAX_HOP_PATH + 1))
);
}
}