Create and use API to report guard/fallback skew. (c3c43b08) · Commits · The Tor Project / Core / Arti

crates/tor-circmgr/src/build.rs

+39 −12

Original line number	Original line	Diff line number	Diff line
	@@ -13,7 +13,7 @@ use std::sync::{
	Arc,		Arc,
	};		};
	use std::time::{Duration, Instant};		use std::time::{Duration, Instant};
	use tor_chanmgr::ChanMgr;		use tor_chanmgr::{ChanMgr, ChanProvenance};
	use tor_guardmgr::GuardStatus;		use tor_guardmgr::GuardStatus;
	use tor_linkspec::{ChanTarget, OwnedChanTarget, OwnedCircTarget};		use tor_linkspec::{ChanTarget, OwnedChanTarget, OwnedCircTarget};
	use tor_proto::circuit::{CircParameters, ClientCirc, PendingClientCirc};		use tor_proto::circuit::{CircParameters, ClientCirc, PendingClientCirc};
	@@ -40,6 +40,7 @@ pub(crate) trait Buildable: Sized {
	async fn create_chantarget<RT: Runtime>(		async fn create_chantarget<RT: Runtime>(
	chanmgr: &ChanMgr<RT>,		chanmgr: &ChanMgr<RT>,
	rt: &RT,		rt: &RT,
			guard_status: &GuardStatusHandle,
	ct: &OwnedChanTarget,		ct: &OwnedChanTarget,
	params: &CircParameters,		params: &CircParameters,
	) -> Result<Self>;		) -> Result<Self>;
	@@ -49,6 +50,7 @@ pub(crate) trait Buildable: Sized {
	async fn create<RT: Runtime>(		async fn create<RT: Runtime>(
	chanmgr: &ChanMgr<RT>,		chanmgr: &ChanMgr<RT>,
	rt: &RT,		rt: &RT,
			guard_status: &GuardStatusHandle,
	ct: &OwnedCircTarget,		ct: &OwnedCircTarget,
	params: &CircParameters,		params: &CircParameters,
	) -> Result<Self>;		) -> Result<Self>;
	@@ -72,15 +74,29 @@ async fn create_common<RT: Runtime, CT: ChanTarget>(
	chanmgr: &ChanMgr<RT>,		chanmgr: &ChanMgr<RT>,
	rt: &RT,		rt: &RT,
	target: &CT,		target: &CT,
			guard_status: &GuardStatusHandle,
	) -> Result<PendingClientCirc> {		) -> Result<PendingClientCirc> {
	let (chan, _provenance) =		// Get or construct the channel.
	chanmgr		let result = chanmgr.get_or_launch(target).await;
	.get_or_launch(target)
	.await		// Report the clock skew if appropriate, and exit if there has been an error.
	.map_err(\|cause\| Error::Channel {		let chan = match result {
			Ok((chan, ChanProvenance::NewlyCreated)) => {
			guard_status.skew(chan.clock_skew());
			chan
			}
			Ok((chan, _)) => chan,
			Err(cause) => {
			if let Some(skew) = cause.clock_skew() {
			guard_status.skew(skew);
			}
			return Err(Error::Channel {
	peer: OwnedChanTarget::from_chan_target(target),		peer: OwnedChanTarget::from_chan_target(target),
	cause,		cause,
	})?;		});
			}
			};
			// Construct the (zero-hop) circuit.
	let (pending_circ, reactor) = chan.new_circ().await.map_err(\|error\| Error::Protocol {		let (pending_circ, reactor) = chan.new_circ().await.map_err(\|error\| Error::Protocol {
	error,		error,
	peer: None, // we don't blame the peer, because new_circ() does no networking.		peer: None, // we don't blame the peer, because new_circ() does no networking.
	@@ -99,10 +115,11 @@ impl Buildable for ClientCirc {
	async fn create_chantarget<RT: Runtime>(		async fn create_chantarget<RT: Runtime>(
	chanmgr: &ChanMgr<RT>,		chanmgr: &ChanMgr<RT>,
	rt: &RT,		rt: &RT,
			guard_status: &GuardStatusHandle,
	ct: &OwnedChanTarget,		ct: &OwnedChanTarget,
	params: &CircParameters,		params: &CircParameters,
	) -> Result<Self> {		) -> Result<Self> {
	let circ = create_common(chanmgr, rt, ct).await?;		let circ = create_common(chanmgr, rt, ct, guard_status).await?;
	circ.create_firsthop_fast(params)		circ.create_firsthop_fast(params)
	.await		.await
	.map_err(\|error\| Error::Protocol {		.map_err(\|error\| Error::Protocol {
	@@ -113,10 +130,11 @@ impl Buildable for ClientCirc {
	async fn create<RT: Runtime>(		async fn create<RT: Runtime>(
	chanmgr: &ChanMgr<RT>,		chanmgr: &ChanMgr<RT>,
	rt: &RT,		rt: &RT,
			guard_status: &GuardStatusHandle,
	ct: &OwnedCircTarget,		ct: &OwnedCircTarget,
	params: &CircParameters,		params: &CircParameters,
	) -> Result<Self> {		) -> Result<Self> {
	let circ = create_common(chanmgr, rt, ct).await?;		let circ = create_common(chanmgr, rt, ct, guard_status).await?;
	circ.create_firsthop_ntor(ct, params.clone())		circ.create_firsthop_ntor(ct, params.clone())
	.await		.await
	.map_err(\|error\| Error::Protocol {		.map_err(\|error\| Error::Protocol {
	@@ -192,8 +210,14 @@ impl<R: Runtime, C: Buildable + Sync + Send + 'static> Builder<R, C> {
	OwnedPath::ChannelOnly(target) => {		OwnedPath::ChannelOnly(target) => {
	// If we fail now, it's the guard's fault.		// If we fail now, it's the guard's fault.
	guard_status.pending(GuardStatus::Failure);		guard_status.pending(GuardStatus::Failure);
	let circ =		let circ = C::create_chantarget(
	C::create_chantarget(&self.chanmgr, &self.runtime, &target, &params).await?;		&self.chanmgr,
			&self.runtime,
			&guard_status,
			&target,
			&params,
			)
			.await?;
	self.timeouts		self.timeouts
	.note_hop_completed(0, self.runtime.now() - start_time, true);		.note_hop_completed(0, self.runtime.now() - start_time, true);
	n_hops_built.fetch_add(1, Ordering::SeqCst);		n_hops_built.fetch_add(1, Ordering::SeqCst);
	@@ -204,7 +228,8 @@ impl<R: Runtime, C: Buildable + Sync + Send + 'static> Builder<R, C> {
	let n_hops = p.len() as u8;		let n_hops = p.len() as u8;
	// If we fail now, it's the guard's fault.		// If we fail now, it's the guard's fault.
	guard_status.pending(GuardStatus::Failure);		guard_status.pending(GuardStatus::Failure);
	let circ = C::create(&self.chanmgr, &self.runtime, &p[0], &params).await?;		let circ =
			C::create(&self.chanmgr, &self.runtime, &guard_status, &p[0], &params).await?;
	self.timeouts		self.timeouts
	.note_hop_completed(0, self.runtime.now() - start_time, n_hops == 0);		.note_hop_completed(0, self.runtime.now() - start_time, n_hops == 0);
	// If we fail after this point, we can't tell whether it's		// If we fail after this point, we can't tell whether it's
	@@ -616,6 +641,7 @@ mod test {
	async fn create_chantarget<RT: Runtime>(		async fn create_chantarget<RT: Runtime>(
	_: &ChanMgr<RT>,		_: &ChanMgr<RT>,
	rt: &RT,		rt: &RT,
			_guard_status: &GuardStatusHandle,
	ct: &OwnedChanTarget,		ct: &OwnedChanTarget,
	_: &CircParameters,		_: &CircParameters,
	) -> Result<Self> {		) -> Result<Self> {
	@@ -635,6 +661,7 @@ mod test {
	async fn create<RT: Runtime>(		async fn create<RT: Runtime>(
	_: &ChanMgr<RT>,		_: &ChanMgr<RT>,
	rt: &RT,		rt: &RT,
			_guard_status: &GuardStatusHandle,
	ct: &OwnedCircTarget,		ct: &OwnedCircTarget,
	_: &CircParameters,		_: &CircParameters,
	) -> Result<Self> {		) -> Result<Self> {

crates/tor-circmgr/src/build/guardstatus.rs

+12 −0

Original line number	Original line	Diff line number	Diff line
	@@ -2,6 +2,7 @@

	use std::sync::Mutex;		use std::sync::Mutex;
	use tor_guardmgr::{GuardMonitor, GuardStatus};		use tor_guardmgr::{GuardMonitor, GuardStatus};
			use tor_proto::ClockSkew;

	/// A shareable object that we can use to report guard status to the guard		/// A shareable object that we can use to report guard status to the guard
	/// manager.		/// manager.
	@@ -45,6 +46,17 @@ impl GuardStatusHandle {
	}		}
	}		}

			/// Change the pending clock skew for this guard.
			///
			/// As with pending status, this value won't be sent to the guard manager
			/// until this `GuardStatusHandle` is dropped or committed.
			pub(crate) fn skew(&self, skew: ClockSkew) {
			let mut mon = self.mon.lock().expect("Poisoned lock");
			if let Some(mon) = mon.as_mut() {
			mon.skew(skew);
			}
			}

	/// Report the provided status to the guard manager.		/// Report the provided status to the guard manager.
	///		///
	/// Future calls to methods on this object will do nothing.		/// Future calls to methods on this object will do nothing.

crates/tor-guardmgr/src/daemon.rs

+3 −2

Original line number	Original line	Diff line number	Diff line
	@@ -9,6 +9,7 @@ use crate::GuardMgrInner;
	#[cfg(test)]		#[cfg(test)]
	use futures::channel::oneshot;		use futures::channel::oneshot;
	use futures::{channel::mpsc, stream::StreamExt};		use futures::{channel::mpsc, stream::StreamExt};
			use tor_proto::ClockSkew;

	use std::sync::{Mutex, Weak};		use std::sync::{Mutex, Weak};

	@@ -17,7 +18,7 @@ use std::sync::{Mutex, Weak};
	pub(crate) enum Msg {		pub(crate) enum Msg {
	/// A message sent by a [`GuardMonitor`](crate::GuardMonitor) to		/// A message sent by a [`GuardMonitor`](crate::GuardMonitor) to
	/// report the status of an attempt to use a guard.		/// report the status of an attempt to use a guard.
	Status(RequestId, GuardStatus),		Status(RequestId, GuardStatus, Option<ClockSkew>),
	/// Tells the task to reply on the provided oneshot::Sender once		/// Tells the task to reply on the provided oneshot::Sender once
	/// it has seen this message. Used to indicate that the message		/// it has seen this message. Used to indicate that the message
	/// queue is flushed.		/// queue is flushed.
	@@ -40,7 +41,7 @@ pub(crate) async fn report_status_events(
	) {		) {
	loop {		loop {
	match events.next().await {		match events.next().await {
	Some(Msg::Status(id, status)) => {		Some(Msg::Status(id, status, _skew)) => {
	// We've got a report about a guard status.		// We've got a report about a guard status.
	if let Some(inner) = inner.upgrade() {		if let Some(inner) = inner.upgrade() {
	let mut inner = inner.lock().expect("Poisoned lock");		let mut inner = inner.lock().expect("Poisoned lock");

crates/tor-guardmgr/src/pending.rs

+14 −1

Original line number	Original line	Diff line number	Diff line
	@@ -20,6 +20,7 @@ use std::pin::Pin;
	use std::sync::atomic::{AtomicU64, Ordering};		use std::sync::atomic::{AtomicU64, Ordering};
	use std::task::{Context, Poll};		use std::task::{Context, Poll};
	use std::time::Instant;		use std::time::Instant;
			use tor_proto::ClockSkew;

	use tor_basic_utils::skip_fmt;		use tor_basic_utils::skip_fmt;

	@@ -126,6 +127,9 @@ pub struct GuardMonitor {
	/// badly against the guard at all: typically, because the circuit's		/// badly against the guard at all: typically, because the circuit's
	/// path is not random.		/// path is not random.
	ignore_indeterminate: bool,		ignore_indeterminate: bool,
			/// If set, we will report the given clock skew as having been observed and
			/// authenticated from this guard or fallback.
			pending_skew: Option<ClockSkew>,
	/// A sender that needs to get told when the attempt to use the guard is		/// A sender that needs to get told when the attempt to use the guard is
	/// finished or abandoned.		/// finished or abandoned.
	///		///
	@@ -143,6 +147,7 @@ impl GuardMonitor {
	id,		id,
	pending_status: GuardStatus::AttemptAbandoned,		pending_status: GuardStatus::AttemptAbandoned,
	ignore_indeterminate: false,		ignore_indeterminate: false,
			pending_skew: None,
	snd: Some(snd),		snd: Some(snd),
	}		}
	}		}
	@@ -183,6 +188,14 @@ impl GuardMonitor {
	self.pending_status = status;		self.pending_status = status;
	}		}

			/// Set the given clock skew value to be reported to the guard manager.
			///
			/// Clock skew can be reported on success or failure, but it should only be
			/// reported if the first hop is actually authenticated.
			pub fn skew(&mut self, skew: ClockSkew) {
			self.pending_skew = Some(skew);
			}

	/// Return the current pending status and "ignore indeterminate"		/// Return the current pending status and "ignore indeterminate"
	/// status for this guard monitor.		/// status for this guard monitor.
	#[cfg(feature = "testing")]		#[cfg(feature = "testing")]
	@@ -214,7 +227,7 @@ impl GuardMonitor {
	.snd		.snd
	.take()		.take()
	.expect("GuardMonitor initialized with no sender")		.expect("GuardMonitor initialized with no sender")
	.unbounded_send(daemon::Msg::Status(self.id, msg));		.unbounded_send(daemon::Msg::Status(self.id, msg, self.pending_skew));
	}		}

	/// Report the pending message for his guard, whatever it is.		/// Report the pending message for his guard, whatever it is.