Lines
73.55 %
Functions
87.21 %
Branches
100 %
//! Directory Mirror Operation.
//!
//! # Specifications
//! * [Directory cache operation](https://spec.torproject.org/dir-spec/directory-cache-operation.html).
//! # Rationale
//! This module implements the "core operation" of a directory mirror.
//! "Core operation" primarily refers to the logic involved in downloading
//! network documents from an upstream authority and inserting them into the
//! database. This module notably **DOES NOT** provide any public (in the HTTP
//! sense) endpoints for querying documents. This is purposely behind a different
//! module, so that the directory authority implementation can also make use of it.
//! You can think of this module as the one implementing the things unique
//! to directory mirrors.
use std::{
collections::{HashMap, HashSet, VecDeque},
fmt::Debug,
hash::Hash,
marker::PhantomData,
mem,
net::SocketAddr,
time::Duration,
};
use educe::Educe;
use r2d2::Pool;
use r2d2_sqlite::SqliteConnectionManager;
use rand::Rng;
use rusqlite::Transaction;
use tokio::net::TcpStream;
use tokio_util::compat::TokioAsyncReadCompatExt;
use tor_checkable::TimeBound;
use tor_dirclient::{
Error as DirClientError, RequestError, RequestFailedError,
request::{AuthCertRequest, ConsensusRequest, Requestable, RouterDescRequest},
use tor_dircommon::{authority::AuthorityContacts, config::DirTolerance};
use tor_error::internal;
use tor_netdoc::{
doc::{
authcert::{AuthCert, AuthCertKeyIds, AuthCertUnverified},
netstatus::{ConsensusFlavor, ConsensusVerifiabilityError},
routerdesc::RouterDescUnverified,
},
parse2::{
self, NetdocParseable, NetdocParseableUnverified, ParseError, ParseInput, VerifyFailed,
use tor_rtcompat::PreferredRuntime;
use tracing::{debug, warn};
use crate::{
database::{
self as db, AuthCertMeta, ConsensusMeta, ContentEncoding, DescriptorMeta, Sha1, Sha256,
Timestamp,
err::{AuthorityRequestError, DatabaseError, OperationError},
types::{FlavoredConsensusUnverified, FlavoredDescriptor, FlavoredDescriptorPlain},
mod poc;
/// The duration to try verifying an unverified consensus.
///
/// This exists primarily to properly handle the
/// [`ConsensusVerifiabilityError::InsufficientTrustedSigners`] error. Besides,
/// it also serves as an upper-limit for the time we can spend in the phase for
/// fetching authority certificates, which could otherwise run for a very long
/// time, as long as every request makes progress and finishes before a
/// connection timeout is reached.
/// Obviously, we cannot use the values inside the consensus, as they are
/// untrusted.
const UNVERIFIED_TTL: Duration = Duration::from_mins(10);
/// Query a maximum.
const DESCRIPTOR_LIMIT: Option<u64> = Some(92);
/// The endpoint(s) of a download authority identifying it.
type DownloadAuthority = [SocketAddr];
/// The various states for the [`StaticEngine`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, strum::Display)]
enum State {
/// Loads the most recent valid (and verified) consensus from the database
/// into memory.
/// Transitions from:
/// * Start, if a recent valid consensus exists in the database.
/// * [`State::StoreConsensus`], if successfully finished.
/// Transitions into:
/// * [`State::Descriptors`]
LoadConsensus,
/// Downloads the most recent consensus from a directory authority.
/// * Start, if no recent valid consensus exists in the database.
/// * [`State::Descriptors`], if ttl is over.
/// * [`State::Hibernate`], if ttl is over.
/// * [`State::AuthCerts`], if we miss authority certificates.
/// * [`State::StoreConsensus`], if all authority certificates exist in the
/// database.
/// * [`State::Hibernate`], if we have insufficient trusted authorities.
// TODO DIRMIRROR: What to do in the case of getting an invalid consensus
// such as junk data? The normal retry logic sounds reasonable here.
FetchConsensus,
/// Downloads, validates, and stores the missing authority certificates from
/// the downloaded unvalidated consensus into the database.
/// * [`State::FetchConsensus`], if we miss authority certificates.
/// * [`State::AuthCerts`], if we still miss authority certificates.
/// * [`State::StoreConsensus`], if we got all authority certificates.
AuthCerts,
/// Validates and stores the downloaded unvalidated consensus into the
/// * [`State::FetchConsensus`], if we have all authority certificates.
/// * [`State::AuthCerts`], if we have all authority certificates.
/// * [`State::LoadConsensus`]
StoreConsensus,
/// Downloads missing network documents (descriptors) from a directory
/// authority.
/// * [`State::LoadConsensus`], if we initialize.
/// * [`State::Descriptors`], if we still have missing descriptors left.
/// * [`State::FetchConsensus`], if ttl is over.
/// * [`State::Hibernate`], if nothing is left.
Descriptors,
/// Hibernate because nothing is left.
/// * [`State::FetchConsensus`], if we have insufficient trusted
/// authorities.
/// * [`State::Descriptors`], if we have downloaded all descriptors.
/// * [`State::FetchConsensus`], if the ttl is over.
Hibernate,
}
/// The execution engine for the finite state machine.
/// The states themselves are explained in [`State`].
/// This data structure itself is static and contains no state, but merely
/// configuration primitives that stay constant throughout the runtime of the
/// program, such as the [`AuthorityContacts`], and the
/// [`DirTolerance`]. It can be kept throughout the entire runtime and only
/// consists for convenience in order to not give each state machine related
/// (then static) method a super long signature containing these fields.
/// Besides these dynamic fields, there is also a generic parameter specifying
/// the consensus flavor that is being used.
/// The state itself is computed fully deterministically from the data found
/// within the database and [`ConsensusBoundData`].
/// This is the reason on why this structure is not called `StateMachine`,
/// because this implies that the type in itself carries state, which is not
/// true, because the state is stored entirely external, with this engine
/// only processing and modifying it.
/// See [`StaticEngine::determine_state()`] for more details.
#[derive(Debug)]
struct StaticEngine<T> {
/// The authorities we are acknowledging.
authorities: AuthorityContacts,
/// The document tolerance we are accepting.
tolerance: DirTolerance,
/// The preferred runtime for compatibility with other arti crates.
/// Generally obtained through [`PreferredRuntime::current()`].
rt: PreferredRuntime,
/// The content encodings we apply to documents.
encodings: HashSet<ContentEncoding>,
/// Utilizes the generic type parameter.
_phantom: PhantomData<T>,
/// Additional state machine data concerning a single consensus.
/// This data type exists to store data that cannot be stored in the database,
/// including the consensus that is currently verified, the consensus that is
/// currently served, the timestamp until we will serve it, etc.
/// # Field Requirements
/// In order for a field to be contained here, at least one of the following
/// criteria should be met:
/// * The item needs to persist between multiple invocation and cannot be
/// obtained from the database.
/// * The item itself serves as an identifier to the database and querying it
/// it again can not be done in a pure fashion, e.g. a [`ConsensusMeta`]
/// identifier.
/// * The item exists purely for optimization purposes and can be thrown away
/// at any time without data loss, such as content-addressable caches.
/// These caches must be pure in the sense that a cache miss must always
/// return the same result as a cache hit and vice versa.
//
// TODO DIRMIRROR: This is probably better of as a struct instead of an enum.
#[derive(Debug, Clone)]
enum ConsensusBoundData<T: FlavoredConsensusUnverified> {
/// No state is loaded in memory at the moment.
None,
/// We have downloaded a consensus but it is not yet verified.
Unverified {
/// The unverified parsed consensus we have.
consensus: T,
/// The unparsed raw consensus we have.
raw: String,
/// When to stop dealing with this consensus and download again.
/// Useful to have in order to not stuck forever in a verification
/// loop, i.e. due to unavailable authentication certificates, etc.
ttl: Timestamp,
/// We have downloaded and verified a consensus.
Verified {
/// The verified consensus we have.
/// Contained in this structure because it is the only way we can
/// uniquely identify a consensus, which is crucial for operation, as
/// things like the `ttl` are derived from it.
/// While methods like [`ConsensusMeta::query`] would probably also
/// work correctly, relying on them feels rather wrong, as they were
/// not made to be pure.
consensus: ConsensusMeta<T>,
/// When to stop dealing with this consensus and fetching a new one.
/// Descriptors we refuse to further download.
/// Right now, this state gets cleaned, everytime we restart or obtain
/// a new consensus. This is probably okay, as we most likely do not
/// want to store it on disk.
futile: Box<FutileDescriptors>,
impl<T: FlavoredConsensusUnverified> StaticEngine<T> {
/// Determines the [`State`] only from the database and [`ConsensusBoundData`].
/// This method is fully idempotent, meaning it only depends upon the data
/// found within the database and the [`ConsensusBoundData`]; there is no
/// internal `state` variable or something contained within [`StaticEngine`].
fn determine_state(
&self,
tx: &Transaction<'_>,
data: &ConsensusBoundData<T>,
now: Timestamp,
) -> Result<State, DatabaseError> {
// Determine the state primarily upon ConsensusBoundData combined with
// a few database queries, as well as the current time of course.
let state = match data {
// ConsensusBoundData::None means that we currently have no
// consensus in memory. This may be the case because we just
// started up or because we just downloaded, validated, and inserted
// a consensus into the database and reset ConsensusBoundData to
// None afterwards.
ConsensusBoundData::None => {
// Check whether there is a valid consensus in the database at all.
match self.recent_consensus(tx, now)? {
// Some consensus means we can load it.
Some(_) => State::LoadConsensus,
// None means we must download it.
None => State::FetchConsensus,
// ConsensusBoundData::Unverified means that we recently downloaded
// a consensus through State::FetchConsensus. It is not fully
// validated yet and we may not even be able due to missing
// authority certificates.
ConsensusBoundData::Unverified { consensus, ttl, .. } if now < *ttl => {
let certs_already = self.certs_already(tx, now)?;
match consensus.can_verify(self.authorities.v3idents(), &certs_already) {
// We can verify and insert the consensus.
Ok(()) => State::StoreConsensus,
// We will never be able to verify the consensus, better
// times may come ...
Err(ConsensusVerifiabilityError::InsufficientTrustedSigners) => {
State::Hibernate
// We need to fetch more authority certificates from the
// network.
Err(ConsensusVerifiabilityError::MissingAuthCerts { .. }) => State::AuthCerts,
ConsensusBoundData::Unverified { .. } => State::FetchConsensus,
// ConsensusBoundData::Verified means that we have successfully
// loaded a recent valid consensus from the database using
// State::LoadConsensus. Depending on this, we download the missing
// network documents (descriptors) from a directory authority, if
// any.
ConsensusBoundData::Verified {
consensus,
ttl,
futile,
} => {
// See if there are any missing descriptors that are not Refused
// yet. It is important to not exclude the NotFound descriptors
// here because they may be available at a different authority.
// TODO DIRMIRROR: What to do in the case there is no authority
// left in our list of endpoints? Right now, this module of the
// code is not aware of the endpoints to try and it is the
// responsibility of the caller to check this. Perhaps we
// should provide a possibility to force hibernate, as in passing
// an Option<DownloadAuthority> or an empty list. An alternative
// worth exploring might also be to perform the RetryDelay
// wait logic, as in to not let the caller perform any timeouts
// themselves. Either way, we want to wait to hibernate and
// wait for the next consensus in the case that no authority
// is left.
// TODO DIRMIRROR: This appears to be prone to copy and paste
// errors, we should do something about it, as in macros and/or
// traits.
let (missing_routers, missing_micros, missing_extras) = (
consensus.missing_routers(tx, futile.routers.refused(), Some(1))?,
consensus.missing_micros(tx, futile.micros.refused(), Some(1))?,
consensus.missing_extras(tx, futile.extras.refused(), Some(1))?,
);
if *ttl <= now {
// The ttl has been surpassed, download a new
// consensus. It is very important TO NOT transition to
// State::LoadConsensus here, because the current consensus
// may still be valid but not fresh anymore, in which case
// State::LoadConsensus will continue to obtain it from the
// database until valid-after has been surpassed, which is
// most definitely not what we want.
State::FetchConsensus
} else if missing_routers.is_empty()
&& missing_micros.is_empty()
&& missing_extras.is_empty()
{
// All queues are empty, meaning we are done, until ttl
// ends.
} else {
// The ttl has not been surpassed and we have stuff
// to download, so we need to obtain the descriptors.
State::Descriptors
Ok(state)
/// Executes a single state iteration in the finite state machine.
/// The return value is of type [`Result<(), OperationError>`].
/// The success type is not of much interest for calling applications.
/// However, the error case itself should be passed towards
/// [`crate::err::IsFatal::is_fatal()`] in order to either abort the
/// application or retry with an appropriate timeout.
// TODO: Use tracing instrumentation here.
// TODO DIRMIRROR: Document the state transition check which we have to do
// because of database invariances no longer holding true.
async fn execute<R: Rng>(
pool: &Pool<SqliteConnectionManager>,
data: &mut ConsensusBoundData<T>,
endpoint: &DownloadAuthority,
rng: &mut R,
) -> Result<(), OperationError> {
// TODO DIRMIRROR: Right now, we pass around the pool object which is
// not very nice. Instead, we should match based on the following
// criteria: Needs write transaction, needs read transaction, needs
// pool, needs no database access.
// TODO: Should we return DatabaseError or something like
// StateDeterminationError? Either way, both cases should be seriously
// fatal.
let state = db::read_tx(pool, |tx| self.determine_state(tx, data, now))??;
debug!("state is {state}");
match state {
State::LoadConsensus => self.load_consensus(pool, data, now, rng),
State::FetchConsensus => Ok(self.fetch_consensus(data, endpoint, now).await?),
State::AuthCerts => self.auth_certs(pool, data, endpoint, now).await,
State::StoreConsensus => self.store_consensus(pool, data, now),
State::Descriptors => self.descriptors(pool, data, endpoint, now).await,
State::Hibernate => self.hibernate(data, now).await,
/// Executes [`State::LoadConsensus`].
/// This method does the following:
/// * Load the most recent valid consensus from the database.
/// * Compute the ttl for it.
/// * Compute the missing descriptors for it.
fn load_consensus<R: Rng>(
// Load the most recent valid consensus from the database.
// If there is no consensus, we should have not entered the state, which
// means that the database must have been externally verified.
// In this case, it is probably better to return a bug, as external
// applications arbitrarily modifying the database while we are running
// leaves too much room for wrong/weird behavior.
let consensus = db::read_tx(pool, |tx| self.recent_consensus(tx, now))??
.ok_or_else(|| internal!("database externally modified?"))?;
let ttl = consensus.ttl(rng);
let futile = Box::new(FutileDescriptors::default());
*data = ConsensusBoundData::Verified {
Ok(())
/// Fetches a consensus from an upstream authority.
// TODO DIRMIRROR: Add logging.
#[allow(clippy::string_slice)] // TODO
async fn fetch_consensus(
) -> Result<(), AuthorityRequestError> {
// Obtain the consensus.
let (raw, consensus) = self
.send_request(endpoint, ConsensusRequest::new(T::flavor()))
.await?;
let mut consensus = consensus.into_iter().collect::<VecDeque<_>>();
// Check for the correct number of results.
if consensus.len() != 1 {
return Err(AuthorityRequestError::Response(
"invalid number of consensus?",
));
let (consensus, start, end) = match consensus.pop_front().expect("pop failed?") {
(Ok(consensus), start, end) => (consensus, start, end),
// We do not mark unparseable consensuses as futile because they
// cannot be referenced in a semi-unique fashion but rather depend
// on the time only.
// TODO DIRMIRROR: Return a more descriptive error; ideally in
// all places where only one document was returned and that document
// failed to parse.
(Err(_), _, _) => {
"invalid consensus received",
let raw = raw[start..end].to_string();
let ttl = now + UNVERIFIED_TTL;
// And store it.
*data = ConsensusBoundData::Unverified {
raw,
/// Fetches, validates, and stores authority certificates.
async fn auth_certs(
let certs_already = db::read_tx(pool, |tx| self.certs_already(tx, now))??;
// Obtain the missing authority certificate identifiers.
let missing = match data {
ConsensusBoundData::Unverified { consensus, .. } => {
Err(ConsensusVerifiabilityError::MissingAuthCerts { missing, .. }) => missing,
_ => return Err(OperationError::Bug(internal!("we have all auth certs"))),
_ => return Err(OperationError::Bug(internal!("data is not unverified"))),
// Compose the request.
let mut requ = AuthCertRequest::new();
for kp in missing.iter().copied() {
requ.push(kp);
// Fire it off.
let (resp, certs) = self
.send_request::<_, AuthCertUnverified>(endpoint, requ)
// Verify each certificate. Invalid certificates and other problems get
// logged and filtered out, with the result being then inserted into
// the database.
let certs = certs
.into_iter()
.filter_map(|(unverified, start, end)| {
let unverified = match unverified {
Ok(v) => v,
Err(e) => {
// We do not mark unparseable authority certificates as
// futile because they do not have an authoritative
// digest we could ad-hoc compare them against.
debug!("ignoring invalid authority certificate: {e}");
return None;
let kp = unverified.inspect_unverified().0.key_ids();
// Skip certificates we did not asked for.
// Not much of an issue because certificate verification will
// usually fail anyways, except for this weird edge-case where we
// actually have that id fingerprint in the v3idents.
if !missing.contains(&kp) {
debug!("authority returned certificate we did not asked for: {kp:?}");
let verified = match unverified.verify(self.authorities.v3idents()) {
Ok(c) => c,
debug!("received invalid auth cert ({kp:?}): {e}");
let timely = match self
.tolerance
.extend_tolerance(verified)
.if_valid_at(&now.into())
debug!("received non-timely auth cert ({kp:?}): {e}");
Some((timely, &resp[start..end]))
})
.collect::<Vec<_>>();
// When we have reached this, it means that this call made no progress,
// i.e. the authority only returned certificates we were not interested
// in.
if certs.is_empty() {
Err(AuthorityRequestError::NoRequestedDocumentsReturned)?;
// Finally, insert them all into the database.
db::rw_tx(pool, |tx| {
for (cert, data) in certs {
AuthCertMeta::insert(tx, self.encodings.iter().copied(), &cert, data)?;
Ok::<_, DatabaseError>(())
})??;
/// Verifies a consensus and inserts it into the database.
fn store_consensus(
// It is fine to replace data with ConsensusBoundData::None because in
// both cases, the following state of this will always be LoadConsensus
// or FetchConsensus, both of them no longer requiring the previous data.
// Yes, we want to explicitly discard this in the case of a recoverable
// error in order to retry again by fetching a new consensus from a
// different authority.
let (unverified, raw) = match mem::replace(data, ConsensusBoundData::None) {
ConsensusBoundData::Unverified { consensus, raw, .. } => (consensus, raw),
// TODO DIRMIRROR: Somewhere in here, we should generate the consensus
// diffs. However, this is a CPU intensive task and needs further
// coordination with the orport developers. See arti#2706
let (verified, sigs) = self.verify_consensus(unverified, &certs_already, now)?;
ConsensusMeta::<T>::insert(
tx,
self.encodings.iter().copied(),
(&verified, &sigs),
&raw,
)?;
Ok::<_, OperationError>(())
})?
/// Downloads, verifies, and inserts the descriptors associated with the
/// currently served consensus.
async fn descriptors(
match T::flavor() {
ConsensusFlavor::Plain => {
// TODO DIRMIRROR: Add extras.
self.descriptors_rd(pool, data, endpoint, now).await
ConsensusFlavor::Microdesc => {
// TODO DIRMIRROR: Add micros.
todo!()
/// Router descriptor part of [`StaticEngine::descriptors()`].
async fn descriptors_rd(
debug_assert_eq!(T::flavor(), ConsensusFlavor::Plain);
let (meta, futile) = match data {
consensus, futile, ..
} => (consensus, futile),
_ => return Err(OperationError::Bug(internal!("data not verified"))),
// Obtain the missing descriptors.
let routers = db::read_tx(pool, |tx| {
meta.missing_routers(tx, futile.routers.excluded(endpoint), DESCRIPTOR_LIMIT)
if routers.is_empty() {
// This is equivalent to sending off an empty request and failing
// afterwards because no progress was made, but it saves us
// bandwidth by not doing a redundant HTTP request. We may enter
// this branch if there are no descriptors left that are not marked
// as not found or refused, in which case we must try the next
// authority as it may have the not found descriptors.
debug!("not downloading any router descriptors");
Err(OperationError::AuthorityRequest(
AuthorityRequestError::NoRequestedDocumentsReturned,
))?;
debug!("downloading the following router descriptors: {routers:?}");
let requ = RouterDescRequest::from_iter(routers.iter().map(|sha1| sha1.0));
let (raw, docs) = self
.send_request::<_, RouterDescUnverified>(endpoint, requ)
// Verify each router descriptor.
// We do this in a closure so we do not accidentially return early,
// because we must mark the not found and refused descriptors as such
// prior to a return, even in an error case.
let process_downloads = || -> DescriptorDownloads<Sha1, FlavoredDescriptorPlain> {
let mut downloads = DescriptorDownloads::from_missing(routers.into_iter());
for (unverified, start, end) in docs {
let Ok(unverified) = unverified else {
// TODO DIRMIRROR: Handle parser failures as potentially futile.
// This is going to be a big can of worms, because in the case
// of a parser failure, we must check whether the part between
// "router" and "router-signature" matches the digest of a
// requested descriptor, while checking that the body and
// signature parts of the document alone parse just fine.
// body_ok && signature_ok => the parser has not failed, unreachable
// body_ok && !signature_ok => not futile
// !body_ok && signature_ok => futile
// !body_ok && !signature_ok => futile
// (with ok meaning that the parser suceeded, no statement
// of cryptographic validity is being made here)
// The `RoutereDescUnverifiedParsedBody as NetdocParseable`
// pattern may be useful for this check.
continue;
// Defensive internal checks.
let Some(raw) = raw.get(start..end) else {
warn!("parse2 returned invalid string slice: {start}..{end}");
let raw = raw.to_string();
let Some(sha1) = unverified.sigs.hashes.sha1 else {
warn!("accumulator contains no SHA-1?");
let sha1 = Sha1(sha1);
// Checks whether the response is weird, as in containing
// unrequested or duplicate descriptors.
match downloads.get(&sha1) {
Some(DescriptorDownloadResult::NotFound) => {}
Some(
DescriptorDownloadResult::Found(_, _) | DescriptorDownloadResult::Refused,
) => {
debug!("{sha1:?} contained multiple times in response");
None => {
debug!("ignoring unrequested router descriptor: {sha1:?}");
// Perform the relevant cryptographic checks.
let sigs = unverified.sigs.clone();
let verified = match unverified.verify() {
Ok(verified) => verified,
// Continuing here implicitly marks the descriptor as not
// found after the loop ends.
debug!("ignoring invalid router descriptor: {sha1:?}: {e}");
Ok(timely) => timely,
// Untimely is considered as refused for the lifespan of the
// current consensus.
debug!("ignoring untimely router descriptor: {sha1:?}: {e}");
let prev = downloads.insert(sha1, DescriptorDownloadResult::Refused);
debug_assert!(matches!(prev, Some(DescriptorDownloadResult::NotFound)));
let prev =
downloads.insert(sha1, DescriptorDownloadResult::Found((timely, sigs), raw));
downloads
// Process all downloads isolated, so that we do not return early
// on accident. While we still have one minor exception below with
// regard to DatabaseError, that one is well documented.
let downloads = process_downloads();
// Determine if we made progress by checking whether we have any
// descriptors that are either found or refused. If this is the case,
// we will return later, once the descriptors are marked as such.
let progress = downloads
.values()
.find(|x| !matches!(x, DescriptorDownloadResult::NotFound))
.map(|_| ()); // Do not borrow.
// Insert the obtained descriptors in one large bulk transaction or
// mark them as futile.
// It is okay if we fail before setting the descriptors as futile here,
// because database failures are pretty much irrecoverable anyways.
// Doing this together in the same loop saves us from having to
// filter and split the large HashMap into multiple ones. It should be
// secure because the inner closure has the return error type of
// DatabaseError, meaning only these fatal errors can return early.
for (digest, res) in downloads.into_iter() {
match res {
DescriptorDownloadResult::NotFound => {
futile.routers.set_not_found(digest, endpoint);
DescriptorDownloadResult::Refused => futile.routers.set_refused(digest),
DescriptorDownloadResult::Found(descriptor, raw) => {
DescriptorMeta::insert(
&descriptor,
// If no progress, return with an error. We can do so here because
// we have finally marked descriptors as futile.
if progress.is_none() {
/// Hibernates for the remaining ttl of the consensus.
async fn hibernate(
match data {
return Err(internal!("hibernating without a consensus?").into());
// TODO DIRMIRROR: Should the Unverified case be an error or a
// success? In other words: Should we exclude the current authority
// from the set of the next authorities to try?
ConsensusBoundData::Verified { ttl, .. }
| ConsensusBoundData::Unverified { ttl, .. } => {
let timeout = *ttl - now;
debug!("hibernating for {}s", timeout.as_secs());
tokio::time::sleep(timeout).await;
/// Convenience wrapper around [`tor_dirclient::send_request()`].
/// It opens a TCP connection, performs the request, and parses the result.
/// Returns the raw response alongside the output of
/// [`parse2::parse_netdoc_multiple_sophisticated()`].
/// An important distinction towards [`tor_dirclient::send_request()`] is,
/// that we do not consider an empty response or a 404 response as an error.
/// Instead, we just return [`Ok`] with an empty [`String`] and an empty
/// [`Vec`] instead. The reason for this is, that the futile document
/// strategy (i.e. determing which documents of the request are missing)
/// should be the same for responses containing zero (requested) documents
/// and responses containing at least one but less than the number of
/// the originally requested documents.
/// The output is required because we need the raw document alongside the
/// offsets to have the actual data we will insert into the database later
/// on.
async fn send_request<R: Requestable, D: NetdocParseable>(
requ: R,
) -> Result<(String, Vec<(Result<D, ParseError>, usize, usize)>), AuthorityRequestError> {
// The check is required to not let Tokio panic.
if endpoint.is_empty() {
return Err(AuthorityRequestError::Bug(internal!("empty endpoint?")));
// Open the TCP connection.
let mut stream = TcpStream::connect(endpoint)
.await
.map_err(AuthorityRequestError::TcpConnect)?
.compat();
// Perform the request and map the result nicely.
let resp = tor_dirclient::send_request(&self.rt, &requ, &mut stream, None)
.map(|resp| {
resp.output_string()
.map(|resp| resp.to_owned())
// Allows us to flatten the result.
.map_err(tor_dirclient::Error::from)
});
// We can immediately drop the connection now, no need to occupy even
// more resources from the authority. Doing so is fine, it is HTTP/1.0
// and there is no connection reuse anyways.
drop(stream);
// Non-trivial error handling strategy explained in the rustdoc comment
// above.
let resp = match resp.flatten() {
Ok(resp) => Ok(resp),
// This error might actually be Ok in certain cases.
Err(DirClientError::RequestFailed(
ref outer @ RequestFailedError {
error: ref inner, ..
)) => match inner {
// If one of these errors occurs, do not fail but rather return
// no documents.
RequestError::EmptyRequest => {
warn!("we should not send an empty request");
return Ok(Default::default());
RequestError::EmptyResponse | RequestError::HttpStatus(404, _) => {
// All other errors imply an underlying error with the
// respective endpoint, suggesting a switch.
_ => Err(outer.clone().into()),
// Should not happen because the other DirClientErrors are either
// Bug or related to CircMgr, which we do not use.
Err(e) => Err(AuthorityRequestError::Bug(internal!(
"unhandled dirclient error: {e}"
))),
}?;
// Parse the response.
let parsed = parse2::parse_netdoc_multiple_sophisticated(&ParseInput::new(&resp, ""))?;
Ok((resp, parsed))
/// Queries and validates the most recent consensus from the database.
/// Effectively a wrapper around [`StaticEngine::verify_consensus()`] with
/// the data obtained from the database.
fn recent_consensus(
) -> Result<Option<ConsensusMeta<T>>, DatabaseError> {
let meta = match ConsensusMeta::<T>::query(tx)?.as_slice() {
&[front] | &[front, ..] => front,
&[] => return Ok(None),
let raw = meta.data(tx)?;
let unverified = match parse2::parse_netdoc::<T>(&ParseInput::new(&raw, "<database>")) {
Ok(unverified) => unverified,
// TODO DIRMIRROR: Perhaps we should remove it?
debug!("unparseable consensus in database: {meta:?} {e}");
return Ok(None);
match self.verify_consensus(unverified, &certs_already, now) {
Ok(_) => Ok(Some(meta)),
debug!("ignoring invalid consensus in database: {meta:?} {e}");
Ok(None)
/// Parses and verifies a raw consensus in a database agnostic fashion.
fn verify_consensus(
unverified: T,
certs_already: &[AuthCert],
) -> Result<(T::Body, T::Signatures), VerifyFailed> {
let sigs = unverified.sigs().clone();
let verified = unverified.verify(self.authorities.v3idents(), certs_already)?;
self.tolerance
.map(|body| (body, sigs))
.map_err(|e| e.into())
/// Loads all authority certificates from the database into a [`Vec<AuthCert>`].
/// Performance wise, it is quite inefficient as it parses and verifies all
/// certificates over again. However, this should be fine as the N for
/// directory authorities is usually very small.
fn certs_already(
) -> Result<Vec<AuthCert>, DatabaseError> {
let raw_certs = AuthCertMeta::query(tx)?
.map(|meta| Ok::<_, DatabaseError>((meta, meta.data(tx)?)))
.collect::<Result<Vec<_>, _>>()?;
let mut certs_already = Vec::new();
for (meta, raw) in raw_certs {
// Only used for debugging.
// TODO DIRMIRROR: Put this into span with tracing.
let key_ids = AuthCertKeyIds {
id_fingerprint: meta.kp_auth_id_rsa_sha1.0.into(),
sk_fingerprint: meta.kp_auth_sign_rsa_sha1.0.into(),
let unverified = match parse2::parse_netdoc::<AuthCertUnverified>(&ParseInput::new(
"<database>",
)) {
debug!("unparseable auth cert ({key_ids:?}) in database: {e}");
debug!("ignoring invalid auth cert ({key_ids:?}): {e}");
debug!("ignoring non-timely auth cert ({key_ids:?}): {e}");
certs_already.push(timely);
Ok(certs_already)
/// Records which allow us to avoid futile download attempts
/// One set of records of "futile" documents, for each kind of document.
#[derive(Debug, Clone, Default)]
struct FutileDescriptors {
/// The futile router descriptors.
routers: FutileDescriptorSet<Sha1>,
/// The futile extra-infos.
extras: FutileDescriptorSet<Sha1>,
/// The futile micro descriptors.
micros: FutileDescriptorSet<Sha256>,
/// A generic struct storing which document are futile for which reason.
/// A document is "futile" if there is no point retrying its download.
/// More precisely,
/// * We note which authorities didn't have a document when we tried to download it.
/// This avoids us trying that same document again, uselessly.
// TODO DIRMIRROR but what if the authority acquires the document in the meantime?
/// * We note the hashes of documents we couldn't parse.
/// If the hash matched, but the parse failed, there is no point trying to get
/// this document at all because the same hash means the same document
/// and therefore the same parse error.
// TODO: Right now, we allow a value to be present in both: not_found and refused.
// Do we want to allow this or should we potentially change the representation
// similar to the way we are doing it in DescriptorDownloadResult? Not blocking,
// as a hazard here is not the end of the world, but solving it may be more
// elegant.
#[derive(Educe)]
#[educe(Debug, Clone, Default)]
struct FutileDescriptorSet<T: Debug + Clone + Copy + Eq + Hash> {
/// The descriptors that are not found on a given endpoint.
not_found: HashSet<(T, Box<DownloadAuthority>)>,
/// The descriptor that are outright refused.
refused: HashSet<T>,
impl<T: Debug + Clone + Copy + Eq + Hash> FutileDescriptorSet<T> {
/// Returns an iterator over the digest that are not found.
fn not_found(&self, endpoint: &DownloadAuthority) -> impl Iterator<Item = T> {
self.not_found
.iter()
.filter(|(_, this_endpoint)| **this_endpoint == *endpoint)
.map(|(d, _)| *d)
/// Returns an iterator over the digest that are refused.
fn refused(&self) -> impl Iterator<Item = T> {
self.refused.iter().copied()
/// Chaining of `.not_found()` and `.refused()`.
fn excluded(&self, endpoint: &DownloadAuthority) -> impl Iterator<Item = T> {
self.not_found(endpoint).chain(self.refused())
/// Marks a digest in combination with an authority as not found.
fn set_not_found(&mut self, digest: T, endpoint: &DownloadAuthority) {
self.not_found.insert((digest, endpoint.into()));
/// Marks a digest as refused.
fn set_refused(&mut self, digest: T) {
self.refused.insert(digest);
/// Intermediate data structure for storing the result of a downloaded
/// descriptor within a batch of downloads.
/// Populated before the insertion into [`FutileDescriptors`] takes place,
/// so that the actual population of [`FutileDescriptors`] remains atomic.
enum DescriptorDownloadResult<D: FlavoredDescriptor> {
/// The descriptor was found and verified.
Found(D, String),
/// The descriptor was not found; implicit default.
NotFound,
/// We do not accept this descriptor.
Refused,
/// Collection of [`DescriptorDownloadResult`] values indexed by their digest.
/// Used during downloading a batch in order to atomically update
/// [`FutileDescriptors`].
#[derive(Debug, Clone, derive_more::Deref, derive_more::DerefMut, derive_more::IntoIterator)]
struct DescriptorDownloads<K, D: FlavoredDescriptor>(HashMap<K, DescriptorDownloadResult<D>>);
impl<K: Eq + Hash, D: FlavoredDescriptor> DescriptorDownloads<K, D> {
/// Creates a [`DescriptorDownloads`] from a set of all missing descriptors.
/// The idea is to call this right after the database call and before
/// any downloads are actually being made, as it initializes the data
/// structure with all the keys to [`DescriptorDownloadResult::NotFound`].
fn from_missing(digests: impl Iterator<Item = K>) -> Self {
Self(
digests
.map(|digest| (digest, DescriptorDownloadResult::NotFound))
.collect(),
)
#[cfg(test)]
mod test {
// @@ begin test lint list maintained by maint/add_warning @@
#![allow(clippy::bool_assert_comparison)]
#![allow(clippy::clone_on_copy)]
#![allow(clippy::dbg_macro)]
#![allow(clippy::mixed_attributes_style)]
#![allow(clippy::print_stderr)]
#![allow(clippy::print_stdout)]
#![allow(clippy::single_char_pattern)]
#![allow(clippy::unwrap_used)]
#![allow(clippy::unchecked_time_subtraction)]
#![allow(clippy::useless_vec)]
#![allow(clippy::needless_pass_by_value)]
#![allow(clippy::string_slice)] // See arti#2571
//! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
use std::{collections::HashSet, iter};
use rusqlite::params;
use strum::IntoEnumIterator;
use tokio::{
io::{AsyncReadExt, AsyncWriteExt},
net::TcpListener,
use tor_basic_utils::test_rng::testing_rng;
use crate::{database::sql, err::IsFatal, testdata2};
use super::*;
type Plain = tor_netdoc::doc::netstatus::plain::NetworkStatusUnverified;
type Md = tor_netdoc::doc::netstatus::md::NetworkStatusUnverified;
/// Returns a [`StaticEngine`] for testing.
fn static_engine<T>() -> StaticEngine<T> {
StaticEngine {
authorities: testdata2::current_auth_cert_contacts(),
tolerance: DirTolerance::default(),
rt: PreferredRuntime::current().unwrap(),
encodings: ContentEncoding::iter().collect(),
_phantom: Default::default(),
/// Tests whether the load consensus state computes missing descriptors
/// properly.
/// For this, the test removes a present router descriptor from the storage
/// to verify that it is detected as missing and added to the download
/// queue.
#[tokio::test]
async fn state_load_consensus() {
let pool = testdata2::test_db();
let mut data = ConsensusBoundData::<Plain>::None;
let engine = static_engine();
let time: Timestamp = testdata2::valid_system_time().into();
let fresh_until: Timestamp = testdata2::current_consensus_ns()
.0
.preamble
.lifetime
.fresh_until
.into();
let valid_until: Timestamp = testdata2::current_consensus_ns()
.valid_until
// This is the middle of valid_until and fresh_until.
let fresh_until_half = fresh_until + ((valid_until - fresh_until) / 2);
// Remove a single router descriptor from our storage to see whether it
// appears in the download queue as expected.
let relay_to_remove = &testdata2::current_consensus_ns().0.routers[0];
let relay_to_remove = db::Sha1::from(*relay_to_remove.doc_digest());
pool.get()
.unwrap()
.execute(
sql!("DELETE FROM descriptor WHERE sha1 = ?1"),
params![relay_to_remove],
.unwrap();
engine
.load_consensus(&pool, &mut data, time, &mut testing_rng())
// El-cheapo assert_eq due to lack of PartialEq for tor-netdoc poc.
ConsensusBoundData::Verified { consensus, ttl, .. } => {
// If everything worked properly, then the queue should only
// contain the relay we removed, because that is missing now.
db::read_tx(&pool, |tx| {
assert_eq!(
consensus.missing_routers(tx, iter::empty(), None).unwrap(),
HashSet::from([relay_to_remove])
assert!(
consensus
.missing_extras(tx, iter::empty(), None)
.is_empty()
.missing_micros(tx, iter::empty(), None)
assert!(ttl >= fresh_until);
assert!(ttl <= fresh_until_half);
_ => panic!("data is not verified"),
/// Tests whether the fetch consensus state properly fetches a consensus
/// and keeps it in memory as unverified.
/// For this, we spawn a tokio task simulating a web server which responds
/// with a consensus.
async fn state_fetch_consensus() {
let now = Timestamp::from(testdata2::invalid_system_time());
let state = db::read_tx(&pool, |tx| engine.determine_state(tx, &data, now))
assert_eq!(state, State::FetchConsensus);
let server = TcpListener::bind("[::1]:0").await.unwrap();
let saddr = server.local_addr().unwrap();
tokio::spawn(async move {
let (mut stream, _) = server.accept().await.unwrap();
let mut buf = vec![0; 1024];
let _ = stream.read(&mut buf).await.unwrap();
let consensus = testdata2::current_consensus_ns().2;
let resp = format!(
"HTTP/1.0 200 OK\r\nContent-Encoding: identity\r\nContent-Length: {}\r\n\r\n{consensus}",
consensus.len()
stream.write_all(resp.as_bytes()).await.unwrap();
.fetch_consensus(&mut data, &[saddr], now)
ConsensusBoundData::Unverified { raw, ttl, .. } => {
assert_eq!(raw, testdata2::current_consensus_ns().2);
assert_eq!(ttl, now + UNVERIFIED_TTL);
_ => panic!("data is not unverified"),
/// Tests the download, verification, and insertion of authority certificates.
/// For this, it starts by removing an existing one from the test database
/// to see it getting re-downloaded, re-verified, and re-inserted again.
async fn state_auth_certs() {
let mut data = ConsensusBoundData::<Plain>::Unverified {
consensus: parse2::parse_netdoc(&ParseInput::new(
testdata2::current_consensus_ns().2,
"",
))
.unwrap(),
raw: testdata2::current_consensus_ns().2.to_owned(),
ttl: Timestamp::from(testdata2::valid_system_time()) + UNVERIFIED_TTL,
let mut engine = static_engine();
// We want to download authority certificates; for this, remove
// one of them from the database.
sql!(
"
DELETE FROM authority_key_certificate
WHERE :kp_auth_id_rsa_sha1 = ?1
),
params![db::Sha1::from(
testdata2::current_auth_cert_ids()[0].to_bytes()
)],
db::read_tx(&pool, |tx| engine.determine_state(
&data,
testdata2::valid_system_time().into()
State::AuthCerts
let mut buf = [0; 1024];
let authcerts = testdata2::current_auth_certs()
.map(|x| x.1)
.collect::<String>();
stream.write_all(format!(
"HTTP/1.0 200 OK\r\nContent-Encoding: identity\r\nContent-Length: {}\r\n\r\n{authcerts}",
authcerts.len()
).as_bytes()).await.unwrap();
// Fetch all authcerts.
.auth_certs(
&pool,
&mut data,
&[saddr],
testdata2::valid_system_time().into(),
// Check whether we are done with all authcerts.
State::StoreConsensus
// Check whether we now have all auth certs in the database.
let certs = db::read_tx(&pool, |tx| {
engine.certs_already(tx, testdata2::valid_system_time().into())
let got = certs
.map(|c| c.dir_identity_key.to_rsa_identity())
.collect::<HashSet<_>>();
let expect = engine.authorities.v3idents().iter().copied().collect();
assert_eq!(got, expect);
// Check an edge case of not having enough trusted signers.
engine.authorities = AuthorityContacts::builder().build().unwrap();
let state = db::read_tx(&pool, |tx| {
engine.determine_state(tx, &data, testdata2::valid_system_time().into())
assert_eq!(state, State::Hibernate);
/// Checks whether we can probably load and parse all certificates from
/// the database.
async fn certs_already() {
let engine = static_engine::<Plain>();
// With a correct system time, this should align with all authority
// fingerprints we have.
let certs_already = engine
.certs_already(tx, testdata2::valid_system_time().into())
assert_eq!(certs_already.len(), engine.authorities.v3idents().len());
assert!(certs_already.iter().all(|c| {
.authorities
.v3idents()
.contains(&c.dir_identity_key.to_rsa_identity())
}));
// With an invalid system time, it should be empty.
.certs_already(tx, testdata2::invalid_system_time().into())
assert!(certs_already.is_empty());
/// Ensures that an unverified consensus gets properly inserted into the
async fn state_store_consensus() {
let unverified: Plain =
parse2::parse_netdoc(&ParseInput::new(testdata2::current_consensus_ns().2, ""))
let now = Timestamp::from(testdata2::valid_system_time());
consensus: unverified.clone(),
raw: testdata2::current_consensus_ns().2.to_string(),
ttl: now + UNVERIFIED_TTL,
// Clean all consensus related tables.
.execute_batch(sql!(
DELETE FROM consensus_descriptor_member;
DELETE FROM consensus_authority_voter;
DELETE FROM consensus;
// Execute the StoreConsensus state.
assert_eq!(state, State::StoreConsensus);
engine.store_consensus(&pool, &mut data, now).unwrap();
assert!(matches!(&data, ConsensusBoundData::None));
// Verify that it got inserted.
assert_eq!(state, State::LoadConsensus);
.load_consensus(&pool, &mut data, now, &mut testing_rng())
assert!(matches!(data, ConsensusBoundData::Verified { .. }));
/// Tests the descriptor bootstrapping for router descriptors.
async fn state_descriptors_plain() {
let mut data = ConsensusBoundData::None;
let rng = &mut testing_rng();
// Server task that can at most handle two requests.
// The first response contains all available router descriptors
// statically, whereas the second one only contains a single router
// descriptor.
let listener = TcpListener::bind("[::1]:0").await.unwrap();
let endpoint = listener.local_addr().unwrap();
let full = testdata2::current_router_descs()
.map(|(_, _, raw)| raw)
let single = testdata2::current_router_descs()
.take(1)
let (mut stream, _) = listener.accept().await.unwrap();
stream
.write_all(
format!(
"HTTP/1.0 200 OK\r\nContent-Length: {}\r\n\r\n{}",
full.len(),
full
.as_bytes(),
single.len(),
single
// Delete all router descriptors from the database and then load
// a consensus into ConsensusBoundData.
db::rw_tx(&pool, |tx| tx.execute(sql!("DELETE FROM descriptor"), ()))
engine.load_consensus(&pool, &mut data, now, rng).unwrap();
assert_eq!(state, State::Descriptors);
// Successful bootstrap of all descriptors.
.descriptors_rd(&pool, &mut data, &[endpoint], now)
// Afterwards, the queue for missing router descriptors should be empty
// as well as futile.
let (meta, futile) = match &data {
consensus: meta,
..
} => (*meta, futile.clone()),
_ => panic!(),
let missing_routers =
db::read_tx(&pool, |tx| meta.missing_routers(tx, iter::empty(), None))
assert!(missing_routers.is_empty());
assert!(futile.routers.not_found.is_empty());
assert!(futile.routers.refused.is_empty());
// Now, if we do this again, futile should be populated.
// This will yield a non-fatal error because it makes no progress, but
// that is okay; we primarily want to test whether the futile
// descriptor handling works.
let err = engine
.unwrap_err();
assert!(!err.is_fatal());
// Now, the missing descriptors should equal the ones that were not
// found on the authority, as in being in the futile set.
let futile = match &data {
ConsensusBoundData::Verified { futile, .. } => futile,
assert!(!missing_routers.is_empty());
missing_routers,
futile
.routers
.not_found
.filter_map(|(digest, this_endpoint)| if **this_endpoint == [endpoint] {
Some(*digest)
None
.collect::<HashSet<_>>()