add correctness check for header related content

Author: perama-v

Cargo.lock

@@ -19,11 +19,12 @@ ethportal-api = { git = "https://github.com/ethereum/trin" }
 glados-core = { path = "../glados-core" }
 migration = { path = "../migration" }
 rand = "0.8.5"
+reqwest = "0.11.16"
 sea-orm = "0.10.3"
 serde_json = "1.0.95"
+thiserror = "1.0.40"
 tokio = "1.21.2"
 tracing = "0.1.37"
 trin-utils = {git = "https://github.com/ethereum/trin" }
 url = "2.3.1"
 web3 = "0.18.0"
-

glados-audit/Cargo.toml

@@ -5,21 +5,46 @@ use url::Url;
 
 const DEFAULT_DB_URL: &str = "sqlite::memory:";
 
-#[derive(Parser, Debug, Eq, PartialEq)]
+#[derive(Parser, Default, Debug, Eq, PartialEq)]
 #[command(author, version, about, long_about = None)]
 pub struct Args {
+    /// Database connection URL, such as SQLite or PostgreSQL.
     #[arg(short, long, default_value = DEFAULT_DB_URL)]
     pub database_url: String,
+
+    /// IPC Path for connection to Portal node.
     #[arg(short, long, requires = "transport")]
     pub ipc_path: Option<PathBuf>,
+
+    /// HTTP URL for connection to Portal node.
     #[arg(short = 'u', long, requires = "transport")]
     pub http_url: Option<Url>,
+
+    /// Portal node connection mode.
     #[arg(short, long)]
     pub transport: TransportType,
+
     #[arg(short, long, default_value = "4", help = "number of auditing threads")]
     pub concurrency: u8,
-    #[arg(short, long, action(ArgAction::Append), value_enum, default_value = None, help = "Specific strategy to use. Default is to use all available strategies. May be passed multiple times for multiple strategies (--strategy latest --strategy random). Duplicates are permitted (--strategy random --strategy random).")]
+
+    /// Specific strategy to use. Default is to use all available strategies.
+    /// May be passed multiple times for multiple strategies
+    /// (--strategy latest --strategy random).
+    /// Duplicates are permitted (--strategy random --strategy random).
+    #[arg(short, long, action(ArgAction::Append), value_enum, default_value = None)]
     pub strategy: Option<Vec<SelectionStrategy>>,
+
+    /// Non-portal Ethereum execution node for validating data against.
+    #[arg(long, default_value = "http")]
+    pub trusted_provider: TrustedProvider,
+
+    /// HTTP URL for connection to non-portal Ethereum execution node.
+    #[arg(long, requires = "trusted_provider")]
+    pub provider_http_url: Option<Url>,
+
+    /// Pandaops URL for connection to non-portal Ethereum execution node.
+    #[arg(long, requires = "trusted_provider")]
+    pub provider_pandaops: Option<String>,
 }
 
 #[cfg(test)]
@@ -37,7 +62,7 @@ mod test {
             concurrency: 4,
             http_url: None,
             transport: TransportType::IPC,
-            strategy: None,
+            ..Default::default()
         };
         assert_eq!(result, expected);
     }
@@ -59,7 +84,7 @@ mod test {
             concurrency: 3,
             http_url: None,
             transport: TransportType::IPC,
-            strategy: None,
+            ..Default::default()
         };
         assert_eq!(result, expected);
     }
@@ -84,6 +109,7 @@ mod test {
             ipc_path: Some(PathBuf::from(IPC_PATH)),
             concurrency: 4,
             strategy: Some(vec![SelectionStrategy::Latest]),
+            ..Default::default()
         };
         assert_eq!(result, expected);
     }
@@ -117,16 +143,30 @@ mod test {
                 SelectionStrategy::Latest,
                 SelectionStrategy::Random,
             ]),
+            ..Default::default()
         };
         assert_eq!(result, expected);
     }
 }
 
 /// Used by a user to specify the intended form of transport
 /// to connect to a Portal node.
-#[derive(Debug, Clone, Eq, PartialEq, ValueEnum)]
+#[derive(Debug, Default, Clone, Eq, PartialEq, ValueEnum)]
 #[clap(rename_all = "snake_case")]
 pub enum TransportType {
+    #[default]
+    HTTP,
     IPC,
+}
+
+/// Used by a user to specify the intended form of transport
+/// to connect to a Portal node.
+#[derive(Debug, Default, Clone, Eq, PartialEq, ValueEnum)]
+#[clap(rename_all = "snake_case")]
+pub enum TrustedProvider {
+    #[default]
+    /// An HTTP-based provider.
     HTTP,
+    /// A Trin operations provider.
+    Pandaops,
 }

glados-audit/src/cli.rs

@@ -1,28 +1,32 @@
 use std::{
+    env,
     sync::{
         atomic::{AtomicU8, Ordering},
         Arc,
     },
     thread::available_parallelism,
 };
 
-use anyhow::{bail, Result};
+use anyhow::{anyhow, bail};
 use clap::Parser;
 use cli::Args;
+use entity::{
+    content,
+    content_audit::{self, SelectionStrategy},
+    execution_metadata,
+};
 use ethportal_api::{HistoryContentKey, OverlayContentKey};
+use glados_core::jsonrpc::{PortalClient, TransportConfig};
+use reqwest::header::{HeaderMap, HeaderValue};
 use sea_orm::DatabaseConnection;
 use tokio::{
     sync::mpsc,
     time::{sleep, Duration},
 };
 use tracing::{debug, error, info, warn};
-
-use entity::{
-    content,
-    content_audit::{self, SelectionStrategy},
-    execution_metadata,
-};
-use glados_core::jsonrpc::{PortalClient, TransportConfig};
+use url::Url;
+use validation::Provider;
+use web3::{transports::Http, Web3};
 
 use crate::{
     cli::TransportType, selection::start_audit_selection_task, validation::content_is_valid,
@@ -43,10 +47,13 @@ pub struct AuditConfig {
     pub transport: TransportConfig,
     /// Specific strategies to run.
     pub strategies: Vec<SelectionStrategy>,
+    /// An Ethereum execution node to validate content received from
+    /// the Portal node against.
+    pub trusted_provider: Provider,
 }
 
 impl AuditConfig {
-    pub fn from_args() -> Result<AuditConfig> {
+    pub fn from_args() -> anyhow::Result<AuditConfig> {
         let args = Args::parse();
         let transport: TransportConfig = match args.transport {
             TransportType::IPC => match args.ipc_path {
@@ -87,11 +94,50 @@ impl AuditConfig {
                 ]
             }
         };
+        let trusted_provider: Provider = match args.trusted_provider {
+            cli::TrustedProvider::HTTP => {
+                match args.provider_http_url{
+                    Some(url) => {
+                        let transport = Http::new(url.as_str())?;
+                        let w3 = Web3::new(transport);
+                        Provider::Http(w3)
+                    },
+                    None => bail!("The '--provider-http-url' flag is required if 'http' is selected for the '--trusted-provider'"),
+                }
+            },
+            cli::TrustedProvider::Pandaops => {
+                match args.provider_pandaops {
+                    Some(provider_url) => {
+                        let mut headers = HeaderMap::new();
+                        let client_id = env::var("PANDAOPS_CLIENT_ID")
+                            .map_err(|_| anyhow!("PANDAOPS_CLIENT_ID env var not set."))?;
+                        let client_id = HeaderValue::from_str(&client_id);
+                        let client_secret = env::var("PANDAOPS_CLIENT_SECRET")
+                            .map_err(|_| anyhow!("PANDAOPS_CLIENT_SECRET env var not set."))?;
+                        let client_secret = HeaderValue::from_str(&client_secret);
+                        headers.insert("CF-Access-Client-Id", client_id?);
+                        headers.insert("CF-Access-Client-Secret", client_secret?);
+
+                        let client = reqwest::Client::builder()
+                            .default_headers(headers)
+                            .build()?;
+                        let url = Url::parse(&provider_url)?;
+                        let transport = Http::with_client(client, url);
+                        let w3 = Web3::new(transport);
+                        Provider::Pandaops(w3)
+                    },
+                    None => bail!("The '--provider-pandaops' flag is required if 'pandaops' is selected for the '--trusted-provider'"),
+                }
+            }
+        }
+        ;
+
         Ok(AuditConfig {
             concurrency: args.concurrency,
             database_url: args.database_url,
             transport,
             strategies,
+            trusted_provider,
         })
     }
 }
@@ -201,9 +247,21 @@ async fn perform_single_audit(
         }
     };
 
-    // If content was absent audit result is 'fail'.
+    // If content was absent or invalid the audit result is 'fail'.
     let audit_result = match content_response {
-        Some(content_bytes) => content_is_valid(&task.content_key, &content_bytes.raw),
+        Some(content_bytes) => {
+            match content_is_valid(&config, &task.content_key, &content_bytes.raw).await {
+                Ok(res) => res,
+                Err(e) => {
+                    error!(
+                        content.key=?task.content_key.to_hex(),
+                        err=?e,
+                        "Problem requesting validation from Trusted provider node.");
+                    active_threads.fetch_sub(1, Ordering::Relaxed);
+                    return;
+                }
+            }
+        }
         None => false,
     };