use std::{
collections::{HashMap, HashSet},
time::{Duration, SystemTime, UNIX_EPOCH},
};
use ed25519_dalek::{Signature, Signer, SigningKey, Verifier, VerifyingKey};
use iroh::{PublicKey, SecretKey};
use iroh_blobs::Hash;
use iroh_gossip::{
TopicId,
api::{GossipSender, GossipTopic},
};
use serde::{Deserialize, Serialize};
use super::{
IndexingDatabase,
resilience::FetchFailureKind,
wot::{ProviderTrustAction, ProviderTrustRecord, TrustPolicy},
};
use crate::{
error::{Result, SyncwebError},
gossip::SignedGossipMessage,
node::gossip_service::GossipService,
};
const REPUTATION_SIGNAL_CONTEXT: &[u8] = b"syncweb/provider-trust/v1\0";
const TRUST_STREAM_TOPIC_SEED: &[u8] = b"syncweb/provider-trust-stream/v1";
const DEFAULT_DECAY_HALF_LIFE: Duration = Duration::from_hours(24);
const DEFAULT_TEMPORARY_BAN: Duration = Duration::from_hours(1);
const DEFAULT_BACKOFF_FACTOR: f64 = 2.0;
const DEFAULT_MAX_BAN: Duration = Duration::from_hours(24 * 30);
const DEFAULT_SIGNAL_TTL: Duration = Duration::from_hours(168);
const DEFAULT_SIGNAL_BATCH_SIZE: usize = 64;
const AUTO_BAN_FAILURE_THRESHOLD: u32 = 3;
#[derive(Clone, Debug, Serialize, Deserialize)]
#[non_exhaustive]
pub struct ProviderReputation {
pub provider: PublicKey,
pub total_fetches: u64,
pub successful_fetches: u64,
pub failed_fetches: u64,
pub consecutive_failures: u32,
pub last_success_at: Option<u64>,
pub last_failure_at: Option<u64>,
}
impl ProviderReputation {
#[must_use]
pub const fn new(provider: PublicKey) -> Self {
Self {
provider,
total_fetches: 0,
successful_fetches: 0,
failed_fetches: 0,
consecutive_failures: 0,
last_success_at: None,
last_failure_at: None,
}
}
pub const fn record_success(&mut self, now: u64) {
self.total_fetches = self.total_fetches.saturating_add(1);
self.successful_fetches = self.successful_fetches.saturating_add(1);
self.consecutive_failures = 0;
self.last_success_at = Some(now);
}
pub const fn record_failure(&mut self, _kind: FetchFailureKind, now: u64) {
self.total_fetches = self.total_fetches.saturating_add(1);
self.failed_fetches = self.failed_fetches.saturating_add(1);
self.consecutive_failures = self.consecutive_failures.saturating_add(1);
self.last_failure_at = Some(now);
}
#[must_use]
pub fn reliability_score(&self, now: u64) -> f64 {
self.reliability_score_with(now, DEFAULT_DECAY_HALF_LIFE, 2.0)
}
#[must_use]
pub fn is_reliable(&self, threshold: f64) -> bool {
self.reliability_score(current_epoch_seconds()) > threshold
}
#[must_use]
pub fn is_reliable_at(&self, threshold: f64, now: u64) -> bool {
self.reliability_score(now) > threshold
}
#[must_use]
pub const fn should_auto_ban(&self, consecutive_threshold: u32) -> bool {
self.consecutive_failures >= consecutive_threshold
}
fn reliability_score_with(&self, now: u64, half_life: Duration, failure_weight: f64) -> f64 {
if self.total_fetches == 0 {
return 0.5;
}
let failure_count = u32::try_from(self.failed_fetches.min(u64::from(u32::MAX))).unwrap_or(u32::MAX);
let success_count = u32::try_from(self.successful_fetches.min(u64::from(u32::MAX))).unwrap_or(u32::MAX);
let failures = std::ops::Mul::mul(failure_weight.max(0.0), f64::from(failure_count));
let denominator = std::ops::Add::add(f64::from(success_count), failures);
let base = if denominator == 0.0 {
0.5
} else {
std::ops::Div::div(f64::from(success_count), denominator)
};
let latest = self.last_success_at.max(self.last_failure_at).unwrap_or(now);
let age_seconds = now.saturating_sub(latest).min(u64::from(u32::MAX));
let age = f64::from(u32::try_from(age_seconds).unwrap_or(u32::MAX));
let half_life_seconds = half_life.as_secs_f64().max(1.0);
let decay = 0.5_f64.powf(std::ops::Div::div(age, half_life_seconds));
std::ops::Sub::sub(base, 0.5).mul_add(decay, 0.5).clamp(0.0, 1.0)
}
}
#[derive(Clone, Copy, Debug)]
#[non_exhaustive]
pub struct ReputationConfig {
pub min_samples: usize,
pub decay_half_life: Duration,
pub failure_weight: f64,
pub temporary_ban_duration: Duration,
pub auto_ban_backoff_factor: f64,
pub max_auto_ban_duration: Duration,
}
impl Default for ReputationConfig {
fn default() -> Self {
Self {
min_samples: 5,
decay_half_life: DEFAULT_DECAY_HALF_LIFE,
failure_weight: 2.0,
temporary_ban_duration: DEFAULT_TEMPORARY_BAN,
auto_ban_backoff_factor: DEFAULT_BACKOFF_FACTOR,
max_auto_ban_duration: DEFAULT_MAX_BAN,
}
}
}
#[derive(Clone, Debug)]
struct AutoBan {
until: u64,
count: u32,
}
#[derive(Clone, Debug)]
pub struct ProviderReputationStore {
reputations: HashMap<PublicKey, ProviderReputation>,
config: ReputationConfig,
policy: TrustPolicy,
auto_bans: HashMap<PublicKey, AutoBan>,
signal_sequences: HashMap<(PublicKey, PublicKey), u64>,
pending_signals: Vec<ProviderTrustSignal>,
max_signal_batch: usize,
reporter: Option<PublicKey>,
next_signal_sequence: u64,
database: Option<IndexingDatabase>,
}
impl Default for ProviderReputationStore {
fn default() -> Self {
Self::new(ReputationConfig::default())
}
}
impl ProviderReputationStore {
#[must_use]
pub fn new(config: ReputationConfig) -> Self {
Self {
reputations: HashMap::new(),
config,
policy: TrustPolicy::new(),
auto_bans: HashMap::new(),
signal_sequences: HashMap::new(),
pending_signals: Vec::new(),
max_signal_batch: DEFAULT_SIGNAL_BATCH_SIZE,
reporter: None,
next_signal_sequence: 1,
database: None,
}
}
pub fn with_database(database: IndexingDatabase, config: ReputationConfig) -> Result<Self> {
let reputations = database.load_all_reputations().unwrap_or_default();
let signal_sequences = database.load_signal_sequences().unwrap_or_default();
let auto_bans = Self::reconstruct_auto_bans(&database);
Ok(Self {
reputations,
config,
policy: TrustPolicy::new(),
auto_bans,
signal_sequences,
pending_signals: Vec::new(),
max_signal_batch: DEFAULT_SIGNAL_BATCH_SIZE,
reporter: None,
next_signal_sequence: 1,
database: Some(database),
})
}
fn reconstruct_auto_bans(database: &IndexingDatabase) -> HashMap<PublicKey, AutoBan> {
let now = SystemTime::now().duration_since(UNIX_EPOCH).map_or(0, |d| d.as_secs());
database
.load_auto_bans()
.unwrap_or_default()
.into_iter()
.filter_map(|(provider, (until, count))| {
until
.filter(|u| *u > now)
.map(|u| (provider, AutoBan { until: u, count }))
})
.collect()
}
#[must_use]
pub fn with_policy(config: ReputationConfig, policy: TrustPolicy) -> Self {
let mut store = Self::new(config);
store.policy = policy;
store
}
#[must_use]
pub const fn config(&self) -> &ReputationConfig {
&self.config
}
pub fn set_policy(&mut self, policy: TrustPolicy) {
self.policy = policy;
}
pub const fn set_reporter(&mut self, reporter: PublicKey) {
self.reporter = Some(reporter);
}
pub fn trust_reporter(&mut self, reporter: &PublicKey) -> Result<()> {
let verifying_key = VerifyingKey::from_bytes(reporter.as_bytes())
.map_err(|error| SyncwebError::InvalidIdentity(format!("invalid trust reporter: {error}")))?;
self.policy.trust_author(&verifying_key);
Ok(())
}
#[must_use]
pub fn reputation(&self, provider: PublicKey) -> ProviderReputation {
self.reputations
.get(&provider)
.cloned()
.unwrap_or_else(|| ProviderReputation::new(provider))
}
#[must_use]
pub fn score(&self, provider: PublicKey, now: u64) -> f64 {
let reputation = self.reputation(provider);
if usize::try_from(reputation.total_fetches).unwrap_or(usize::MAX) < self.config.min_samples {
0.5
} else {
reputation.reliability_score_with(now, self.config.decay_half_life, self.config.failure_weight)
}
}
pub fn record_fetch_result(&mut self, provider: PublicKey, success: bool, kind: FetchFailureKind, now: u64) {
let previous_score = self.score(provider, now);
{
let reputation = self
.reputations
.entry(provider)
.or_insert_with(|| ProviderReputation::new(provider));
if success {
reputation.record_success(now);
} else {
reputation.record_failure(kind, now);
if reputation.should_auto_ban(AUTO_BAN_FAILURE_THRESHOLD)
&& self.auto_bans.get(&provider).is_none_or(|ban| ban.until <= now)
{
self.apply_auto_ban(provider, now);
}
}
}
let current_score = self.score(provider, now);
let crossed_to_unreliable = previous_score >= 0.5 && current_score < 0.5;
let crossed_to_reliable = previous_score < 0.5 && current_score >= 0.5;
if (crossed_to_unreliable || crossed_to_reliable)
&& let Some(reporter) = self.reporter
{
let signal_kind = if crossed_to_reliable {
TrustSignalKind::ObservedSuccess
} else {
TrustSignalKind::ObservedFailure
};
let signal = ProviderTrustSignal {
provider,
signal: signal_kind,
hash: None,
reporter,
timestamp: now,
sequence: self.next_signal_sequence,
signature: None,
};
self.next_signal_sequence = self.next_signal_sequence.saturating_add(1);
self.enqueue_signal(signal);
}
if let Some(ref database) = self.database {
let auto_ban_until = self.auto_bans.get(&provider).map(|ban| ban.until);
let auto_ban_count = self.auto_bans.get(&provider).map_or(0, |ban| ban.count);
if let Some(rep) = self.reputations.get(&provider) {
let _ = database.upsert_reputation(&provider, rep, auto_ban_until, auto_ban_count);
}
}
}
pub fn record_success(&mut self, provider: PublicKey, now: u64) {
self.record_fetch_result(provider, true, FetchFailureKind::Unknown, now);
}
pub fn record_failure(&mut self, provider: PublicKey, kind: FetchFailureKind, now: u64) {
self.record_fetch_result(provider, false, kind, now);
}
#[must_use]
pub fn is_banned(&self, provider: PublicKey, now: u64) -> bool {
self.auto_bans.get(&provider).is_some_and(|ban| ban.until > now)
}
#[must_use]
pub fn auto_ban_until(&self, provider: PublicKey) -> Option<u64> {
self.auto_bans.get(&provider).map(|ban| ban.until)
}
#[must_use]
pub fn auto_ban_count(&self, provider: PublicKey) -> u32 {
self.auto_bans.get(&provider).map_or(0, |ban| ban.count)
}
pub fn rejoin(&mut self, provider: PublicKey, now: u64) -> bool {
self.auto_bans.get(&provider).is_some_and(|ban| ban.until <= now)
}
#[must_use]
pub fn should_skip_provider(&self, provider: PublicKey, now: u64, threshold: f64) -> bool {
self.is_banned(provider, now) || self.score(provider, now) < threshold
}
#[must_use]
pub fn rank_providers(&self, now: u64, hash: Hash) -> Vec<PublicKey> {
let providers = self.reputations.keys().copied().collect::<Vec<_>>();
self.rank_provider_list(now, hash, &providers)
}
#[must_use]
pub fn rank_provider_list(&self, now: u64, hash: Hash, providers: &[PublicKey]) -> Vec<PublicKey> {
let mut unique = HashSet::new();
let mut ranked = providers
.iter()
.copied()
.filter(|provider| unique.insert(*provider))
.map(|provider| (self.score(provider, now), provider))
.collect::<Vec<_>>();
ranked.sort_by(|left, right| {
right
.0
.total_cmp(&left.0)
.then_with(|| xor_distance(hash, right.1).cmp(&xor_distance(hash, left.1)))
.then_with(|| left.1.as_bytes().cmp(right.1.as_bytes()))
});
ranked.into_iter().map(|(_, provider)| provider).collect()
}
pub fn purge_stale(&mut self, now: u64, ttl: Duration) {
let ttl_seconds = ttl.as_secs();
self.reputations.retain(|_, reputation| {
let last = reputation.last_success_at.max(reputation.last_failure_at);
last.is_some_and(|timestamp| now.saturating_sub(timestamp) <= ttl_seconds)
});
self.auto_bans.retain(|_, ban| ban.until > now || ban.count > 0);
if let Some(ref database) = self.database {
let _ = database.delete_stale_reputations(now, ttl_seconds);
}
}
pub fn queue_trust_signal(&mut self, signal: ProviderTrustSignal) -> Result<bool> {
signal.validate()?;
Ok(self.enqueue_signal(signal))
}
fn enqueue_signal(&mut self, signal: ProviderTrustSignal) -> bool {
if self.pending_signals.iter().any(|existing| {
existing.provider == signal.provider
&& existing.reporter == signal.reporter
&& existing.signal == signal.signal
&& existing.hash == signal.hash
}) {
return false;
}
if self.pending_signals.len() >= self.max_signal_batch {
self.pending_signals.remove(0);
}
self.pending_signals.push(signal);
true
}
#[must_use]
pub fn pending_trust_signals(&self) -> &[ProviderTrustSignal] {
&self.pending_signals
}
pub fn take_pending_trust_signals(&mut self) -> Vec<ProviderTrustSignal> {
std::mem::take(&mut self.pending_signals)
}
pub fn set_max_signal_batch(&mut self, max_signal_batch: usize) {
self.max_signal_batch = max_signal_batch.max(1);
let keep_from = self.pending_signals.len().saturating_sub(self.max_signal_batch);
if keep_from > 0 {
self.pending_signals.drain(..keep_from);
}
}
pub fn ingest_trust_signal(&mut self, signal: ProviderTrustSignal) -> Result<bool> {
signal.verify()?;
let ProviderTrustSignal {
provider,
signal: kind,
hash,
reporter,
timestamp,
sequence,
signature,
} = signal;
drop(signature);
let reporter_id = hex::encode(reporter.as_bytes());
if !self.policy.is_trusted_for_at(&reporter_id, hash.as_ref(), timestamp) {
return Ok(false);
}
let sequence_key = (reporter, provider);
if self
.signal_sequences
.get(&sequence_key)
.is_some_and(|last_sequence| *last_sequence >= sequence)
{
return Ok(false);
}
self.signal_sequences.insert(sequence_key, sequence);
match kind {
TrustSignalKind::ObservedSuccess => {
self.record_fetch_result(provider, true, FetchFailureKind::Unknown, timestamp);
}
TrustSignalKind::ObservedFailure => {
self.record_fetch_result(provider, false, FetchFailureKind::Unknown, timestamp);
}
TrustSignalKind::ObservedCorruption => {
self.record_fetch_result(provider, false, FetchFailureKind::Corruption, timestamp);
}
}
Ok(true)
}
pub async fn publish_signal(
&self,
gossip: &GossipService,
sender: &GossipSender,
signal: &ProviderTrustSignal,
) -> Result<()> {
signal.verify()?;
gossip.publish(sender, signal.to_bytes()?).await
}
pub async fn subscribe_trust_stream(
&self,
gossip: &GossipService,
bootstrap: Vec<PublicKey>,
) -> Result<GossipTopic> {
gossip.subscribe(trust_stream_topic(), bootstrap).await
}
fn apply_auto_ban(&mut self, provider: PublicKey, now: u64) {
let previous_count = self.auto_bans.get(&provider).map_or(0, |ban| ban.count);
let count = previous_count.saturating_add(1);
let exponent = i32::try_from(count.saturating_sub(1)).unwrap_or(i32::MAX);
let scaled = std::ops::Mul::mul(
self.config.temporary_ban_duration.as_secs_f64(),
self.config.auto_ban_backoff_factor.max(1.0).powi(exponent),
);
let duration = Duration::from_secs_f64(scaled.min(self.config.max_auto_ban_duration.as_secs_f64()));
self.auto_bans.insert(
provider,
AutoBan {
until: now.saturating_add(duration.as_secs()),
count,
},
);
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[non_exhaustive]
pub struct ProviderTrustSignal {
pub provider: PublicKey,
pub signal: TrustSignalKind,
pub hash: Option<Hash>,
pub reporter: PublicKey,
pub timestamp: u64,
pub sequence: u64,
pub signature: Option<String>,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[non_exhaustive]
pub enum TrustSignalKind {
ObservedSuccess,
ObservedFailure,
ObservedCorruption,
}
impl ProviderTrustSignal {
pub fn new(
provider: PublicKey,
kind: TrustSignalKind,
hash: Option<Hash>,
reporter: PublicKey,
timestamp: u64,
sequence: u64,
) -> Result<Self> {
let signal = Self {
provider,
signal: kind,
hash,
reporter,
timestamp,
sequence,
signature: None,
};
signal.validate()?;
Ok(signal)
}
pub fn new_with_time(
provider: PublicKey,
kind: TrustSignalKind,
hash: Option<Hash>,
sequence: u64,
signing_key: &SigningKey,
) -> Result<Self> {
let reporter = PublicKey::from_bytes(&signing_key.verifying_key().to_bytes())
.map_err(|error| SyncwebError::InvalidIdentity(format!("invalid trust reporter: {error}")))?;
let mut created = Self::new(provider, kind, hash, reporter, current_epoch_seconds(), sequence)?;
created.sign(signing_key)?;
Ok(created)
}
pub fn from_trust_record(record: &ProviderTrustRecord, signing_key: &SigningKey) -> Result<Self> {
Self::new_with_time(
record.provider,
match record.action {
ProviderTrustAction::Vouch => TrustSignalKind::ObservedSuccess,
ProviderTrustAction::Distrust => TrustSignalKind::ObservedFailure,
ProviderTrustAction::Trust | ProviderTrustAction::Warn => {
return Err(SyncwebError::InvalidConfig(
"only vouch/distrust records can be converted to trust signals".into(),
));
}
},
record.scope,
record.sequence,
signing_key,
)
}
pub fn sign(&mut self, signing_key: &SigningKey) -> Result<()> {
if self.reporter.as_bytes() != signing_key.verifying_key().as_bytes() {
return Err(SyncwebError::InvalidIdentity(
"provider trust signal signer does not match reporter".to_owned(),
));
}
self.signature = Some(hex::encode(signing_key.sign(&self.unsigned_bytes()?).to_bytes()));
Ok(())
}
pub fn sign_with_secret_key(&mut self, secret_key: &SecretKey) -> Result<()> {
self.sign(&SigningKey::from_bytes(&secret_key.to_bytes()))
}
pub fn verify(&self) -> Result<()> {
self.verify_at(current_epoch_seconds())
}
pub fn verify_at(&self, now: u64) -> Result<()> {
self.validate()?;
if self.timestamp > now {
return Err(SyncwebError::InvalidConfig(
"provider trust signal is from the future".to_owned(),
));
}
if now.saturating_sub(self.timestamp) > DEFAULT_SIGNAL_TTL.as_secs() {
return Err(SyncwebError::InvalidConfig(
"provider trust signal has expired".to_owned(),
));
}
let signature_text = self
.signature
.as_deref()
.ok_or_else(|| SyncwebError::InvalidConfig("provider trust signal must contain a signature".to_owned()))?;
let signature_bytes = hex::decode(signature_text)
.map_err(|error| SyncwebError::InvalidConfig(format!("invalid provider trust signature: {error}")))?;
let signature = Signature::from_slice(&signature_bytes)
.map_err(|error| SyncwebError::InvalidConfig(format!("invalid provider trust signature: {error}")))?;
let key = VerifyingKey::from_bytes(self.reporter.as_bytes())
.map_err(|error| SyncwebError::InvalidIdentity(format!("invalid provider trust reporter: {error}")))?;
key.verify(&self.unsigned_bytes()?, &signature)
.map_err(|error| SyncwebError::InvalidConfig(format!("provider trust signature is invalid: {error}")))
}
pub fn validate(&self) -> Result<()> {
if self.sequence == 0 {
return Err(SyncwebError::InvalidConfig(
"provider trust signal sequence must be greater than zero".to_owned(),
));
}
let _provider = VerifyingKey::from_bytes(self.provider.as_bytes())
.map_err(|error| SyncwebError::InvalidIdentity(format!("invalid provider identity: {error}")))?;
let _reporter = VerifyingKey::from_bytes(self.reporter.as_bytes())
.map_err(|error| SyncwebError::InvalidIdentity(format!("invalid reporter identity: {error}")))?;
Ok(())
}
pub fn to_bytes(&self) -> Result<Vec<u8>> {
self.validate()?;
serde_json::to_vec(self)
.map_err(|error| SyncwebError::operation("failed to serialize provider trust signal", error))
}
pub fn from_bytes(bytes: impl AsRef<[u8]>) -> Result<Self> {
let signal: Self = serde_json::from_slice(bytes.as_ref())
.map_err(|error| SyncwebError::operation("failed to deserialize provider trust signal", error))?;
signal.validate()?;
Ok(signal)
}
fn unsigned_bytes(&self) -> Result<Vec<u8>> {
let mut unsigned = self.clone();
unsigned.signature = None;
let encoded = serde_json::to_vec(&unsigned)
.map_err(|error| SyncwebError::operation("failed to serialize unsigned provider trust signal", error))?;
let mut bytes = Vec::with_capacity(REPUTATION_SIGNAL_CONTEXT.len().saturating_add(encoded.len()));
bytes.extend_from_slice(REPUTATION_SIGNAL_CONTEXT);
bytes.extend_from_slice(&encoded);
Ok(bytes)
}
}
#[must_use]
pub fn trust_stream_topic() -> TopicId {
TopicId::from_bytes(*blake3::hash(TRUST_STREAM_TOPIC_SEED).as_bytes())
}
fn xor_distance(hash: Hash, provider: PublicKey) -> [u8; 32] {
let mut distance = [0_u8; 32];
for ((target, hash_byte), provider_byte) in distance.iter_mut().zip(hash.as_bytes()).zip(provider.as_bytes()) {
*target = *hash_byte ^ *provider_byte;
}
distance
}
impl SignedGossipMessage for ProviderTrustSignal {
fn verify_signature(&self) -> Result<()> {
self.verify()
}
}
fn current_epoch_seconds() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |duration| duration.as_secs())
}