dirmgr: Remember where netdocs came from.

This isn't complete (see TODO), but it's enough to let us report the
right ErrorKind if something fails to parse.

Cargo.lock

@@ -3044,6 +3044,7 @@ dependencies = [
  "async-trait",
  "base64",
  "derive_builder",
+ "derive_more",
  "digest 0.10.1",
  "event-listener",
  "float_eq",

crates/tor-dirmgr/Cargo.toml

@@ -33,6 +33,7 @@ tor-rtcompat = { path = "../tor-rtcompat", version = "0.0.4"}
 async-trait = "0.1.2"
 base64 = "0.13.0"
 derive_builder = "0.10"
+derive_more = "0.99"
 digest = "0.10.0"
 event-listener = "2"
 futures = "0.3.14"

crates/tor-dirmgr/src/err.rs

@@ -2,6 +2,7 @@
 
 use std::sync::Arc;
 
+use crate::DocSource;
 use futures::task::SpawnError;
 use thiserror::Error;
 use tor_error::{ErrorKind, HasKind};
@@ -59,8 +60,14 @@ pub enum Error {
     #[error("Invalid hexadecimal id in directory cache")]
     BadHexInCache(#[source] hex::FromHexError),
     /// An error given by the network document crate.
-    #[error("netdoc error: {0}")]
+    #[error("netdoc error from {source}: {cause}")]
-    NetDocError(#[from] tor_netdoc::Error),
+    NetDocError {
+        /// Where the document came from.
+        source: DocSource,
+        /// What error we got.
+        #[source]
+        cause: tor_netdoc::Error,
+    },
     /// An error given by dirclient
     #[error("dirclient error: {0}")]
     DirClientError(#[from] tor_dirclient::Error),
@@ -109,6 +116,13 @@ impl Error {
             cause: Arc::new(err),
         }
     }
+
+    /// Construct a new `Error` from `tor_netdoc::Error`.
+    ///
+    /// Also takes a source so that we can keep track of where the document came from.
+    pub(crate) fn from_netdoc(source: DocSource, cause: tor_netdoc::Error) -> Error {
+        Error::NetDocError { source, cause }
+    }
 }
 
 impl From<rusqlite::Error> for Error {
@@ -143,7 +157,10 @@ impl HasKind for Error {
             E::CantAdvanceState => EK::DirectoryStalled,
             E::StorageError(_) => EK::CacheAccessFailed,
             E::ConsensusDiffError(_) => EK::TorProtocolViolation,
-            E::NetDocError(_) => todo!(), // TODO: depends on source
+            E::NetDocError { source, .. } => match source {
+                DocSource::LocalCache => EK::CacheCorrupted,
+                DocSource::DirServer { .. } => EK::TorProtocolViolation,
+            },
             E::DirClientError(e) => e.kind(),
             E::SignatureError(_) => EK::TorProtocolViolation,
             E::IOError(_) => EK::CacheAccessFailed,

crates/tor-dirmgr/src/lib.rs

@@ -181,6 +181,24 @@ impl<'a> BoolResetter<'a> {
     }
 }
 
+/// The possible origins of a document.
+///
+/// Used (for example) to report where we got a document from if it fails to
+/// parse.
+#[derive(Debug, Clone, derive_more::Display)]
+#[non_exhaustive]
+pub enum DocSource {
+    /// We loaded the document from our cache.
+    #[display(fmt = "local cache")]
+    LocalCache,
+    /// We fetched the document from a server.
+    //
+    // TODO: we'll should add a lot more information here in the future, once
+    // it's available from tor_dirclient::DirSource,
+    #[display(fmt = "directory server")]
+    DirServer {},
+}
+
 impl<R: Runtime> DirMgr<R> {
     /// Try to load the directory from disk, without launching any
     /// kind of update process.

crates/tor-dirmgr/src/state.rs

@@ -22,7 +22,6 @@ use tor_netdoc::doc::netstatus::Lifetime;
 use tracing::{info, warn};
 
 use crate::event::{DirStatus, DirStatusInner};
-use crate::DirEvent;
 use crate::{
     docmeta::{AuthCertMeta, ConsensusMeta},
     retry::DownloadSchedule,
@@ -31,6 +30,7 @@ use crate::{
     CacheUsage, ClientRequest, DirMgrConfig, DirState, DocId, DocumentText, Error, Readiness,
     Result,
 };
+use crate::{DirEvent, DocSource};
 use tor_checkable::{ExternallySigned, SelfSigned, Timebound};
 use tor_llcrypto::pk::rsa::RsaIdentity;
 use tor_netdoc::doc::{
@@ -229,7 +229,8 @@ impl<DM: WriteNetDir> DirState for GetConsensusState<DM> {
             _ => return Err(Error::Unwanted("Not an md consensus")),
         };
 
-        self.add_consensus_text(true, text.as_str().map_err(Error::BadUtf8InCache)?)
+        let source = DocSource::LocalCache;
+        self.add_consensus_text(source, text.as_str().map_err(Error::BadUtf8InCache)?)
             .map(|meta| meta.is_some())
     }
     fn add_from_download(
@@ -238,7 +239,8 @@ impl<DM: WriteNetDir> DirState for GetConsensusState<DM> {
         _request: &ClientRequest,
         storage: Option<&Mutex<SqliteStore>>,
     ) -> Result<bool> {
-        if let Some(meta) = self.add_consensus_text(false, text)? {
+        let source = DocSource::DirServer {};
+        if let Some(meta) = self.add_consensus_text(source, text)? {
             if let Some(store) = storage {
                 let mut w = store.lock().expect("Directory storage lock poisoned");
                 w.store_consensus(meta, ConsensusFlavor::Microdesc, true, text)?;
@@ -268,12 +270,13 @@ impl<DM: WriteNetDir> GetConsensusState<DM> {
     /// correct, or if it is ill-formed.
     fn add_consensus_text(
         &mut self,
-        from_cache: bool,
+        source: DocSource,
         text: &str,
     ) -> Result<Option<&ConsensusMeta>> {
         // Try to parse it and get its metadata.
         let (consensus_meta, unvalidated) = {
-            let (signedval, remainder, parsed) = MdConsensus::parse(text)?;
+            let (signedval, remainder, parsed) =
+                MdConsensus::parse(text).map_err(|e| Error::from_netdoc(source.clone(), e))?;
             let now = current_time(&self.writedir)?;
             if let Ok(timely) = parsed.check_valid_at(&now) {
                 let meta = ConsensusMeta::from_unvalidated(signedval, remainder, &timely);
@@ -303,7 +306,7 @@ impl<DM: WriteNetDir> GetConsensusState<DM> {
 
         self.next = Some(GetCertsState {
             cache_usage: self.cache_usage,
-            from_cache,
+            consensus_source: source,
             unvalidated,
             consensus_meta,
             missing_certs: desired_certs,
@@ -334,9 +337,8 @@ impl<DM: WriteNetDir> GetConsensusState<DM> {
 struct GetCertsState<DM: WriteNetDir> {
     /// The cache usage we had in mind when we began.  Used to reset.
     cache_usage: CacheUsage,
-    /// True iff we loaded the consensus from our cache.
+    /// Where did we get our consensus?
-    #[allow(dead_code)]
+    consensus_source: DocSource,
-    from_cache: bool,
     /// The consensus that we are trying to validate.
     unvalidated: UnvalidatedMdConsensus,
     /// Metadata for the consensus.
@@ -398,7 +400,9 @@ impl<DM: WriteNetDir> DirState for GetCertsState<DM> {
         // our input and remembering them.
         for id in &self.missing_docs() {
             if let Some(cert) = docs.get(id) {
-                let parsed = AuthCert::parse(cert.as_str().map_err(Error::BadUtf8InCache)?)?
+                let text = cert.as_str().map_err(Error::BadUtf8InCache)?;
+                let parsed = AuthCert::parse(text)
+                    .map_err(|e| Error::from_netdoc(DocSource::LocalCache, e))?
                     .check_signature()?;
                 let now = current_time(&self.writedir)?;
                 if let Ok(cert) = parsed.check_valid_at(&now) {
@@ -482,7 +486,11 @@ impl<DM: WriteNetDir> DirState for GetCertsState<DM> {
     }
     fn advance(self: Box<Self>) -> Result<Box<dyn DirState>> {
         if self.can_advance() {
-            let validated = self.unvalidated.check_signature(&self.certs[..])?;
+            let consensus_source = self.consensus_source.clone();
+            let validated = self
+                .unvalidated
+                .check_signature(&self.certs[..])
+                .map_err(|e| Error::from_netdoc(consensus_source, e))?;
             Ok(Box::new(GetMicrodescsState::new(
                 self.cache_usage,
                 validated,
@@ -1086,7 +1094,7 @@ mod test {
         let req = tor_dirclient::request::ConsensusRequest::new(ConsensusFlavor::Microdesc);
         let req = crate::docid::ClientRequest::Consensus(req);
         let outcome = state.add_from_download("this isn't a consensus", &req, Some(&store));
-        assert!(matches!(outcome, Err(Error::NetDocError(_))));
+        assert!(matches!(outcome, Err(Error::NetDocError { .. })));
         // make sure it wasn't stored...
         assert!(store
             .lock()

crates/tor-dirmgr/src/storage/sqlite.rs

@@ -704,7 +704,8 @@ fn cmeta_from_row(row: &rusqlite::Row<'_>) -> Result<ConsensusMeta> {
     let vu: OffsetDateTime = row.get(2)?;
     let d_signed: String = row.get(3)?;
     let d_all: String = row.get(4)?;
-    let lifetime = Lifetime::new(va.into(), fu.into(), vu.into())?;
+    let lifetime = Lifetime::new(va.into(), fu.into(), vu.into())
+        .map_err(|_| Error::CacheCorruption("inconsistent lifetime in database"))?;
     Ok(ConsensusMeta::new(
         lifetime,
         digest_from_hex(&d_signed)?,