diff --git a/Cargo.lock b/Cargo.lock index d2317c0..1c613c5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3218,6 +3218,7 @@ dependencies = [ "csv", "entities", "git2", + "miette", "migration", "regex", "reqwest", diff --git a/entities/src/state/scanner.rs b/entities/src/state/scanner.rs index a98afa1..5ad98ab 100644 --- a/entities/src/state/scanner.rs +++ b/entities/src/state/scanner.rs @@ -41,6 +41,8 @@ pub struct Config { pub cleanup_interval: Duration, /// Amount of latest errors to keep per instance/host pub error_retention_per_host: usize, + /// Path for connectivity checks + pub connectivity_path: String, } impl Config { @@ -65,6 +67,7 @@ impl Config { bad_hosts: vec![String::from("tweet.whateveritworks.org")], cleanup_interval: Duration::from_secs(24*60*60), error_retention_per_host: 100, + connectivity_path: String::from("/"), }) } } diff --git a/scanner/Cargo.toml b/scanner/Cargo.toml index 9528058..6629f5d 100644 --- a/scanner/Cargo.toml +++ b/scanner/Cargo.toml @@ -10,6 +10,7 @@ reqwest = { workspace = true, features = ["deflate","gzip","brotli","cookies", " sea-orm = { workspace = true, features = [ "sqlx-sqlite", "runtime-tokio-native-tls", "macros" ] } sea-query = { workspace = true } thiserror = { workspace = true } +miette = { workspace = true } tracing = { workspace = true } tokio = { workspace = true, features = ["full"] } scraper = "0.17.1" diff --git a/scanner/src/cache_update.rs b/scanner/src/cache_update.rs index 93c47a7..0ab0208 100644 --- a/scanner/src/cache_update.rs +++ b/scanner/src/cache_update.rs @@ -1,5 +1,5 @@ -use std::cmp; // SPDX-License-Identifier: AGPL-3.0-only +use std::cmp; use std::collections::HashMap; use chrono::{TimeZone, Duration}; diff --git a/scanner/src/lib.rs b/scanner/src/lib.rs index 0823b98..b480378 100644 --- a/scanner/src/lib.rs +++ b/scanner/src/lib.rs @@ -2,31 +2,26 @@ use std::{ net::IpAddr, sync::{Arc, Mutex}, - time::Instant, }; use about_parser::AboutParser; use chrono::{Utc, DateTime, TimeZone, Duration}; -use entities::{ - host, - prelude::*, - state::{error_cache::HostError, scanner::ScannerConfig, AppState}, health_check, -}; +use entities::{state::{error_cache::HostError, scanner::ScannerConfig, AppState}, health_check}; use instance_parser::InstanceParser; +use miette::{IntoDiagnostic, Context}; use profile_parser::ProfileParser; use regex::{Regex, RegexBuilder}; use reqwest::{ header::{HeaderMap, HeaderValue}, - Client, ClientBuilder, Url, + Client, ClientBuilder }; use sea_orm::{ - sea_query::OnConflict, ActiveModelTrait, ActiveValue, ColumnTrait, ConnectionTrait, - DatabaseConnection, DbBackend, EntityTrait, FromQueryResult, QueryFilter, Statement, - TransactionTrait, QuerySelect, + ColumnTrait, ConnectionTrait, + DatabaseConnection, DbBackend, EntityTrait, FromQueryResult, Statement, + QuerySelect, }; use thiserror::Error; -use tokio::{task::JoinSet, time::sleep}; -use tracing::instrument; +use tokio::time::sleep; type Result = std::result::Result; @@ -37,6 +32,7 @@ mod instance_parser; mod profile_parser; mod version_check; mod cleanup; +mod list_update; const CAPTCHA_TEXT: &'static str = "Enable JavaScript and cookies to continue"; const CAPTCHA_CODE: u16 = 403; @@ -110,13 +106,21 @@ impl FetchError { } } +#[derive(Debug, FromQueryResult, Default)] +pub(crate) struct LatestCheck { + pub host: i32, + pub healthy: bool, + pub domain: String, +} + pub async fn run_scanner( db: DatabaseConnection, config: ScannerConfig, app_state: AppState, disable_health_checks: bool, -) -> Result<()> { - let scanner = Scanner::new(db, config, app_state).await?; +) -> miette::Result<()> { + let scanner = Scanner::new(db, config, app_state).await + .wrap_err("Initializing scanner!")?; scanner.schedule_cleanup().unwrap(); if disable_health_checks { @@ -124,7 +128,7 @@ pub async fn run_scanner( return Ok(()) } tokio::spawn(async move { - scanner.run().await.unwrap(); + scanner.run().await.expect("Failed to run scanner daemon!"); }); Ok(()) @@ -150,13 +154,6 @@ struct InnerScanner { client_ipv6: Client, } -#[derive(Debug, FromQueryResult, Default)] -pub struct LatestCheck { - pub host: i32, - pub healthy: bool, - pub domain: String, -} - impl Scanner { pub(crate) fn client_builder(config: &ScannerConfig) -> ClientBuilder { let mut headers = HeaderMap::with_capacity(HEADERS.len()); @@ -181,20 +178,22 @@ impl Scanner { db: DatabaseConnection, config: ScannerConfig, app_state: AppState, - ) -> Result { + ) -> miette::Result { let mut builder_regex_rss = RegexBuilder::new(&config.rss_content); builder_regex_rss.case_insensitive(true); let http_client = Self::client_builder(&config); let client_ipv4 = Scanner::client_builder(&config) .local_address("0.0.0.0".parse::().unwrap()) .build() - .unwrap(); + .into_diagnostic()?; let client_ipv6 = Scanner::client_builder(&config) .local_address("::".parse::().unwrap()) .build() - .unwrap(); + .into_diagnostic()?; - let last_uptime_check = Self::query_last_fetch(&db).await?; + let last_uptime_check = Self::query_last_fetch(&db).await + .into_diagnostic() + .wrap_err("Fetching last uptime check failed!")?; tracing::info!(?last_uptime_check); let scanner = Self { inner: Arc::new(InnerScanner { @@ -211,10 +210,12 @@ impl Scanner { last_uptime_check: Mutex::new(last_uptime_check), rss_check_regex: builder_regex_rss .build() - .expect("Invalid RSS Content regex!"), + .into_diagnostic() + .wrap_err("Invalid RSS Content regex!")?, }), }; - scanner.update_cache().await?; + scanner.update_cache().await.into_diagnostic() + .wrap_err("Initial cache update failed!")?; Ok(scanner) } @@ -228,6 +229,29 @@ impl Scanner { Ok(Utc.timestamp_opt(time,0).unwrap()) } + pub async fn query_latest_check( + &self, + connection: &T, + ) -> Result> { + let health_checks = LatestCheck::find_by_statement(Statement::from_sql_and_values( + DbBackend::Sqlite, + r#" + WITH latest AS( + SELECT u.host,MAX(u.time) as time FROM health_check u + GROUP BY u.host + ) + SELECT u.host,healthy,h.domain FROM health_check u + JOIN host h ON h.id = u.host + JOIN latest l ON l.host = u.host AND l.time = u.time + WHERE h.enabled = true + "#, + [], + )) + .all(connection) + .await?; + Ok(health_checks) + } + pub async fn run(mut self) -> Result<()> { loop { if self.is_instance_list_outdated() { @@ -283,158 +307,11 @@ impl Scanner { Utc::now().signed_duration_since(val).to_std().unwrap() >= self.inner.config.list_fetch_interval } - #[instrument] - async fn update_instacelist(&mut self) -> Result<()> { - let start = Instant::now(); - let html: String = self.fetch_instance_list().await?; - let parsed_instances = self.inner.instance_parser.parse_instancelist( - &html, - &self.inner.config.additional_hosts, - &self.inner.config.additional_host_country, - false, - )?; - - let transaction = self.inner.db.begin().await?; - - // find all currently enabled instances - let enabled_hosts = Host::find() - .filter(host::Column::Enabled.eq(true)) - .all(&transaction) - .await?; - // make a diff and remove the ones not found while parsing - let time: chrono::DateTime = Utc::now(); - let mut removed = 0; - for host in enabled_hosts.iter() { - if !parsed_instances.contains_key(&host.domain) { - host::ActiveModel { - id: ActiveValue::Set(host.id), - enabled: ActiveValue::Set(false), - updated: ActiveValue::Set(time.timestamp()), - ..Default::default() - } - .update(&transaction) - .await?; - removed += 1; - } - } - // now update/insert the existing ones - let found_instances: usize = parsed_instances.len(); - // find last update checks to detect spam - let last_status = self.query_latest_check(&transaction).await?; - let mut join_set = JoinSet::new(); - for (_, instance) in parsed_instances { - // TODO: parallelize this! - let scanner_c = self.clone(); - // detect already offline host and prevent log spam - let muted_host = match self.inner.config.auto_mute { - false => false, - true => last_status - .iter() - .find(|v| v.domain == instance.domain) - .map_or(false, |check| !check.healthy), - }; - // tracing::trace!(muted_host,instance=?instance,last_status=?last_status); - join_set.spawn(async move { - let (connectivity, rss, version, version_url) = match Url::parse(&instance.url) { - Err(_) => { - if !muted_host { - tracing::info!(url = instance.url, "Instance URL invalid"); - } - (None, false, None, None) - } - Ok(mut url) => { - let connectivity = scanner_c.check_connectivity(url.as_str()).await; - let rss = scanner_c.has_rss(&mut url, muted_host).await; - match scanner_c.nitter_version(&mut url, muted_host).await { - Some(version) => ( - connectivity, - rss, - Some(version.version_name), - Some(version.url), - ), - None => (connectivity, rss, None, None), - } - } - }; - - host::ActiveModel { - id: ActiveValue::NotSet, - domain: ActiveValue::Set(instance.domain), - country: ActiveValue::Set(instance.country), - url: ActiveValue::Set(instance.url), - enabled: ActiveValue::Set(true), - version: ActiveValue::Set(version), - version_url: ActiveValue::Set(version_url), - rss: ActiveValue::Set(rss), - updated: ActiveValue::Set(time.timestamp()), - connectivity: ActiveValue::Set(connectivity), - } - }); - } - while let Some(update_model) = join_set.join_next().await.map(|v| v.unwrap()) { - Host::insert(update_model) - .on_conflict( - OnConflict::column(host::Column::Domain) - .update_columns([ - host::Column::Enabled, - host::Column::Updated, - host::Column::Url, - host::Column::Rss, - host::Column::Version, - host::Column::VersionUrl, - host::Column::Country, - host::Column::Connectivity, - ]) - .to_owned(), - ) - .exec(&transaction) - .await?; - } - - transaction.commit().await?; - let end = Instant::now(); - let took_ms = end.saturating_duration_since(start).as_millis(); - { - *self.inner.last_list_fetch.lock().unwrap() = Utc::now(); - } - tracing::debug!( - removed = removed, - found = found_instances, - took_ms = took_ms - ); - Ok(()) - } - async fn fetch_instance_list(&self) -> Result { let (_, body) = self.fetch_url(&self.inner.config.instance_list_url).await?; Ok(body) } - /// Check ipv4/6 connectivity of host - async fn check_connectivity(&self, url: &str) -> Option { - let ipv4 = self - .inner - .client_ipv4 - .get(url) - .send() - .await - .map_or(false, |res| res.status().is_success()); - let ipv6 = self - .inner - .client_ipv6 - .get(url) - .send() - .await - .map_or(false, |res| res.status().is_success()); - - match (ipv4, ipv6) { - (true, true) => Some(host::Connectivity::All), - (false, true) => Some(host::Connectivity::IPv6), - (true, false) => Some(host::Connectivity::IPv4), - (false, false) => None, - } - } - async fn fetch_url(&self, url: &str) -> std::result::Result<(u16, String), FetchError> { let fetch_res = self.inner.client.get(url).send().await?; let code = fetch_res.status().as_u16(); @@ -481,29 +358,6 @@ impl Scanner { Ok((code, body)) } - - pub async fn query_latest_check( - &self, - connection: &T, - ) -> Result> { - let health_checks = LatestCheck::find_by_statement(Statement::from_sql_and_values( - DbBackend::Sqlite, - r#" - WITH latest AS( - SELECT u.host,MAX(u.time) as time FROM health_check u - GROUP BY u.host - ) - SELECT u.host,healthy,h.domain FROM health_check u - JOIN host h ON h.id = u.host - JOIN latest l ON l.host = u.host AND l.time = u.time - WHERE h.enabled = true - "#, - [], - )) - .all(connection) - .await?; - Ok(health_checks) - } } #[cfg(test)] @@ -514,10 +368,10 @@ mod test { use entities::health_check; use entities::state::scanner::Config; use migration::MigratorTrait; - use sea_orm::{ConnectOptions, Database}; + use sea_orm::{ConnectOptions, Database, ActiveValue, ActiveModelTrait}; use tokio::{fs::File, io::AsyncWriteExt}; - async fn db_init() -> DatabaseConnection { + pub(crate) async fn db_init() -> DatabaseConnection { let db = Database::connect(ConnectOptions::new( "sqlite:./test_db.db?mode=rwc".to_owned(), )) @@ -582,27 +436,6 @@ mod test { .unwrap(); dbg!(scanner.generate_cache_data().await.unwrap()); } - - #[tokio::test(flavor = "multi_thread", worker_threads = 1)] - #[ignore] - async fn connectivity_test() { - let db = db_init().await; - let scanner = Scanner::new(db, Config::test_defaults(), entities::state::new()) - .await - .unwrap(); - assert_eq!( - scanner.check_connectivity("https://v4.ipv6test.app").await, - Some(host::Connectivity::IPv4) - ); - assert_eq!( - scanner.check_connectivity("https://ipv6test.app").await, - Some(host::Connectivity::All) - ); - assert_eq!( - scanner.check_connectivity("https://v6.ipv6test.app").await, - Some(host::Connectivity::IPv6) - ); - } } impl std::fmt::Debug for Scanner { diff --git a/scanner/src/list_update.rs b/scanner/src/list_update.rs new file mode 100644 index 0000000..396958b --- /dev/null +++ b/scanner/src/list_update.rs @@ -0,0 +1,205 @@ +// SPDX-License-Identifier: AGPL-3.0-only +//! Updates the list of available instances, fetching all required fields + +use std::time::Duration; +use std::time::Instant; + +use chrono::Utc; +use entities::host; +use entities::prelude::Host; +use reqwest::Url; +use sea_orm::{QueryFilter, TransactionTrait, ActiveModelTrait, ActiveValue, ColumnTrait, EntityTrait}; +use sea_query::OnConflict; +use tokio::task::JoinSet; +use tracing::instrument; + +use crate::Scanner; +use crate::Result; + +impl Scanner { + /// Fetches the list of all instances from the wiki. + /// Updates all fields for host::Model, including connectivity, rss, version and enabled. + #[instrument] + pub(crate) async fn update_instacelist(&mut self) -> Result<()> { + let start = Instant::now(); + let html: String = self.fetch_instance_list().await?; + let parsed_instances = self.inner.instance_parser.parse_instancelist( + &html, + &self.inner.config.additional_hosts, + &self.inner.config.additional_host_country, + false, + )?; + + let transaction = self.inner.db.begin().await?; + + // find all currently enabled instances + let enabled_hosts = Host::find() + .filter(host::Column::Enabled.eq(true)) + .all(&transaction) + .await?; + // make a diff and remove the ones not found while parsing + let time: chrono::DateTime = Utc::now(); + let mut removed = 0; + for host in enabled_hosts.iter() { + if !parsed_instances.contains_key(&host.domain) { + host::ActiveModel { + id: ActiveValue::Set(host.id), + enabled: ActiveValue::Set(false), + updated: ActiveValue::Set(time.timestamp()), + ..Default::default() + } + .update(&transaction) + .await?; + removed += 1; + } + } + // now update/insert the existing ones + let found_instances: usize = parsed_instances.len(); + // find last update checks to detect spam + let last_status = self.query_latest_check(&transaction).await?; + let mut join_set = JoinSet::new(); + for (_, instance) in parsed_instances { + // TODO: parallelize this! + let scanner_c = self.clone(); + // detect already offline host and prevent log spam + let muted_host = match self.inner.config.auto_mute { + false => false, + true => last_status + .iter() + .find(|v| v.domain == instance.domain) + .map_or(false, |check| !check.healthy), + }; + // tracing::trace!(muted_host,instance=?instance,last_status=?last_status); + join_set.spawn(async move { + let (connectivity, rss, version, version_url) = match Url::parse(&instance.url) { + Err(_) => { + if !muted_host { + tracing::info!(url = instance.url, "Instance URL invalid"); + } + (None, false, None, None) + } + Ok(mut url) => { + let connectivity = scanner_c.check_connectivity(&mut url).await; + // prevent DoS + tokio::time::sleep(Duration::from_secs(1)).await; + let rss = scanner_c.has_rss(&mut url, muted_host).await; + tokio::time::sleep(Duration::from_secs(1)).await; + match scanner_c.nitter_version(&mut url, muted_host).await { + Some(version) => ( + connectivity, + rss, + Some(version.version_name), + Some(version.url), + ), + None => (connectivity, rss, None, None), + } + } + }; + + host::ActiveModel { + id: ActiveValue::NotSet, + domain: ActiveValue::Set(instance.domain), + country: ActiveValue::Set(instance.country), + url: ActiveValue::Set(instance.url), + enabled: ActiveValue::Set(true), + version: ActiveValue::Set(version), + version_url: ActiveValue::Set(version_url), + rss: ActiveValue::Set(rss), + updated: ActiveValue::Set(time.timestamp()), + connectivity: ActiveValue::Set(connectivity), + } + }); + } + while let Some(update_model) = join_set.join_next().await.map(|v| v.unwrap()) { + Host::insert(update_model) + .on_conflict( + OnConflict::column(host::Column::Domain) + .update_columns([ + host::Column::Enabled, + host::Column::Updated, + host::Column::Url, + host::Column::Rss, + host::Column::Version, + host::Column::VersionUrl, + host::Column::Country, + host::Column::Connectivity, + ]) + .to_owned(), + ) + .exec(&transaction) + .await?; + } + + transaction.commit().await?; + let end = Instant::now(); + let took_ms = end.saturating_duration_since(start).as_millis(); + { + *self.inner.last_list_fetch.lock().unwrap() = Utc::now(); + } + tracing::debug!( + removed = removed, + found = found_instances, + took_ms = took_ms + ); + Ok(()) + } + + /// Check ipv4/6 connectivity of host + async fn check_connectivity(&self, url: &mut Url) -> Option { + url.set_path(&self.inner.config.connectivity_path); + let ipv4 = self + .inner + .client_ipv4 + .get(url.as_str()) + .send() + .await + .map_or(false, |res| res.status().is_success()); + // prevent DoS + tokio::time::sleep(Duration::from_secs(1)).await; + let ipv6 = self + .inner + .client_ipv6 + .get(url.as_str()) + .send() + .await + .map_or(false, |res| res.status().is_success()); + + match (ipv4, ipv6) { + (true, true) => Some(host::Connectivity::All), + (false, true) => Some(host::Connectivity::IPv6), + (true, false) => Some(host::Connectivity::IPv4), + (false, false) => None, + } + } +} + +#[cfg(test)] +mod test { + use super::*; + use entities::state::scanner::Config; + use tracing_test::traced_test; + + use crate::{Scanner, test::db_init}; + + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + #[traced_test] + #[ignore] + async fn connectivity_test() { + let db = db_init().await; + let scanner = Scanner::new(db, Config::test_defaults(), entities::state::new()) + .await + .unwrap(); + assert_eq!( + scanner.check_connectivity(&mut Url::parse("https://v4.ipv6test.app").unwrap()).await, + Some(host::Connectivity::IPv4) + ); + assert_eq!( + scanner.check_connectivity(&mut Url::parse("https://ipv6test.app").unwrap()).await, + Some(host::Connectivity::All) + ); + assert_eq!( + scanner.check_connectivity(&mut Url::parse("https://v6.ipv6test.app").unwrap()).await, + Some(host::Connectivity::IPv6) + ); + } +} \ No newline at end of file diff --git a/server/src/website.rs b/server/src/website.rs index 0503222..05f556c 100644 --- a/server/src/website.rs +++ b/server/src/website.rs @@ -49,6 +49,7 @@ pub async fn about( paths.push(&scanner_config.about_path); paths.push(&scanner_config.rss_path); paths.push(&scanner_config.profile_path); + paths.push(&scanner_config.connectivity_path); context.insert("checked_paths", &paths); context.insert( "uptime_interval_s", diff --git a/src/main.rs b/src/main.rs index c5df25c..7d83ea8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -91,8 +91,7 @@ async fn _main() -> miette::Result<()> { disable_health_checks, ) .await - .into_diagnostic() - .wrap_err("Failed starting background scanner")?; + .wrap_err("Crash starting background scanner")?; let addr = std::net::SocketAddr::from(([127, 0, 0, 1], port)); server::start(&addr, pool, server_config, scanner_config, cache) @@ -157,6 +156,7 @@ fn read_scanner_cfg() -> miette::Result { bad_hosts, cleanup_interval: Duration::from_secs(cleanup_interval), error_retention_per_host, + connectivity_path: String::from("/"), })) }