This commit is contained in:
4cuck committed 2023-10-22 01:29:27 +02:00
1 parent fe0e8deca4
commit f336fe5be9
8 files changed
+120 -77

No files matched your search

+1 -1
View File
@@ -65,7 +65,7 @@ impl Config {
source_git_branch: String::from("master"),
source_git_url: String::from("https://github.com/zedeus/nitter.git"),
bad_hosts: vec![String::from("tweet.whateveritworks.org")],
cleanup_interval: Duration::from_secs(24*60*60),
cleanup_interval: Duration::from_secs(24 * 60 * 60),
error_retention_per_host: 100,
connectivity_path: String::from("/"),
})
+12 -9
View File
@@ -2,8 +2,8 @@
use std::cmp;
use std::collections::HashMap;
use chrono::{TimeZone, Duration};
use chrono::{Days, Utc};
use chrono::{Duration, TimeZone};
use entities::host;
use entities::prelude::*;
use entities::state::CacheData;
@@ -139,7 +139,8 @@ impl Scanner {
let host_ping_data = ping_data.remove(&host.id);
let last_healthy = last_healthy_check.remove(&host.id);
let __show_last_seen = last_healthy.map_or(true, |e|(time_now - e) > Duration::hours(12));
let __show_last_seen =
last_healthy.map_or(true, |e| (time_now - e) > Duration::hours(12));
host_statistics.push(CacheHost {
last_healthy: last_healthy,
__show_last_seen,
@@ -163,10 +164,12 @@ impl Scanner {
recent_checks: self.query_latest_health_checks(22, host.id).await?,
})
}
host_statistics.sort_unstable_by(|a,b| {
host_statistics.sort_unstable_by(|a, b| {
if a.points > 0 {
match a.points.cmp(&b.points) {
cmp::Ordering::Equal => a.healthy_percentage_overall.cmp(&b.healthy_percentage_overall),
cmp::Ordering::Equal => a
.healthy_percentage_overall
.cmp(&b.healthy_percentage_overall),
v => v,
}
} else {
@@ -174,11 +177,11 @@ impl Scanner {
if cmp_v.is_ne() {
cmp_v
} else {
match (a.last_healthy,b.last_healthy) {
(Some(a),Some(b)) => a.cmp(&b),
(Some(_),None) => cmp::Ordering::Greater,
(None,Some(_)) => cmp::Ordering::Less,
(None,None) => cmp::Ordering::Equal,
match (a.last_healthy, b.last_healthy) {
(Some(a), Some(b)) => a.cmp(&b),
(Some(_), None) => cmp::Ordering::Greater,
(None, Some(_)) => cmp::Ordering::Less,
(None, None) => cmp::Ordering::Equal,
}
}
}
+17 -14
View File
@@ -7,8 +7,8 @@ use sea_orm::QueryFilter;
use sea_query::Query;
use tokio::time::interval;
use crate::Scanner;
use crate::Result;
use crate::Scanner;
impl Scanner {
/// Setup scheduled job for cleaning up old data
@@ -37,20 +37,23 @@ impl Scanner {
let hosts = host::Entity::find().all(&self.inner.db).await?;
for host in hosts {
let res = check_errors::Entity::delete_many()
.filter(check_errors::Column::Host.eq(host.id))
.filter(check_errors::Column::Time.not_in_subquery(
Query::select()
.column(check_errors::Column::Time)
.from(check_errors::Entity)
.and_where(check_errors::Column::Host.eq(host.id))
.order_by(check_errors::Column::Time, Order::Desc)
.limit(self.inner.config.error_retention_per_host as _)
.to_owned()
))
.exec(&self.inner.db).await?;
tracing::debug!(host=host.id,deleted_errors=res.rows_affected);
.filter(check_errors::Column::Host.eq(host.id))
.filter(
check_errors::Column::Time.not_in_subquery(
Query::select()
.column(check_errors::Column::Time)
.from(check_errors::Entity)
.and_where(check_errors::Column::Host.eq(host.id))
.order_by(check_errors::Column::Time, Order::Desc)
.limit(self.inner.config.error_retention_per_host as _)
.to_owned(),
),
)
.exec(&self.inner.db)
.await?;
tracing::debug!(host = host.id, deleted_errors = res.rows_affected);
}
Ok(())
}
}
}
+28 -19
View File
@@ -5,20 +5,22 @@ use std::{
};
use about_parser::AboutParser;
use chrono::{Utc, DateTime, TimeZone, Duration};
use entities::{state::{error_cache::HostError, scanner::ScannerConfig, AppState}, health_check};
use chrono::{DateTime, Duration, TimeZone, Utc};
use entities::{
health_check,
state::{error_cache::HostError, scanner::ScannerConfig, AppState},
};
use instance_parser::InstanceParser;
use miette::{IntoDiagnostic, Context};
use miette::{Context, IntoDiagnostic};
use profile_parser::ProfileParser;
use regex::{Regex, RegexBuilder};
use reqwest::{
header::{HeaderMap, HeaderValue},
Client, ClientBuilder
Client, ClientBuilder,
};
use sea_orm::{
ColumnTrait, ConnectionTrait,
DatabaseConnection, DbBackend, EntityTrait, FromQueryResult, Statement,
QuerySelect,
ColumnTrait, ConnectionTrait, DatabaseConnection, DbBackend, EntityTrait, FromQueryResult,
QuerySelect, Statement,
};
use thiserror::Error;
use tokio::time::sleep;
@@ -27,12 +29,12 @@ type Result<T> = std::result::Result<T, ScannerError>;
mod about_parser;
mod cache_update;
mod cleanup;
mod instance_check;
mod instance_parser;
mod list_update;
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;
@@ -119,13 +121,14 @@ pub async fn run_scanner(
app_state: AppState,
disable_health_checks: bool,
) -> miette::Result<()> {
let scanner = Scanner::new(db, config, app_state).await
let scanner = Scanner::new(db, config, app_state)
.await
.wrap_err("Initializing scanner!")?;
scanner.schedule_cleanup().unwrap();
if disable_health_checks {
tracing::error!("Health checks disabled!");
return Ok(())
return Ok(());
}
tokio::spawn(async move {
scanner.run().await.expect("Failed to run scanner daemon!");
@@ -191,7 +194,8 @@ impl Scanner {
.build()
.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);
@@ -214,8 +218,11 @@ impl Scanner {
.wrap_err("Invalid RSS Content regex!")?,
}),
};
scanner.update_cache().await.into_diagnostic()
.wrap_err("Initial cache update failed!")?;
scanner
.update_cache()
.await
.into_diagnostic()
.wrap_err("Initial cache update failed!")?;
Ok(scanner)
}
@@ -237,7 +244,7 @@ impl Scanner {
.map(|model| model.max)
.flatten()
.unwrap_or_default();
Ok(Utc.timestamp_opt(time_max,0).unwrap())
Ok(Utc.timestamp_opt(time_max, 0).unwrap())
}
pub async fn query_latest_check<T: ConnectionTrait>(
@@ -287,7 +294,7 @@ impl Scanner {
self.last_uptime_check() + self.inner.config.instance_check_interval;
let delay_list_update = self.last_list_fetch() + self.inner.config.list_fetch_interval;
tracing::debug!(?delay_list_update,?delay_instance_check);
tracing::debug!(?delay_list_update, ?delay_instance_check);
let next_deadline = delay_instance_check.min(delay_list_update);
let now = Utc::now();
let sleep_time = next_deadline.signed_duration_since(now);
@@ -310,12 +317,14 @@ impl Scanner {
fn is_instance_check_outdated(&self) -> bool {
let val = self.last_uptime_check();
Utc::now().signed_duration_since(val).to_std().unwrap() >= self.inner.config.instance_check_interval
Utc::now().signed_duration_since(val).to_std().unwrap()
>= self.inner.config.instance_check_interval
}
fn is_instance_list_outdated(&self) -> bool {
let val = self.last_list_fetch();
Utc::now().signed_duration_since(val).to_std().unwrap() >= self.inner.config.list_fetch_interval
Utc::now().signed_duration_since(val).to_std().unwrap()
>= self.inner.config.list_fetch_interval
}
async fn fetch_instance_list(&self) -> Result<String> {
@@ -379,7 +388,7 @@ mod test {
use entities::health_check;
use entities::state::scanner::Config;
use migration::MigratorTrait;
use sea_orm::{ConnectOptions, Database, ActiveValue, ActiveModelTrait};
use sea_orm::{ActiveModelTrait, ActiveValue, ConnectOptions, Database};
use tokio::{fs::File, io::AsyncWriteExt};
pub(crate) async fn db_init() -> DatabaseConnection {
+15 -7
View File
@@ -8,13 +8,15 @@ use chrono::Utc;
use entities::host;
use entities::prelude::Host;
use reqwest::Url;
use sea_orm::{QueryFilter, TransactionTrait, ActiveModelTrait, ActiveValue, ColumnTrait, EntityTrait};
use sea_orm::{
ActiveModelTrait, ActiveValue, ColumnTrait, EntityTrait, QueryFilter, TransactionTrait,
};
use sea_query::OnConflict;
use tokio::task::JoinSet;
use tracing::instrument;
use crate::Scanner;
use crate::Result;
use crate::Scanner;
impl Scanner {
/// Fetches the list of all instances from the wiki.
@@ -179,7 +181,7 @@ mod test {
use entities::state::scanner::Config;
use tracing_test::traced_test;
use crate::{Scanner, test::db_init};
use crate::{test::db_init, Scanner};
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[traced_test]
@@ -190,16 +192,22 @@ mod test {
.await
.unwrap();
assert_eq!(
scanner.check_connectivity(&mut Url::parse("https://v4.ipv6test.app").unwrap()).await,
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,
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,
scanner
.check_connectivity(&mut Url::parse("https://v6.ipv6test.app").unwrap())
.await,
Some(host::Connectivity::IPv6)
);
}
}
}
+18 -11
View File
@@ -116,7 +116,10 @@ pub async fn login(
Err(_) => {
let mut ids = HashSet::with_capacity(1);
ids.insert(host.id);
ActiveLogin { hosts: ids, admin: config.admin_domains.iter().any(|e| e == domain) }
ActiveLogin {
hosts: ids,
admin: config.admin_domains.iter().any(|e| e == domain),
}
}
};
session.insert(LOGIN_KEY, session_value)?;
@@ -276,7 +279,7 @@ pub async fn overview(
) -> Result<axum::response::Response> {
tracing::info!(?session);
let (login,hosts) = get_all_login_hosts(&session, db).await?;
let (login, hosts) = get_all_login_hosts(&session, db).await?;
let mut context = tera::Context::new();
let res = {
@@ -385,16 +388,20 @@ async fn get_all_login_hosts(
let login = get_session_login(&session)?;
let host_res = match login.admin {
true => host::Entity::find()
.filter(host::Column::Enabled.eq(true))
.all(db)
.await?,
false => host::Entity::find()
.filter(host::Column::Id.is_in(login.hosts.iter().map(|v| *v)))
.all(db)
.await?,
true => {
host::Entity::find()
.filter(host::Column::Enabled.eq(true))
.all(db)
.await?
}
false => {
host::Entity::find()
.filter(host::Column::Id.is_in(login.hosts.iter().map(|v| *v)))
.all(db)
.await?
}
};
Ok((login,host_res))
Ok((login, host_res))
}
/// Get wanted [host::Model] for current [Session] if valid for this user
+25 -15
View File
@@ -1,5 +1,5 @@
// SPDX-License-Identifier: AGPL-3.0-only
use std::{borrow::Cow, net::SocketAddr, sync::Arc, collections::HashMap};
use std::{borrow::Cow, collections::HashMap, net::SocketAddr, sync::Arc};
use axum::{
error_handling::HandleErrorLayer,
@@ -7,17 +7,17 @@ use axum::{
http::HeaderValue,
response::{Html, Redirect},
routing::{get, get_service},
Router, BoxError,
BoxError, Router,
};
use chrono::TimeZone;
use entities::state::{scanner::ScannerConfig, AppState};
use hyper::{header, StatusCode};
use reqwest::Client;
use sea_orm::DatabaseConnection;
use tera::{Tera, from_value, to_value};
use tera::{from_value, to_value, Tera};
use thiserror::Error;
use tower::ServiceBuilder;
use tower_governor::{GovernorLayer, governor::GovernorConfigBuilder};
use tower_governor::{governor::GovernorConfigBuilder, GovernorLayer};
use tower_http::{
cors::CorsLayer,
limit::RequestBodyLimitLayer,
@@ -70,9 +70,14 @@ pub async fn start(
tracing::warn!("debug build, sessions are not secure!");
}
let pool = tower_sessions::sqlx::SqlitePool::connect(&config.session_db_uri).await.expect("failed to initialize session store");
let pool = tower_sessions::sqlx::SqlitePool::connect(&config.session_db_uri)
.await
.expect("failed to initialize session store");
let session_store = SqliteStore::new(pool);
session_store.migrate().await.expect("failed to migrate session store");
session_store
.migrate()
.await
.expect("failed to migrate session store");
tokio::task::spawn(
session_store
.clone()
@@ -127,12 +132,12 @@ pub async fn start(
.unwrap(),
);
let rate_limit_layer = ServiceBuilder::new()
.layer(HandleErrorLayer::new(|e: BoxError| async move {
tower_governor::errors::display_error(e) // too many requests
}))
.layer(GovernorLayer {
config: Box::leak(per_ip_governor_conf),
});
.layer(HandleErrorLayer::new(|e: BoxError| async move {
tower_governor::errors::display_error(e) // too many requests
}))
.layer(GovernorLayer {
config: Box::leak(per_ip_governor_conf),
});
let router = Router::new()
.nest_service(
@@ -239,11 +244,16 @@ fn fmt_date(args: &HashMap<String, tera::Value>) -> tera::Result<tera::Value> {
Ok(time_i64) => {
let format = match args.get("fmt") {
None => "%Y.%m.%d %H:%M:%S",
Some(v) => v.as_str().ok_or_else(||tera::Error::from("fmt has to be a string"))?,
Some(v) => v
.as_str()
.ok_or_else(|| tera::Error::from("fmt has to be a string"))?,
};
let time = chrono::Utc.timestamp_opt(time_i64,0).single().ok_or_else(||tera::Error::from("Invalid timestamp"))?;
let time = chrono::Utc
.timestamp_opt(time_i64, 0)
.single()
.ok_or_else(|| tera::Error::from("Invalid timestamp"))?;
Ok(to_value(time.format(format).to_string()).unwrap())
},
}
Err(_) => Err("timestamp not an i64".into()),
},
None => Err("no value provided".into()),
+4 -1
View File
@@ -181,7 +181,10 @@ fn read_server_config(instance_ping_interval: usize) -> miette::Result<server::C
.parse()
.expect("SESSION_TTL_SECONDS must be a positive number");
let login_token_name = require_env_str("LOGIN_TOKEN_NAME")?;
let admin_domains = require_env_str("ADMIN_DOMAINS")?.split(",").map(|v|v.trim().to_string()).collect();
let admin_domains = require_env_str("ADMIN_DOMAINS")?
.split(",")
.map(|v| v.trim().to_string())
.collect();
let session_db_uri = require_env_str("SESSION_DB_URI")?;
Ok(server::Config {