feat(core): harden saved-device control plane against hostile peers

Bound pending offers, enforce pre-approval limits and max_saved_devices,
add per-identity cooldowns, silently reject invalid traffic, and redact
sensitive values from production events and errors without inventing
accepted-transfer quotas.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
2026-08-11 04:33:38 +02:00
parent a01d8328c9
commit 89c2206f0f
16 changed files with 1029 additions and 44 deletions

View File

@@ -288,10 +288,22 @@ pub struct CoreLimits {
pub max_metadata_bytes: u64,
pub max_events: u64,
pub max_pending_approvals: u64,
/// Incoming pairing offers awaiting the local user's decision.
/// Incoming pairing / targeted offers awaiting the local user's decision.
pub max_pending_offers: u64,
pub max_concurrent_transfers: u64,
pub event_queue_capacity: u64,
/// Cap on Saved + pending mutual-consent relationships (design §14).
pub max_saved_devices: u64,
/// Quiet period after a decline or repeated malformed control-plane traffic.
pub identity_cooldown_ms: u64,
/// Malformed control-plane messages from one identity before cooldown.
pub malformed_strike_limit: u64,
/// Pairing RPC / acknowledgement wait bound (milliseconds).
pub pairing_timeout_ms: u64,
/// Pre-approval offer decision wait bound (milliseconds).
pub offer_timeout_ms: u64,
/// Connection establishment bound for targeted transfers (milliseconds).
pub connection_timeout_ms: u64,
}
impl Default for CoreLimits {
@@ -313,6 +325,12 @@ impl Default for CoreLimits {
max_pending_offers: 16,
max_concurrent_transfers: 8,
event_queue_capacity: 1_024,
max_saved_devices: 256,
identity_cooldown_ms: 60_000,
malformed_strike_limit: 5,
pairing_timeout_ms: 15_000,
offer_timeout_ms: 120_000,
connection_timeout_ms: 30_000,
}
}
}
@@ -331,6 +349,12 @@ impl CoreLimits {
("max_pending_offers", self.max_pending_offers),
("max_concurrent_transfers", self.max_concurrent_transfers),
("event_queue_capacity", self.event_queue_capacity),
("max_saved_devices", self.max_saved_devices),
("identity_cooldown_ms", self.identity_cooldown_ms),
("malformed_strike_limit", self.malformed_strike_limit),
("pairing_timeout_ms", self.pairing_timeout_ms),
("offer_timeout_ms", self.offer_timeout_ms),
("connection_timeout_ms", self.connection_timeout_ms),
];
for (name, value) in positive {
if value == 0 {
@@ -342,6 +366,8 @@ impl CoreLimits {
("max_pending_offers", self.max_pending_offers),
("max_concurrent_transfers", self.max_concurrent_transfers),
("event_queue_capacity", self.event_queue_capacity),
("max_saved_devices", self.max_saved_devices),
("malformed_strike_limit", self.malformed_strike_limit),
] {
usize::try_from(value)
.with_context(|| format!("core limit {name} exceeds platform capacity"))?;

View File

@@ -0,0 +1,280 @@
//! Saved-device control-plane hardening (design §14).
//!
//! Bounds hostile / noisy peers without imposing quotas on transfers the
//! receiver has already accepted.
use std::{
collections::HashMap,
sync::{Arc, Mutex},
};
use data_encoding::HEXLOWER;
use serde_json::{Map, Value};
use crate::util::now_ms;
/// Per-identity quiet period after declines or repeated malformed traffic.
#[derive(Clone)]
pub(crate) struct IdentityCooldown {
inner: Arc<Mutex<CooldownInner>>,
cooldown_ms: i64,
strike_limit: u32,
}
#[derive(Default)]
struct CooldownInner {
until: HashMap<String, i64>,
strikes: HashMap<String, u32>,
}
impl IdentityCooldown {
pub(crate) fn new(cooldown_ms: u64, strike_limit: u64) -> Self {
Self {
inner: Arc::new(Mutex::new(CooldownInner::default())),
cooldown_ms: cooldown_ms as i64,
strike_limit: strike_limit as u32,
}
}
pub(crate) fn is_cooling(&self, identity: &str) -> bool {
let now = now_ms();
let mut state = self
.inner
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
state.until.retain(|_, until| *until > now);
state.until.contains_key(identity)
}
pub(crate) fn record_decline(&self, identity: &str) {
let until = now_ms() + self.cooldown_ms;
let mut state = self
.inner
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
state.until.insert(identity.to_string(), until);
state.strikes.remove(identity);
}
/// Count a malformed / spoofed / ineligible control-plane message.
///
/// Trips cooldown once the strike limit is reached. Returns whether the
/// identity is now cooling (including an already-active cooldown).
pub(crate) fn record_malformed(&self, identity: &str) -> bool {
if self.is_cooling(identity) {
return true;
}
let mut state = self
.inner
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let strikes = state.strikes.entry(identity.to_string()).or_insert(0);
*strikes = strikes.saturating_add(1);
if *strikes >= self.strike_limit {
state
.until
.insert(identity.to_string(), now_ms() + self.cooldown_ms);
state.strikes.remove(identity);
return true;
}
false
}
pub(crate) fn clear_strikes(&self, identity: &str) {
let mut state = self
.inner
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
state.strikes.remove(identity);
}
}
const REDACTED: &str = "[redacted]";
/// Keys whose values must never appear in production events / diagnostics.
fn is_sensitive_key(key: &str) -> bool {
matches!(
key,
"endpoint_id"
| "peer_endpoint_id"
| "sender_endpoint_id"
| "receiver_endpoint_id"
| "from_endpoint_id"
| "remote_endpoint_id"
| "local_endpoint_id"
| "ticket"
| "blob_ticket"
| "authorization"
| "capability"
| "secret"
| "grant"
| "grant_id"
| "proof"
| "mac"
| "filename"
| "file_name"
| "path"
| "display_name"
| "transfer_name"
| "sender_display_name"
| "remote_display_name"
| "address"
| "addrs"
| "relay_url"
| "relay_urls"
)
}
/// Stable fingerprint so diagnostics can correlate without leaking raw values.
pub(crate) fn fingerprint(value: &str) -> String {
let digest = blake3::hash(value.as_bytes());
let hex = HEXLOWER.encode(digest.as_bytes());
format!("<redacted:{}>", &hex[..8])
}
pub(crate) fn redact_json(value: Value) -> Value {
match value {
Value::Object(map) => Value::Object(redact_object(map)),
Value::Array(items) => Value::Array(items.into_iter().map(redact_json).collect()),
other => other,
}
}
fn redact_object(map: Map<String, Value>) -> Map<String, Value> {
map.into_iter()
.map(|(key, value)| {
if is_sensitive_key(&key) {
let redacted = match value {
Value::String(raw) if !raw.is_empty() => Value::String(fingerprint(&raw)),
Value::Null => Value::Null,
_ => Value::String(REDACTED.to_string()),
};
(key, redacted)
} else {
(key, redact_json(value))
}
})
.collect()
}
/// Scrub ticket-like and long opaque blobs from free-form error / log text.
pub(crate) fn redact_text(input: &str) -> String {
let mut out = String::with_capacity(input.len());
let mut rest = input;
while let Some(idx) = rest.find("vnd1:") {
out.push_str(&rest[..idx]);
out.push_str(REDACTED);
rest = &rest[idx + 5..];
// Skip the remainder of the ticket token (non-whitespace).
let end = rest
.find(|c: char| c.is_whitespace() || c == '"' || c == '\'')
.unwrap_or(rest.len());
rest = &rest[end..];
}
out.push_str(rest);
// Collapse long hex runs that look like endpoint ids / grant material.
collapse_long_hex(&out)
}
fn collapse_long_hex(input: &str) -> String {
let mut out = String::with_capacity(input.len());
let bytes = input.as_bytes();
let mut i = 0;
while i < bytes.len() {
if bytes[i].is_ascii_hexdigit() {
let start = i;
while i < bytes.len() && bytes[i].is_ascii_hexdigit() {
i += 1;
}
let len = i - start;
if len >= 32 {
out.push_str(REDACTED);
} else {
out.push_str(&input[start..i]);
}
} else {
out.push(bytes[i] as char);
i += 1;
}
}
out
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
/// Documented non-enforcement: accepted transfers have no per-device quota.
const ACCEPTED_TRANSFER_QUOTA_FIELDS: &[&str] = &[
"max_per_device_files",
"max_per_device_bytes",
"max_per_device_bandwidth",
"max_per_device_transfers",
];
#[test]
fn cooldown_trips_after_strike_limit_and_isolates_identities() {
let guard = IdentityCooldown::new(60_000, 3);
assert!(!guard.record_malformed("a"));
assert!(!guard.record_malformed("a"));
assert!(guard.record_malformed("a"));
assert!(guard.is_cooling("a"));
assert!(!guard.is_cooling("b"));
guard.record_decline("b");
assert!(guard.is_cooling("b"));
assert!(guard.is_cooling("a"));
}
#[test]
fn redaction_scrubs_sensitive_event_fields() {
let raw = json!({
"transfer_id": "ok-to-keep",
"sender_endpoint_id": "abc123endpointid000000000000000000000000000000000000000000000000",
"transfer_name": "secret.pdf",
"ticket": "vnd1:deadbeef",
"file_count": 2,
"nested": { "capability": [1, 2, 3], "state": "saved" }
});
let redacted = redact_json(raw);
let obj = redacted.as_object().unwrap();
assert_eq!(obj.get("transfer_id").unwrap(), "ok-to-keep");
assert_eq!(obj.get("file_count").unwrap(), 2);
let sender = obj.get("sender_endpoint_id").unwrap().as_str().unwrap();
assert!(sender.starts_with("<redacted:"));
assert!(!sender.contains("abc123"));
let name = obj.get("transfer_name").unwrap().as_str().unwrap();
assert!(name.starts_with("<redacted:"));
assert!(!name.contains("secret"));
assert!(obj
.get("ticket")
.unwrap()
.as_str()
.unwrap()
.starts_with("<redacted:"));
let nested = obj.get("nested").unwrap().as_object().unwrap();
assert_eq!(nested.get("capability").unwrap(), REDACTED);
assert_eq!(nested.get("state").unwrap(), "saved");
}
#[test]
fn redact_text_strips_tickets_and_long_hex() {
let text = "ticket vnd1:abcDEF123 and id 0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
let scrubbed = redact_text(text);
assert!(!scrubbed.contains("vnd1:"));
assert!(!scrubbed.contains("0123456789abcdef"));
assert!(scrubbed.contains(REDACTED));
}
#[test]
fn accepted_transfer_quota_fields_are_not_core_limits() {
// Control-plane hardening must not invent per-device accepted-transfer quotas.
let encoded = serde_json::to_string(&crate::api::CoreLimits::default()).unwrap();
for field in ACCEPTED_TRANSFER_QUOTA_FIELDS {
assert!(
!encoded.contains(field),
"CoreLimits must not enforce {field}"
);
}
}
}

View File

@@ -46,7 +46,6 @@ use crypto::{
};
const PENDING_TTL_MS: i64 = 30 * 60 * 1_000;
const PAIRING_RPC_TIMEOUT: Duration = Duration::from_secs(15);
#[derive(Clone)]
pub(crate) struct DeviceRelationshipService {
@@ -58,6 +57,8 @@ pub(crate) struct DeviceRelationshipService {
endpoint: Endpoint,
relay_mode: CoreRelayMode,
custom_relay_urls: Vec<RelayUrl>,
max_saved_devices: u64,
pairing_timeout: Duration,
peer_locks: Arc<TokioMutex<HashMap<String, Arc<TokioMutex<()>>>>>,
}
@@ -75,6 +76,8 @@ impl DeviceRelationshipService {
endpoint: Endpoint,
relay_mode: CoreRelayMode,
custom_relay_urls: Vec<RelayUrl>,
max_saved_devices: u64,
pairing_timeout_ms: u64,
) -> Self {
Self {
pool,
@@ -85,10 +88,36 @@ impl DeviceRelationshipService {
endpoint,
relay_mode,
custom_relay_urls,
max_saved_devices,
pairing_timeout: Duration::from_millis(pairing_timeout_ms),
peer_locks: Arc::new(TokioMutex::new(HashMap::new())),
}
}
/// Slots consumed by Saved or in-flight mutual-consent relationships.
async fn relationship_slots_used(&self) -> Result<u64, VnidropError> {
let row = sqlx::query(
r#"
SELECT COUNT(*) AS n FROM device_relationships
WHERE state IN ('saved', 'pending_outgoing', 'pending_incoming')
"#,
)
.fetch_one(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(row.get::<i64, _>("n") as u64)
}
async fn can_create_new_relationship(
&self,
peer_endpoint_id: &str,
) -> Result<bool, VnidropError> {
if self.find_row(peer_endpoint_id).await?.is_some() {
return Ok(true);
}
Ok(self.relationship_slots_used().await? < self.max_saved_devices)
}
pub(super) async fn lock_peer(&self, peer_endpoint_id: &str) -> Arc<TokioMutex<()>> {
let mut locks = self.peer_locks.lock().await;
locks
@@ -254,6 +283,10 @@ impl DeviceRelationshipService {
}
}
if !self.can_create_new_relationship(&peer_endpoint_id).await? {
return Ok(false);
}
let Some(taken) = self.eligibility.take_eligibility(&peer_endpoint_id).await? else {
return Ok(false);
};
@@ -316,7 +349,7 @@ impl DeviceRelationshipService {
let addr = self.peer_addr(&peer_endpoint_id).await?;
let client = RelationshipClient::connect(self.endpoint.clone(), addr);
let response = match tokio::time::timeout(
PAIRING_RPC_TIMEOUT,
self.pairing_timeout,
client.pairing_request(PairingRequest {
session_id: taken.session_id.clone(),
capability: taken.capability.to_vec(),
@@ -382,7 +415,7 @@ impl DeviceRelationshipService {
let addr = self.peer_addr(&peer_endpoint_id).await?;
let client = RelationshipClient::connect(self.endpoint.clone(), addr);
let _ = tokio::time::timeout(
PAIRING_RPC_TIMEOUT,
self.pairing_timeout,
client.pairing_consent(PairingConsent {
accepted: false,
grant: None,
@@ -413,7 +446,7 @@ impl DeviceRelationshipService {
let addr = self.peer_addr(&peer_endpoint_id).await?;
let client = RelationshipClient::connect(self.endpoint.clone(), addr);
let response = match tokio::time::timeout(
PAIRING_RPC_TIMEOUT,
self.pairing_timeout,
client.pairing_consent(PairingConsent {
accepted: true,
grant: Some(grant.clone()),
@@ -456,7 +489,7 @@ impl DeviceRelationshipService {
)
.await?;
match tokio::time::timeout(
PAIRING_RPC_TIMEOUT,
self.pairing_timeout,
client.pairing_ack(PairingAck {
possession_proof: proof,
challenge: ack_challenge.encode(),
@@ -542,6 +575,11 @@ impl DeviceRelationshipService {
return PairingRequestResponse::Rejected;
}
match self.can_create_new_relationship(&remote_endpoint_id).await {
Ok(true) => {}
Ok(false) | Err(_) => return PairingRequestResponse::Rejected,
}
let now = now_ms();
if self
.upsert_relationship(RelationshipUpsert {
@@ -726,7 +764,7 @@ impl DeviceRelationshipService {
let addr = self.peer_addr(peer_endpoint_id).await?;
let client = RelationshipClient::connect(self.endpoint.clone(), addr);
let response = match tokio::time::timeout(
PAIRING_RPC_TIMEOUT,
self.pairing_timeout,
client.pairing_consent(PairingConsent {
accepted: true,
grant: Some(grant),
@@ -767,7 +805,7 @@ impl DeviceRelationshipService {
.await?;
if let Ok(Ok(PairingAckResponse::Acknowledged | PairingAckResponse::AlreadySaved)) =
tokio::time::timeout(
PAIRING_RPC_TIMEOUT,
self.pairing_timeout,
client.pairing_ack(PairingAck {
possession_proof: proof,
challenge: ack_challenge.encode(),
@@ -1483,7 +1521,7 @@ impl DeviceRelationshipService {
};
let client = RelationshipClient::connect(self.endpoint.clone(), addr);
let _ = tokio::time::timeout(
PAIRING_RPC_TIMEOUT,
self.pairing_timeout,
client.revoke_notice(RevokeNotice {
generation,
issued_grant_id,

View File

@@ -55,7 +55,7 @@ impl VnidropError {
pub(crate) fn ticket(error: impl Into<anyhow::Error>) -> Self {
Self::Ticket {
reason: error.into().to_string(),
reason: crate::control_plane::redact_text(&error.into().to_string()),
}
}
@@ -168,7 +168,7 @@ impl VnidropError {
}
fn classify(error: anyhow::Error, fallback: impl FnOnce(String) -> Self) -> Self {
let reason = error.to_string();
let reason = crate::control_plane::redact_text(&error.to_string());
if let Some(existing) = error.chain().find_map(|cause| cause.downcast_ref::<Self>()) {
return existing.with_reason(reason);
}
@@ -193,7 +193,7 @@ impl VnidropError {
}
fn from_error(error: anyhow::Error, fallback: impl FnOnce(String) -> Self) -> Self {
let reason = error.to_string();
let reason = crate::control_plane::redact_text(&error.to_string());
if let Some(existing) = error.chain().find_map(|cause| cause.downcast_ref::<Self>()) {
existing.with_reason(reason)
} else {

View File

@@ -8,6 +8,7 @@ use tokio::{
use crate::{
api::{CoreEvent, CoreEventSink},
control_plane::redact_json,
repository::Repository,
transfer_state::TransferDirection,
util::now_ms,
@@ -51,6 +52,10 @@ enum EventPhase {
Pairing,
/// Prototype contact lifecycle notifications.
Contacts,
/// Prototype contact offer prompts.
Offer,
/// Saved-device targeted-transfer pre-approval prompts.
TargetedTransfer,
}
impl EventPhase {
@@ -74,6 +79,8 @@ impl EventPhase {
"delivery" => Some(Self::Delivery),
"pairing" => Some(Self::Pairing),
"contacts" => Some(Self::Contacts),
"offer" => Some(Self::Offer),
"targeted_transfer" => Some(Self::TargetedTransfer),
_ => None,
}
}
@@ -98,6 +105,8 @@ impl EventPhase {
Self::Delivery => "delivery",
Self::Pairing => "pairing",
Self::Contacts => "contacts",
Self::Offer => "offer",
Self::TargetedTransfer => "targeted_transfer",
}
}
}
@@ -245,6 +254,8 @@ impl EventHub {
// Compose observes this event synchronously, while SQLite persistence is
// serialized through the queue. That keeps the UI responsive without
// losing the ability to flush persisted history during shutdown/tests.
// Production diagnostics redact endpoint ids, tickets, grants, and paths;
// typed UniFFI list APIs still expose the values the UI needs.
let event = CoreEvent {
id,
timestamp,
@@ -253,7 +264,7 @@ impl EventHub {
direction: direction.map(|direction| direction.as_str().to_string()),
phase: phase.as_str().to_string(),
kind: kind.0,
data_json: data.to_string(),
data_json: redact_json(data).to_string(),
};
if let Err(error) = self.tx.try_send(EventCommand::Persist(event.clone())) {
tracing::warn!(event_id = %event.id, %error, "event persistence queue dropped event");

View File

@@ -2,6 +2,7 @@ mod access_policy;
mod api;
mod approval;
mod contacts;
mod control_plane;
mod device_relationship;
mod error;
mod event_hub;

View File

@@ -20,10 +20,6 @@ use crate::{event_hub::EventHub, offer::OfferResponse, util::now_ms};
/// How long the sender waits for the receiving user to decide.
const OFFER_WAIT_TIMEOUT: Duration = Duration::from_secs(120);
/// Quiet period after a decline, so a paired device cannot re-prompt on a loop.
/// A contact is not a stranger, but it is not unlimited either.
const DECLINE_COOLDOWN_MS: i64 = 60 * 1_000;
#[derive(Debug, Clone)]
pub(crate) struct PendingOffer {
pub(crate) offer_id: String,
@@ -50,16 +46,22 @@ pub(crate) struct OfferInbox {
/// Endpoint → time before which new offers are refused.
cooldowns: Arc<Mutex<HashMap<String, i64>>>,
max_pending: usize,
decline_cooldown_ms: i64,
}
impl OfferInbox {
pub(crate) fn new(event_hub: Arc<EventHub>, max_pending: usize) -> Self {
pub(crate) fn new(
event_hub: Arc<EventHub>,
max_pending: usize,
identity_cooldown_ms: u64,
) -> Self {
Self {
event_hub,
pending: Arc::new(Mutex::new(HashMap::new())),
waiters: Arc::new(Mutex::new(HashMap::new())),
cooldowns: Arc::new(Mutex::new(HashMap::new())),
max_pending,
decline_cooldown_ms: identity_cooldown_ms as i64,
}
}
@@ -229,7 +231,7 @@ impl OfferInbox {
if !accepted {
self.cooldowns.lock().await.insert(
offer.from_endpoint_id.clone(),
now_ms() + DECLINE_COOLDOWN_MS,
now_ms() + self.decline_cooldown_ms,
);
}
if let Some(waiter) = waiter {

View File

@@ -90,6 +90,22 @@ impl VnidropCore {
event_sink: Arc<dyn CoreEventSink>,
store: Arc<dyn crate::secure_secret::SecureSecretStore>,
network_config: CoreNetworkConfig,
) -> Result<Arc<Self>, VnidropError> {
Self::initialize_with_test_secret_store_limits_and_network(
app_data_dir,
event_sink,
store,
CoreLimits::default(),
network_config,
)
}
pub(crate) fn initialize_with_test_secret_store_limits_and_network(
app_data_dir: String,
event_sink: Arc<dyn CoreEventSink>,
store: Arc<dyn crate::secure_secret::SecureSecretStore>,
limits: CoreLimits,
network_config: CoreNetworkConfig,
) -> Result<Arc<Self>, VnidropError> {
let app_data_path = PathBuf::from(&app_data_dir);
std::fs::create_dir_all(&app_data_path).map_err(VnidropError::filesystem)?;
@@ -99,7 +115,7 @@ impl VnidropCore {
Self::initialize_with_identity_mode(
app_data_dir,
event_sink,
CoreLimits::default(),
limits,
network_config,
IdentityMode::Protected {
store,

View File

@@ -394,9 +394,21 @@ impl CoreInner {
{
tracing::warn!(%error, "failed to sweep dead grants");
}
let offers = OfferInbox::new(event_hub.clone(), limits.max_pending_offers as usize);
let targeted_offers =
TargetedOfferInbox::new(event_hub.clone(), limits.max_pending_offers as usize);
let offers = OfferInbox::new(
event_hub.clone(),
limits.max_pending_offers as usize,
limits.identity_cooldown_ms,
);
let identity_cooldown = crate::control_plane::IdentityCooldown::new(
limits.identity_cooldown_ms,
limits.malformed_strike_limit,
);
let targeted_offers = TargetedOfferInbox::new(
event_hub.clone(),
limits.max_pending_offers as usize,
identity_cooldown,
limits.offer_timeout_ms,
);
let device_relationships = Arc::new(DeviceRelationshipService::new(
repository.sqlite_pool(),
secret_custody.clone(),
@@ -406,6 +418,8 @@ impl CoreInner {
endpoint.clone(),
relay_mode,
relay_urls.clone(),
limits.max_saved_devices,
limits.pairing_timeout_ms,
));
let router = Router::builder(endpoint.clone())
.accept(iroh_blobs::ALPN, blobs)

View File

@@ -27,9 +27,15 @@ use crate::{
util::{non_empty, now_ms},
};
const OFFER_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
impl CoreInner {
fn connection_timeout(&self) -> std::time::Duration {
std::time::Duration::from_millis(self.limits.connection_timeout_ms)
}
fn offer_wait_timeout(&self) -> std::time::Duration {
std::time::Duration::from_millis(self.limits.offer_timeout_ms)
}
pub(super) fn targeted_store(&self) -> TargetedTransferStore {
TargetedTransferStore::new(self.repository.sqlite_pool())
}
@@ -131,7 +137,7 @@ impl CoreInner {
{
let client = TargetedTransferProtocol::client(self.endpoint.clone(), addr);
let _ = tokio::time::timeout(
OFFER_CONNECT_TIMEOUT,
self.connection_timeout(),
client.cancel_offer(CancelTargetedOffer {
transfer_id: id.clone(),
}),
@@ -287,7 +293,8 @@ impl CoreInner {
.await?;
let client = TargetedTransferProtocol::client(self.endpoint.clone(), addr);
let challenge =
match tokio::time::timeout(OFFER_CONNECT_TIMEOUT, client.request_challenge()).await {
match tokio::time::timeout(self.connection_timeout(), client.request_challenge()).await
{
Ok(Ok(challenge)) => challenge,
Ok(Err(error)) => {
let _ = store
@@ -331,7 +338,7 @@ impl CoreInner {
.await?;
let response = match tokio::time::timeout(
OFFER_CONNECT_TIMEOUT + std::time::Duration::from_secs(120),
self.connection_timeout() + self.offer_wait_timeout(),
client.submit_offer(SubmitTargetedOffer {
proof,
generation,

View File

@@ -10,9 +10,7 @@ use serde_json::json;
use tokio::sync::{watch, Mutex};
use uuid::Uuid;
use crate::{api::PendingTargetedOffer, event_hub::EventHub};
const OFFER_WAIT_TIMEOUT: Duration = Duration::from_secs(120);
use crate::{api::PendingTargetedOffer, control_plane::IdentityCooldown, event_hub::EventHub};
#[derive(Debug, Clone)]
pub(crate) struct PendingTargetedOfferRecord {
@@ -40,27 +38,45 @@ pub(crate) struct TargetedOfferInbox {
decisions: Arc<Mutex<HashMap<String, PendingWaiter>>>,
auths: Arc<Mutex<HashMap<String, AuthWaiter>>>,
settled: Arc<Mutex<HashMap<String, SettledOfferResult>>>,
cooldown: IdentityCooldown,
max_pending: usize,
offer_timeout: Duration,
}
impl TargetedOfferInbox {
pub(crate) fn new(event_hub: Arc<EventHub>, max_pending: usize) -> Self {
pub(crate) fn new(
event_hub: Arc<EventHub>,
max_pending: usize,
cooldown: IdentityCooldown,
offer_timeout_ms: u64,
) -> Self {
Self {
event_hub,
pending: Arc::new(Mutex::new(HashMap::new())),
decisions: Arc::new(Mutex::new(HashMap::new())),
auths: Arc::new(Mutex::new(HashMap::new())),
settled: Arc::new(Mutex::new(HashMap::new())),
cooldown,
max_pending,
offer_timeout: Duration::from_millis(offer_timeout_ms),
}
}
pub(crate) fn cooldown(&self) -> &IdentityCooldown {
&self.cooldown
}
/// Surface a validated offer and block until the local user decides.
///
/// Replaying the same transfer identity returns the settled result or joins
/// the existing pending wait — never a second prompt.
pub(crate) async fn submit(&self, offer: PendingTargetedOffer) -> TargetedOfferDecision {
let transfer_id = offer.transfer_id.clone();
if self.cooldown.is_cooling(&offer.sender_endpoint_id) {
return TargetedOfferDecision::Refused {
reason: "identity-cooldown".to_string(),
};
}
if let Some(settled) = self.settled.lock().await.get(&transfer_id).cloned() {
return settled_to_decision(settled);
}
@@ -76,11 +92,7 @@ impl TargetedOfferInbox {
reason: "immutable-transfer-mismatch".to_string(),
};
}
if pending.len() >= self.max_pending {
return TargetedOfferDecision::Refused {
reason: "too-many-pending-offers".to_string(),
};
}
// Prefer the more specific per-sender refusal before the global bound.
if pending
.values()
.any(|entry| entry.offer.sender_endpoint_id == offer.sender_endpoint_id)
@@ -89,6 +101,11 @@ impl TargetedOfferInbox {
reason: "offer-already-pending".to_string(),
};
}
if pending.len() >= self.max_pending {
return TargetedOfferDecision::Refused {
reason: "too-many-pending-offers".to_string(),
};
}
}
let (decision_tx, _decision_rx) = watch::channel(None);
@@ -99,6 +116,19 @@ impl TargetedOfferInbox {
drop(pending);
return self.wait_existing_decision(&transfer_id).await;
}
if pending
.values()
.any(|entry| entry.offer.sender_endpoint_id == offer.sender_endpoint_id)
{
return TargetedOfferDecision::Refused {
reason: "offer-already-pending".to_string(),
};
}
if pending.len() >= self.max_pending {
return TargetedOfferDecision::Refused {
reason: "too-many-pending-offers".to_string(),
};
}
pending.insert(
transfer_id.clone(),
PendingTargetedOfferRecord {
@@ -112,6 +142,7 @@ impl TargetedOfferInbox {
decision: decision_tx,
},
);
self.cooldown.clear_strikes(&offer.sender_endpoint_id);
self.event_hub.emit_endpoint(
"targeted_transfer",
@@ -153,7 +184,7 @@ impl TargetedOfferInbox {
}
};
match tokio::time::timeout(OFFER_WAIT_TIMEOUT, wait).await {
match tokio::time::timeout(self.offer_timeout, wait).await {
Ok(true) => TargetedOfferDecision::Accepted,
Ok(false) => {
self.discard(transfer_id).await;
@@ -206,10 +237,15 @@ impl TargetedOfferInbox {
return Ok(Some(auth));
}
let exists = self.pending.lock().await.contains_key(transfer_id);
if !exists {
let sender_endpoint_id = {
let pending = self.pending.lock().await;
pending
.get(transfer_id)
.map(|entry| entry.offer.sender_endpoint_id.clone())
};
let Some(sender_endpoint_id) = sender_endpoint_id else {
return Err(RespondError::Unknown);
}
};
let waiter = {
let decisions = self.decisions.lock().await;
decisions
@@ -222,6 +258,7 @@ impl TargetedOfferInbox {
if !accepted {
let _ = decision_tx.send(Some(false));
self.discard(transfer_id).await;
self.cooldown.record_decline(&sender_endpoint_id);
self.settled.lock().await.insert(
transfer_id.to_string(),
SettledOfferResult::Declined {
@@ -258,7 +295,7 @@ impl TargetedOfferInbox {
}
};
match tokio::time::timeout(OFFER_WAIT_TIMEOUT, wait_auth).await {
match tokio::time::timeout(self.offer_timeout, wait_auth).await {
Ok(Ok(auth)) => {
self.pending.lock().await.remove(transfer_id);
self.auths.lock().await.remove(transfer_id);

View File

@@ -89,17 +89,25 @@ impl TargetedTransferProtocol {
offer: SubmitTargetedOffer,
) -> TargetedOfferResponse {
let expected = experimental_saved_device_capabilities().targeted_transfer_protocol_version;
if self.inbox.cooldown().is_cooling(remote_endpoint_id) {
return TargetedOfferResponse::Refused {
reason: "identity-cooldown".to_string(),
};
}
if offer.protocol_version != expected {
self.inbox.cooldown().record_malformed(remote_endpoint_id);
return TargetedOfferResponse::Refused {
reason: "protocol-incompatible".to_string(),
};
}
if offer.receiver_endpoint_id != self.local_endpoint_id {
self.inbox.cooldown().record_malformed(remote_endpoint_id);
return TargetedOfferResponse::Refused {
reason: "receiver-mismatch".to_string(),
};
}
if offer.sender_endpoint_id != remote_endpoint_id {
self.inbox.cooldown().record_malformed(remote_endpoint_id);
return TargetedOfferResponse::Refused {
reason: "sender-mismatch".to_string(),
};
@@ -112,6 +120,7 @@ impl TargetedTransferProtocol {
|| offer.manifest_id.is_empty()
|| offer.content_hash.is_empty()
{
self.inbox.cooldown().record_malformed(remote_endpoint_id);
return TargetedOfferResponse::Refused {
reason: "manifest-limits".to_string(),
};
@@ -120,6 +129,7 @@ impl TargetedTransferProtocol {
.limits
.validate_metadata_text("transfer name", Some(offer.transfer_name.as_str()))
{
self.inbox.cooldown().record_malformed(remote_endpoint_id);
return TargetedOfferResponse::Refused {
reason: error.to_string(),
};
@@ -190,7 +200,8 @@ impl TargetedTransferProtocol {
)
.await
{
tracing::debug!(%error, "targeted offer relationship proof rejected");
tracing::debug!(error = %error, "targeted offer relationship proof rejected");
self.inbox.cooldown().record_malformed(remote_endpoint_id);
if matches!(error, VnidropError::ProtocolIncompatible { .. }) {
return TargetedOfferResponse::Refused {
reason: "protocol-incompatible".to_string(),

View File

@@ -4,6 +4,8 @@ mod access_policy_tests;
mod contact_polling_tests;
#[path = "tests/contacts.rs"]
mod contacts_tests;
#[path = "tests/control_plane.rs"]
mod control_plane_tests;
#[path = "tests/device_relationship.rs"]
mod device_relationship_tests;
#[path = "tests/error.rs"]

View File

@@ -0,0 +1,527 @@
//! Control-plane hardening: offer bounds, cooldowns, saved-device cap, redaction.
use std::sync::{Arc, Mutex};
use crate::{
api::{CoreEvent, CoreEventSink, CoreLimits, PendingTargetedOffer},
control_plane::IdentityCooldown,
event_hub::EventHub,
repository::Repository,
secure_secret::FaultInjectingSecretStore,
targeted_transfer::inbox::{TargetedOfferDecision, TargetedOfferInbox},
CoreNetworkConfig, DeviceRelationshipState, ShareMetadataInput, ShareSource, SourceKind,
TransferAccessMode, VnidropCore, VnidropError,
};
struct RecordingSink {
events: Mutex<Vec<CoreEvent>>,
}
impl CoreEventSink for RecordingSink {
fn on_event(&self, event: CoreEvent) {
self.events.lock().unwrap().push(event);
}
}
impl RecordingSink {
fn events(&self) -> Vec<CoreEvent> {
self.events.lock().unwrap().clone()
}
fn kinds(&self) -> Vec<String> {
self.events().into_iter().map(|event| event.kind).collect()
}
}
fn sample_offer(transfer_id: &str, sender: &str) -> PendingTargetedOffer {
PendingTargetedOffer {
transfer_id: transfer_id.to_string(),
sender_endpoint_id: sender.to_string(),
receiver_endpoint_id: "receiver".to_string(),
manifest_id: "manifest".to_string(),
content_hash: "hash".to_string(),
transfer_name: "secret-name.pdf".to_string(),
file_count: 1,
total_size: 12,
protocol_version: 1,
received_at: 1,
}
}
async fn inbox_with_limits(
max_pending: usize,
cooldown_ms: u64,
strikes: u64,
) -> (TargetedOfferInbox, Arc<RecordingSink>) {
let temp = tempfile::tempdir().unwrap();
let repository = Repository::open(temp.path()).await.unwrap();
let sink = Arc::new(RecordingSink {
events: Mutex::new(Vec::new()),
});
let hub = Arc::new(EventHub::start(repository, sink.clone(), 64, 100));
let cooldown = IdentityCooldown::new(cooldown_ms, strikes);
let inbox = TargetedOfferInbox::new(hub, max_pending, cooldown, 5_000);
// Keep temp dir alive for the hub's repository by leaking — tests are short-lived.
std::mem::forget(temp);
(inbox, sink)
}
#[tokio::test]
async fn one_unresolved_offer_per_sender_and_global_queue_bound() {
let (inbox, sink) = inbox_with_limits(1, 60_000, 5).await;
let first = sample_offer("t1", "sender-a");
let submit_first = {
let inbox = inbox.clone();
tokio::spawn(async move { inbox.submit(first).await })
};
// Wait until the prompt is live.
let started = std::time::Instant::now();
loop {
if !inbox.list().await.is_empty() {
break;
}
assert!(started.elapsed() < std::time::Duration::from_secs(2));
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
}
assert!(sink.kinds().contains(&"offer-received".to_string()));
let same_sender = inbox.submit(sample_offer("t2", "sender-a")).await;
assert_eq!(
same_sender,
TargetedOfferDecision::Refused {
reason: "offer-already-pending".to_string()
}
);
let other_sender = inbox.submit(sample_offer("t3", "sender-b")).await;
assert_eq!(
other_sender,
TargetedOfferDecision::Refused {
reason: "too-many-pending-offers".to_string()
}
);
assert_eq!(inbox.list().await.len(), 1);
// Excess rejects never emit a second prompt.
assert_eq!(
sink.kinds()
.into_iter()
.filter(|kind| kind == "offer-received")
.count(),
1
);
inbox.respond("t1", false).await.unwrap();
let _ = submit_first.await.unwrap();
}
#[tokio::test]
async fn decline_cools_sender_without_affecting_unrelated_devices() {
let (inbox, _sink) = inbox_with_limits(8, 60_000, 5).await;
let offer = sample_offer("decline-1", "noisy");
let wait = {
let inbox = inbox.clone();
tokio::spawn(async move { inbox.submit(offer).await })
};
let started = std::time::Instant::now();
loop {
if inbox.get_pending("decline-1").await.is_some() {
break;
}
assert!(started.elapsed() < std::time::Duration::from_secs(2));
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
}
inbox.respond("decline-1", false).await.unwrap();
let _ = wait.await.unwrap();
let cooled = inbox.submit(sample_offer("decline-2", "noisy")).await;
assert_eq!(
cooled,
TargetedOfferDecision::Refused {
reason: "identity-cooldown".to_string()
}
);
assert!(inbox.list().await.is_empty());
let unrelated = {
let inbox = inbox.clone();
tokio::spawn(async move { inbox.submit(sample_offer("ok", "friend")).await })
};
let started = std::time::Instant::now();
loop {
if inbox.get_pending("ok").await.is_some() {
break;
}
assert!(started.elapsed() < std::time::Duration::from_secs(2));
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
}
inbox.respond("ok", false).await.unwrap();
let _ = unrelated.await.unwrap();
}
#[tokio::test]
async fn malformed_strikes_trip_cooldown() {
let cooldown = IdentityCooldown::new(60_000, 2);
assert!(!cooldown.record_malformed("attacker"));
assert!(cooldown.record_malformed("attacker"));
assert!(cooldown.is_cooling("attacker"));
assert!(!cooldown.is_cooling("bystander"));
}
#[tokio::test]
async fn offer_received_events_redact_endpoint_ids_and_names() {
let (inbox, sink) = inbox_with_limits(4, 60_000, 5).await;
let sender = "abc123endpointid000000000000000000000000000000000000000000000000";
let wait = {
let inbox = inbox.clone();
let offer = sample_offer("redact-1", sender);
tokio::spawn(async move { inbox.submit(offer).await })
};
let started = std::time::Instant::now();
loop {
if sink
.kinds()
.into_iter()
.any(|kind| kind == "offer-received")
{
break;
}
assert!(started.elapsed() < std::time::Duration::from_secs(2));
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
}
let event = sink
.events()
.into_iter()
.find(|event| event.kind == "offer-received")
.expect("offer event");
assert!(!event.data_json.contains(sender));
assert!(!event.data_json.contains("secret-name"));
assert!(event.data_json.contains("redacted"));
inbox.respond("redact-1", false).await.unwrap();
let _ = wait.await.unwrap();
}
struct ProtectedNode {
_data_dir: tempfile::TempDir,
_secret_store: Arc<FaultInjectingSecretStore>,
_limits: CoreLimits,
_network_config: CoreNetworkConfig,
sink: Arc<RecordingSink>,
core: Option<Arc<VnidropCore>>,
}
impl ProtectedNode {
fn with_limits(limits: CoreLimits) -> Self {
let data_dir = tempfile::tempdir().unwrap();
let sink = Arc::new(RecordingSink {
events: Mutex::new(Vec::new()),
});
let store = Arc::new(FaultInjectingSecretStore::default());
let network_config = CoreNetworkConfig::default();
let core = VnidropCore::initialize_with_test_secret_store_limits_and_network(
data_dir.path().to_string_lossy().into_owned(),
sink.clone(),
store.clone(),
limits.clone(),
network_config.clone(),
)
.expect("protected test core");
Self {
_data_dir: data_dir,
_secret_store: store,
_limits: limits,
_network_config: network_config,
sink,
core: Some(core),
}
}
fn core(&self) -> Arc<VnidropCore> {
self.core.as_ref().expect("core alive").clone()
}
}
impl Drop for ProtectedNode {
fn drop(&mut self) {
if let Some(core) = self.core.take() {
core.shutdown();
}
}
}
fn share_path(
core: &VnidropCore,
source: &std::path::Path,
transfer_id: u64,
) -> crate::ShareResult {
core.share_files(
vec![ShareSource {
kind: SourceKind::Path,
value: source.to_string_lossy().into_owned(),
display_name: Some("hello.txt".to_string()),
is_directory: false,
}],
ShareMetadataInput {
transfer_id,
transfer_name: Some("hello.txt".to_string()),
sender_name: Some("sender".to_string()),
access_mode: TransferAccessMode::ApprovalRequired,
},
)
.unwrap()
}
fn wait_for_receiver_request(sender: &VnidropCore, transfer_id: u64) -> crate::ReceiverRequest {
let started = std::time::Instant::now();
loop {
if let Some(request) = sender
.list_receiver_requests(transfer_id)
.unwrap()
.into_iter()
.find(|request| request.status == "requested")
{
return request;
}
assert!(
started.elapsed() < std::time::Duration::from_secs(15),
"timed out waiting for receiver request"
);
std::thread::sleep(std::time::Duration::from_millis(25));
}
}
fn complete_transfer(sender: &ProtectedNode, receiver: &ProtectedNode, transfer_id: u64) {
let source_dir = tempfile::tempdir().unwrap();
let output_dir = tempfile::tempdir().unwrap();
let source_path = source_dir.path().join("hello.txt");
std::fs::write(&source_path, b"mutual consent").unwrap();
let share = share_path(&sender.core(), &source_path, transfer_id);
let output_dir = output_dir.path().to_string_lossy().to_string();
let receiver_core = receiver.core().clone();
let ticket = share.ticket.clone();
let handle = std::thread::spawn(move || {
receiver_core.receive(ticket, output_dir, Some("receiver".to_string()))
});
let request = wait_for_receiver_request(&sender.core(), share.transfer_id);
sender
.core()
.respond_receiver_request(request.id, true, None)
.unwrap();
handle.join().unwrap().unwrap();
let started = std::time::Instant::now();
let peer = receiver.core().status().endpoint_id.clone();
loop {
if sender
.core()
.list_pairing_eligibilities()
.unwrap()
.iter()
.any(|entry| entry.peer_endpoint_id == peer)
{
break;
}
assert!(
started.elapsed() < std::time::Duration::from_secs(15),
"eligibility never appeared"
);
std::thread::sleep(std::time::Duration::from_millis(25));
}
}
fn establish_saved(alice: &ProtectedNode, bob: &ProtectedNode, transfer_id: u64) {
complete_transfer(alice, bob, transfer_id);
let bob_id = bob.core().status().endpoint_id.clone();
assert!(alice
.core()
.request_saved_device_pairing(bob_id.clone())
.unwrap());
let started = std::time::Instant::now();
loop {
let pending = bob
.core()
.list_device_relationships()
.unwrap()
.into_iter()
.find(|entry| {
entry.remote_endpoint_id == alice.core().status().endpoint_id
&& entry.state == DeviceRelationshipState::PendingIncoming
});
if pending.is_some() {
break;
}
assert!(
started.elapsed() < std::time::Duration::from_secs(15),
"pairing prompt never arrived"
);
std::thread::sleep(std::time::Duration::from_millis(25));
}
assert!(bob
.core()
.respond_to_device_pairing(alice.core().status().endpoint_id.clone(), true)
.unwrap());
let started = std::time::Instant::now();
loop {
if alice.core().list_saved_devices().unwrap().len() == 1
&& bob.core().list_saved_devices().unwrap().len() == 1
{
break;
}
assert!(
started.elapsed() < std::time::Duration::from_secs(15),
"saved relationship never activated"
);
std::thread::sleep(std::time::Duration::from_millis(25));
}
}
#[test]
fn default_limits_include_saved_device_cap_and_control_plane_timeouts() {
let limits = CoreLimits::default();
limits.validate().unwrap();
assert_eq!(limits.max_saved_devices, 256);
assert!(limits.identity_cooldown_ms > 0);
assert!(limits.malformed_strike_limit > 0);
assert!(limits.pairing_timeout_ms > 0);
assert!(limits.offer_timeout_ms > 0);
assert!(limits.connection_timeout_ms > 0);
assert!(
limits.max_pending_offers <= 64,
"pending offers stay tightly bounded"
);
}
#[test]
fn saved_device_cap_blocks_only_new_relationships() {
let tight = CoreLimits {
max_saved_devices: 1,
..CoreLimits::default()
};
let alice = ProtectedNode::with_limits(tight.clone());
let bob = ProtectedNode::with_limits(tight.clone());
let carol = ProtectedNode::with_limits(tight);
establish_saved(&alice, &bob, 13_001);
assert_eq!(alice.core().list_saved_devices().unwrap().len(), 1);
// Existing relationship remains usable for targeted transfer.
let source_dir = tempfile::tempdir().unwrap();
let source_path = source_dir.path().join("payload.txt");
std::fs::write(&source_path, b"still works").unwrap();
let bob_id = bob.core().status().endpoint_id.clone();
let bob_core = bob.core().clone();
let accept = std::thread::spawn(move || {
let started = std::time::Instant::now();
let offer = loop {
if let Some(offer) = bob_core.list_pending_targeted_offers().into_iter().next() {
break offer;
}
assert!(
started.elapsed() < std::time::Duration::from_secs(20),
"offer never arrived for existing saved peer"
);
std::thread::sleep(std::time::Duration::from_millis(25));
};
bob_core
.respond_to_targeted_offer(offer.transfer_id, true)
.unwrap()
});
alice
.core()
.create_targeted_transfer(
bob_id,
vec![ShareSource {
kind: SourceKind::Path,
value: source_path.to_string_lossy().into_owned(),
display_name: Some("payload.txt".to_string()),
is_directory: false,
}],
Some("payload.txt".to_string()),
)
.unwrap();
accept.join().unwrap().unwrap();
// New relationship is refused while the cap is full.
complete_transfer(&alice, &carol, 13_002);
let carol_id = carol.core().status().endpoint_id.clone();
assert!(
!alice.core().request_saved_device_pairing(carol_id).unwrap(),
"cap must block only new relationships"
);
assert!(alice.core().list_saved_devices().unwrap().len() <= 1);
assert!(carol.core().list_saved_devices().unwrap().is_empty());
}
#[test]
fn silent_reject_keeps_invalid_offers_off_the_prompt_surface() {
let alice = ProtectedNode::with_limits(CoreLimits::default());
let bob = ProtectedNode::with_limits(CoreLimits::default());
// No Saved relationship — create must fail before any receiver prompt.
let bob_id = bob.core().status().endpoint_id.clone();
let source_dir = tempfile::tempdir().unwrap();
let source_path = source_dir.path().join("payload.txt");
std::fs::write(&source_path, b"nope").unwrap();
let err = alice
.core()
.create_targeted_transfer(
bob_id,
vec![ShareSource {
kind: SourceKind::Path,
value: source_path.to_string_lossy().into_owned(),
display_name: Some("payload.txt".to_string()),
is_directory: false,
}],
Some("payload.txt".to_string()),
)
.unwrap_err();
assert!(matches!(err, VnidropError::Permission { .. }));
assert!(bob.core().list_pending_targeted_offers().is_empty());
assert!(
!bob.sink
.kinds()
.into_iter()
.any(|kind| kind == "offer-received"),
"silent reject must not emit offer prompts"
);
}
#[test]
fn blocked_peer_cannot_create_pairing_prompt() {
let alice = ProtectedNode::with_limits(CoreLimits::default());
let bob = ProtectedNode::with_limits(CoreLimits::default());
complete_transfer(&alice, &bob, 13_010);
let alice_id = alice.core().status().endpoint_id.clone();
bob.core().block_device(alice_id.clone()).unwrap();
assert!(
!alice
.core()
.request_saved_device_pairing(bob.core().status().endpoint_id.clone())
.unwrap()
|| bob
.core()
.list_device_relationships()
.unwrap()
.iter()
.all(|entry| entry.state != DeviceRelationshipState::PendingIncoming)
);
// Stronger: after block, bob must not surface an incoming pairing prompt.
std::thread::sleep(std::time::Duration::from_millis(200));
assert!(bob
.core()
.list_device_relationships()
.unwrap()
.into_iter()
.filter(|entry| entry.remote_endpoint_id == alice_id)
.all(|entry| entry.state != DeviceRelationshipState::PendingIncoming));
}
#[test]
fn ticket_errors_redact_raw_ticket_blobs() {
let err = VnidropError::ticket(anyhow::anyhow!(
"bad ticket vnd1:abcDEF1234567890 and endpoint 0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
));
let rendered = err.to_string();
assert!(!rendered.contains("vnd1:abcDEF"));
assert!(!rendered.contains("0123456789abcdef0123456789abcdef"));
assert!(rendered.contains("[redacted]"));
}

View File

@@ -11,6 +11,8 @@ fn default_limits_bound_ticket_and_approval_pressure() {
assert!(limits.max_ticket_bytes <= 256 * 1024);
assert!(limits.max_pending_approvals <= 64);
assert!(limits.max_total_bytes <= 256 * 1024 * 1024 * 1024);
assert_eq!(limits.max_saved_devices, 256);
assert!(limits.max_pending_offers <= 16);
}
#[test]
@@ -21,3 +23,12 @@ fn zero_limit_is_rejected() {
};
assert!(limits.validate().is_err());
}
#[test]
fn zero_saved_device_limit_is_rejected() {
let limits = CoreLimits {
max_saved_devices: 0,
..CoreLimits::default()
};
assert!(limits.validate().is_err());
}

View File

@@ -98,9 +98,11 @@ fn transfers_file_between_two_cores() {
let completed = wait_for_sender_transfer_event(&sender, share.transfer_id, "completed");
assert!(completed.data_json.contains("\"connection_id\":"));
assert!(completed.data_json.contains("\"request_id\":"));
assert!(completed
// Production events redact endpoint ids; typed APIs remain the source of truth.
assert!(!completed
.data_json
.contains(receiver.core.status().endpoint_id.as_str()));
assert!(completed.data_json.contains("redacted"));
receiver.core.delete_receive_history().unwrap();
assert_eq!(receiver.core.list_received_artifacts().unwrap(), artifacts);